From 777c81f865202843687524ac4f75be252540e02b Mon Sep 17 00:00:00 2001 From: Alan Wu Date: Mon, 13 Jul 2026 16:09:20 -0400 Subject: [PATCH] Add attempt count to task attributes, Validate task on standby (#10985) ## What changed? Plumb executable attempt count to task attributes during task validation/execution. Check task validator in side effect standby task executor. ## Why? Add a field to allow task validators to decide if task is best effort, meaning they attempt execution a number of times before being invalidated. ## How did you test it? - [X] built - [X] run locally and tested manually - [X] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) --- chasm/chasmtest/task_helpers.go | 4 +- chasm/lib/activity/activity_tasks.go | 10 ++--- chasm/lib/callback/tasks.go | 4 +- .../lib/nexusoperation/cancellation_tasks.go | 4 +- .../nexusoperation/cancellation_tasks_test.go | 4 +- chasm/lib/nexusoperation/operation_tasks.go | 10 ++--- .../nexusoperation/operation_tasks_test.go | 10 ++--- chasm/lib/scheduler/backfiller_tasks.go | 2 +- chasm/lib/scheduler/generator_tasks.go | 2 +- .../scheduler/invoker_execute_task_test.go | 4 +- .../invoker_process_buffer_task_test.go | 2 +- chasm/lib/scheduler/invoker_tasks.go | 4 +- .../lib/scheduler/metrics_consistency_test.go | 8 ++-- .../scheduler/scheduler_idle_tasks_test.go | 6 +-- chasm/lib/scheduler/scheduler_migrate_task.go | 2 +- chasm/lib/scheduler/scheduler_tasks.go | 4 +- chasm/lib/scheduler/sentinel_test.go | 6 +-- chasm/lib/tests/tasks.go | 8 ++-- chasm/registrable_task.go | 10 ++--- chasm/task.go | 17 +++++++- chasm/task_mock.go | 6 +-- chasm/tree.go | 30 ++++++++------ chasm/tree_test.go | 40 ++++++++++++++----- chasm/visibility.go | 2 +- chasm/visibility_task_test.go | 4 +- service/history/chasm_task_util_test.go | 4 +- .../outbound_queue_active_task_executor.go | 1 + ...utbound_queue_active_task_executor_test.go | 2 + .../outbound_queue_standby_task_executor.go | 8 ++-- ...tbound_queue_standby_task_executor_test.go | 9 +++-- service/history/tasks/chasm_task.go | 4 ++ .../timer_queue_active_task_executor.go | 1 + .../timer_queue_standby_task_executor.go | 8 ++-- .../timer_queue_standby_task_executor_test.go | 4 +- .../transfer_queue_active_task_executor.go | 1 + .../transfer_queue_standby_task_executor.go | 8 ++-- ...ansfer_queue_standby_task_executor_test.go | 4 +- .../history/visibility_queue_task_executor.go | 1 + 38 files changed, 158 insertions(+), 100 deletions(-) diff --git a/chasm/chasmtest/task_helpers.go b/chasm/chasmtest/task_helpers.go index a8e6d3c6ec..f5dadae2ef 100644 --- a/chasm/chasmtest/task_helpers.go +++ b/chasm/chasmtest/task_helpers.go @@ -40,7 +40,7 @@ func ExecutePureTask[C chasm.Component, T any]( return fmt.Errorf("component type mismatch: got %T", c) } var valid bool - valid, err = handler.Validate(mutableCtx, typedC, attrs, task) + valid, err = handler.Validate(mutableCtx, typedC, chasm.TaskInvocation{TaskAttributes: attrs}, task) if err != nil { return err } @@ -92,7 +92,7 @@ func ExecuteSideEffectTask[C chasm.Component, T any]( if !ok { return fmt.Errorf("component type mismatch: got %T", c) } - valid, err = handler.Validate(chasmCtx, typedC, attrs, task) + valid, err = handler.Validate(chasmCtx, typedC, chasm.TaskInvocation{TaskAttributes: attrs}, task) return err }, ); err != nil { diff --git a/chasm/lib/activity/activity_tasks.go b/chasm/lib/activity/activity_tasks.go index e22b2f586a..897698c62f 100644 --- a/chasm/lib/activity/activity_tasks.go +++ b/chasm/lib/activity/activity_tasks.go @@ -32,7 +32,7 @@ func newActivityDispatchTaskHandler(opts activityDispatchTaskHandlerOptions) *ac func (h *activityDispatchTaskHandler) Validate( ctx chasm.Context, activity *Activity, - _ chasm.TaskAttributes, + _ chasm.TaskInvocation, task *activitypb.ActivityDispatchTask, ) (bool, error) { // TODO(saa-preview): make sure we handle resets when we support them, as they will reset the attempt count @@ -90,7 +90,7 @@ func newScheduleToStartTimeoutTaskHandler() *scheduleToStartTimeoutTaskHandler { func (h *scheduleToStartTimeoutTaskHandler) Validate( ctx chasm.Context, activity *Activity, - _ chasm.TaskAttributes, + _ chasm.TaskInvocation, task *activitypb.ScheduleToStartTimeoutTask, ) (bool, error) { return (activity.Status == activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED && @@ -126,7 +126,7 @@ func newScheduleToCloseTimeoutTaskHandler() *scheduleToCloseTimeoutTaskHandler { func (h *scheduleToCloseTimeoutTaskHandler) Validate( _ chasm.Context, activity *Activity, - _ chasm.TaskAttributes, + _ chasm.TaskInvocation, _ *activitypb.ScheduleToCloseTimeoutTask, ) (bool, error) { return TransitionTimedOut.Possible(activity), nil @@ -160,7 +160,7 @@ func newStartToCloseTimeoutTaskHandler() *startToCloseTimeoutTaskHandler { func (h *startToCloseTimeoutTaskHandler) Validate( ctx chasm.Context, activity *Activity, - _ chasm.TaskAttributes, + _ chasm.TaskInvocation, task *activitypb.StartToCloseTimeoutTask, ) (bool, error) { valid := ((activity.Status == activitypb.ACTIVITY_EXECUTION_STATUS_STARTED || @@ -211,7 +211,7 @@ func newHeartbeatTimeoutTaskHandler() *heartbeatTimeoutTaskHandler { func (h *heartbeatTimeoutTaskHandler) Validate( ctx chasm.Context, activity *Activity, - taskAttrs chasm.TaskAttributes, + taskAttrs chasm.TaskInvocation, task *activitypb.HeartbeatTimeoutTask, ) (bool, error) { // Let T = user-configured heartbeat timeout and let hb_i be the time of the ith user-submitted diff --git a/chasm/lib/callback/tasks.go b/chasm/lib/callback/tasks.go index b53efafea1..c54e065e07 100644 --- a/chasm/lib/callback/tasks.go +++ b/chasm/lib/callback/tasks.go @@ -104,7 +104,7 @@ func newInvocationTaskHandler(opts invocationTaskHandlerOptions) *invocationTask } } -func (h *invocationTaskHandler) Validate(ctx chasm.Context, cb *Callback, attrs chasm.TaskAttributes, task *callbackspb.InvocationTask) (bool, error) { +func (h *invocationTaskHandler) Validate(ctx chasm.Context, cb *Callback, attrs chasm.TaskInvocation, task *callbackspb.InvocationTask) (bool, error) { return cb.Attempt == task.Attempt && cb.Status == callbackspb.CALLBACK_STATUS_SCHEDULED, nil } @@ -175,7 +175,7 @@ func (h *backoffTaskHandler) Execute( func (h *backoffTaskHandler) Validate( ctx chasm.Context, callback *Callback, - taskAttr chasm.TaskAttributes, + taskAttr chasm.TaskInvocation, task *callbackspb.BackoffTask, ) (bool, error) { return callback.Status == callbackspb.CALLBACK_STATUS_BACKING_OFF && callback.Attempt == task.Attempt, nil diff --git a/chasm/lib/nexusoperation/cancellation_tasks.go b/chasm/lib/nexusoperation/cancellation_tasks.go index 893b3c1793..3b90bcc6dc 100644 --- a/chasm/lib/nexusoperation/cancellation_tasks.go +++ b/chasm/lib/nexusoperation/cancellation_tasks.go @@ -88,7 +88,7 @@ func newCancellationInvocationTaskHandler(opts cancellationInvocationTaskHandler func (h *cancellationInvocationTaskHandler) Validate( _ chasm.Context, cancellation *Cancellation, - _ chasm.TaskAttributes, + _ chasm.TaskInvocation, task *nexusoperationpb.CancellationTask, ) (bool, error) { return cancellation.Status == nexusoperationpb.CANCELLATION_STATUS_SCHEDULED && @@ -213,7 +213,7 @@ func newCancellationBackoffTaskHandler(opts commonTaskHandlerOptions) *cancellat func (h *cancellationBackoffTaskHandler) Validate( _ chasm.Context, cancellation *Cancellation, - _ chasm.TaskAttributes, + _ chasm.TaskInvocation, task *nexusoperationpb.CancellationBackoffTask, ) (bool, error) { isValid := cancellation.Status == nexusoperationpb.CANCELLATION_STATUS_BACKING_OFF && cancellation.GetAttempt() == task.GetAttempt() diff --git a/chasm/lib/nexusoperation/cancellation_tasks_test.go b/chasm/lib/nexusoperation/cancellation_tasks_test.go index aac64ef321..694c4fec70 100644 --- a/chasm/lib/nexusoperation/cancellation_tasks_test.go +++ b/chasm/lib/nexusoperation/cancellation_tasks_test.go @@ -186,7 +186,7 @@ func TestCancellationInvocationTaskHandler_Validate(t *testing.T) { Attempt: tc.cancelAttempt, }) - valid, err := handler.Validate(ctx, c, chasm.TaskAttributes{}, &nexusoperationpb.CancellationTask{Attempt: tc.taskAttempt}) + valid, err := handler.Validate(ctx, c, chasm.TaskInvocation{}, &nexusoperationpb.CancellationTask{Attempt: tc.taskAttempt}) require.NoError(t, err) require.Equal(t, tc.valid, valid) }) @@ -234,7 +234,7 @@ func TestCancellationBackoffTaskHandler_Validate(t *testing.T) { Attempt: tc.attempt, }) - valid, err := handler.Validate(ctx, c, chasm.TaskAttributes{}, tc.task) + valid, err := handler.Validate(ctx, c, chasm.TaskInvocation{}, tc.task) require.NoError(t, err) require.Equal(t, tc.valid, valid) }) diff --git a/chasm/lib/nexusoperation/operation_tasks.go b/chasm/lib/nexusoperation/operation_tasks.go index 8c93d17436..7e42fc3f69 100644 --- a/chasm/lib/nexusoperation/operation_tasks.go +++ b/chasm/lib/nexusoperation/operation_tasks.go @@ -86,7 +86,7 @@ func newOperationInvocationTaskHandler(opts operationInvocationTaskHandlerOption func (h *operationInvocationTaskHandler) Validate( _ chasm.Context, op *Operation, - _ chasm.TaskAttributes, + _ chasm.TaskInvocation, task *nexusoperationpb.InvocationTask, ) (bool, error) { isValid := op.Status == nexusoperationpb.OPERATION_STATUS_SCHEDULED && op.GetAttempt() == task.GetAttempt() @@ -353,7 +353,7 @@ func newOperationBackoffTaskHandler(opts operationTaskHandlerOptions) *operation func (h *operationBackoffTaskHandler) Validate( ctx chasm.Context, op *Operation, - attrs chasm.TaskAttributes, + attrs chasm.TaskInvocation, task *nexusoperationpb.InvocationBackoffTask, ) (bool, error) { return op.Status == nexusoperationpb.OPERATION_STATUS_BACKING_OFF && op.GetAttempt() == task.GetAttempt(), nil @@ -387,7 +387,7 @@ func newOperationScheduleToStartTimeoutTaskHandler(opts operationTaskHandlerOpti func (h *operationScheduleToStartTimeoutTaskHandler) Validate( ctx chasm.Context, op *Operation, - attrs chasm.TaskAttributes, + attrs chasm.TaskInvocation, task *nexusoperationpb.ScheduleToStartTimeoutTask, ) (bool, error) { return TransitionStarted.Possible(op), nil @@ -428,7 +428,7 @@ func newOperationStartToCloseTimeoutTaskHandler(opts operationTaskHandlerOptions func (h *operationStartToCloseTimeoutTaskHandler) Validate( ctx chasm.Context, op *Operation, - attrs chasm.TaskAttributes, + attrs chasm.TaskInvocation, task *nexusoperationpb.StartToCloseTimeoutTask, ) (bool, error) { return op.Status == nexusoperationpb.OPERATION_STATUS_STARTED, nil @@ -469,7 +469,7 @@ func newOperationScheduleToCloseTimeoutTaskHandler(opts operationTaskHandlerOpti func (h *operationScheduleToCloseTimeoutTaskHandler) Validate( ctx chasm.Context, op *Operation, - attrs chasm.TaskAttributes, + attrs chasm.TaskInvocation, task *nexusoperationpb.ScheduleToCloseTimeoutTask, ) (bool, error) { return TransitionTimedOut.Possible(op), nil diff --git a/chasm/lib/nexusoperation/operation_tasks_test.go b/chasm/lib/nexusoperation/operation_tasks_test.go index 13168eb706..95105d9909 100644 --- a/chasm/lib/nexusoperation/operation_tasks_test.go +++ b/chasm/lib/nexusoperation/operation_tasks_test.go @@ -637,7 +637,7 @@ func TestInvocationTaskHandler_Validate(t *testing.T) { op.Status = tc.status op.Attempt = tc.opAttempt - valid, err := handler.Validate(ctx, op, chasm.TaskAttributes{}, &nexusoperationpb.InvocationTask{Attempt: tc.taskAttempt}) + valid, err := handler.Validate(ctx, op, chasm.TaskInvocation{}, &nexusoperationpb.InvocationTask{Attempt: tc.taskAttempt}) require.NoError(t, err) require.Equal(t, tc.valid, valid) }) @@ -684,7 +684,7 @@ func TestBackoffTaskHandler_Validate(t *testing.T) { op.Status = tc.status op.Attempt = tc.attempt - valid, err := handler.Validate(ctx, op, chasm.TaskAttributes{}, tc.task) + valid, err := handler.Validate(ctx, op, chasm.TaskInvocation{}, tc.task) require.NoError(t, err) require.Equal(t, tc.valid, valid) }) @@ -754,7 +754,7 @@ func TestScheduleToStartTimeoutTaskHandler_Validate(t *testing.T) { op := newTestOperation() op.Status = tc.status - valid, err := handler.Validate(ctx, op, chasm.TaskAttributes{}, &nexusoperationpb.ScheduleToStartTimeoutTask{}) + valid, err := handler.Validate(ctx, op, chasm.TaskInvocation{}, &nexusoperationpb.ScheduleToStartTimeoutTask{}) require.NoError(t, err) require.Equal(t, tc.valid, valid) }) @@ -829,7 +829,7 @@ func TestStartToCloseTimeoutTaskHandler_Validate(t *testing.T) { op := newTestOperation() op.Status = tc.status - valid, err := handler.Validate(ctx, op, chasm.TaskAttributes{}, &nexusoperationpb.StartToCloseTimeoutTask{}) + valid, err := handler.Validate(ctx, op, chasm.TaskInvocation{}, &nexusoperationpb.StartToCloseTimeoutTask{}) require.NoError(t, err) require.Equal(t, tc.valid, valid) }) @@ -899,7 +899,7 @@ func TestScheduleToCloseTimeoutTaskHandler_Validate(t *testing.T) { op := newTestOperation() op.Status = tc.status - valid, err := handler.Validate(ctx, op, chasm.TaskAttributes{}, &nexusoperationpb.ScheduleToCloseTimeoutTask{}) + valid, err := handler.Validate(ctx, op, chasm.TaskInvocation{}, &nexusoperationpb.ScheduleToCloseTimeoutTask{}) require.NoError(t, err) require.Equal(t, tc.valid, valid) }) diff --git a/chasm/lib/scheduler/backfiller_tasks.go b/chasm/lib/scheduler/backfiller_tasks.go index df0e1a9cf5..9b4034a044 100644 --- a/chasm/lib/scheduler/backfiller_tasks.go +++ b/chasm/lib/scheduler/backfiller_tasks.go @@ -52,7 +52,7 @@ const ( func (b *BackfillerTaskHandler) Validate( ctx chasm.Context, backfiller *Backfiller, - attrs chasm.TaskAttributes, + attrs chasm.TaskInvocation, _ *schedulerpb.BackfillerTask, ) (bool, error) { valid, err := validateTaskHighWaterMark(backfiller.GetLastProcessedTime(), attrs.ScheduledTime) diff --git a/chasm/lib/scheduler/generator_tasks.go b/chasm/lib/scheduler/generator_tasks.go index e2b19204ac..f5376be24b 100644 --- a/chasm/lib/scheduler/generator_tasks.go +++ b/chasm/lib/scheduler/generator_tasks.go @@ -191,7 +191,7 @@ func (g *GeneratorTaskHandler) logSchedule(ctx chasm.MutableContext, logger log. func (g *GeneratorTaskHandler) Validate( ctx chasm.Context, generator *Generator, - attrs chasm.TaskAttributes, + attrs chasm.TaskInvocation, _ *schedulerpb.GeneratorTask, ) (bool, error) { return validateTaskHighWaterMark( diff --git a/chasm/lib/scheduler/invoker_execute_task_test.go b/chasm/lib/scheduler/invoker_execute_task_test.go index c413541fb8..fe61fe2162 100644 --- a/chasm/lib/scheduler/invoker_execute_task_test.go +++ b/chasm/lib/scheduler/invoker_execute_task_test.go @@ -523,7 +523,7 @@ func TestExecuteTask_Validate_BackoffEqualToLPTIsEligible(t *testing.T) { BackoffTime: timestamppb.New(now), }} - valid, err := env.handler.Validate(ctx, invoker, chasm.TaskAttributes{}, &schedulerpb.InvokerExecuteTask{}) + valid, err := env.handler.Validate(ctx, invoker, chasm.TaskInvocation{}, &schedulerpb.InvokerExecuteTask{}) require.NoError(t, err) require.True(t, valid, "BackoffTime == LastProcessedTime must be eligible (<=, not strict <)") } @@ -605,7 +605,7 @@ func TestExecuteTask_Validate(t *testing.T) { invoker.TerminateWorkflows = nil invoker.LastProcessedTime = nil c.setup() - valid, err := env.handler.Validate(ctx, invoker, chasm.TaskAttributes{}, &schedulerpb.InvokerExecuteTask{}) + valid, err := env.handler.Validate(ctx, invoker, chasm.TaskInvocation{}, &schedulerpb.InvokerExecuteTask{}) require.NoError(t, err) require.Equal(t, c.expectedValid, valid) }) diff --git a/chasm/lib/scheduler/invoker_process_buffer_task_test.go b/chasm/lib/scheduler/invoker_process_buffer_task_test.go index 1b0d998da7..4a6ec91201 100644 --- a/chasm/lib/scheduler/invoker_process_buffer_task_test.go +++ b/chasm/lib/scheduler/invoker_process_buffer_task_test.go @@ -39,7 +39,7 @@ func TestProcessBufferTask_Validate(t *testing.T) { for _, c := range cases { t.Run(c.name, func(t *testing.T) { invoker.LastProcessedTime = c.lastProcessedTime - valid, err := handler.Validate(env.MutableContext(), invoker, chasm.TaskAttributes{ScheduledTime: c.scheduledTime}, &schedulerpb.InvokerProcessBufferTask{}) + valid, err := handler.Validate(env.MutableContext(), invoker, chasm.TaskInvocation{TaskAttributes: chasm.TaskAttributes{ScheduledTime: c.scheduledTime}}, &schedulerpb.InvokerProcessBufferTask{}) require.NoError(t, err) require.Equal(t, c.expectedValid, valid) }) diff --git a/chasm/lib/scheduler/invoker_tasks.go b/chasm/lib/scheduler/invoker_tasks.go index af50b27962..f54a3547f8 100644 --- a/chasm/lib/scheduler/invoker_tasks.go +++ b/chasm/lib/scheduler/invoker_tasks.go @@ -138,7 +138,7 @@ func (h *InvokerExecuteTaskHandler) recordDuplicateExecuteDrops(scheduler *Sched func (h *InvokerExecuteTaskHandler) Validate( ctx chasm.Context, invoker *Invoker, - _ chasm.TaskAttributes, + _ chasm.TaskInvocation, _ *schedulerpb.InvokerExecuteTask, ) (bool, error) { // If another execute task already happened to kick everything off, we don't need @@ -409,7 +409,7 @@ func (h *InvokerExecuteTaskHandler) startWorkflows( func (h *InvokerProcessBufferTaskHandler) Validate( ctx chasm.Context, invoker *Invoker, - attrs chasm.TaskAttributes, + attrs chasm.TaskInvocation, _ *schedulerpb.InvokerProcessBufferTask, ) (bool, error) { valid, err := validateTaskHighWaterMark(invoker.GetLastProcessedTime(), attrs.ScheduledTime) diff --git a/chasm/lib/scheduler/metrics_consistency_test.go b/chasm/lib/scheduler/metrics_consistency_test.go index c0e8f16c32..91dfbb8b58 100644 --- a/chasm/lib/scheduler/metrics_consistency_test.go +++ b/chasm/lib/scheduler/metrics_consistency_test.go @@ -78,7 +78,7 @@ func TestScheduleIdleTask_ConsistentLabels(t *testing.T) { // invalidated: Execute set Closed=true above, so Validate now rejects with // reason=closed. valid, err := h.Validate(env.MutableContext(), env.Scheduler, - chasm.TaskAttributes{ScheduledTime: env.TimeSource.Now()}, + chasm.TaskInvocation{TaskAttributes: chasm.TaskAttributes{ScheduledTime: env.TimeSource.Now()}}, &schedulerpb.SchedulerIdleTask{IdleTimeTotal: durationpb.New(10 * time.Minute)}) require.NoError(t, err) require.False(t, valid) @@ -109,7 +109,7 @@ func TestScheduleInvokerProcessBufferTask_ConsistentLabels(t *testing.T) { // invalidated: a task scheduled before the high water mark is stale. valid, err := h.Validate(env.ReadContext(), invoker, - chasm.TaskAttributes{ScheduledTime: env.TimeSource.Now().Add(-time.Minute)}, + chasm.TaskInvocation{TaskAttributes: chasm.TaskAttributes{ScheduledTime: env.TimeSource.Now().Add(-time.Minute)}}, &schedulerpb.InvokerProcessBufferTask{}) require.NoError(t, err) require.False(t, valid) @@ -156,7 +156,7 @@ func TestScheduleBackfillerTask_ConsistentLabels(t *testing.T) { // invalidated: a task scheduled before the high water mark is stale. backfiller.LastProcessedTime = timestamppb.New(env.TimeSource.Now()) valid, err := h.Validate(env.ReadContext(), backfiller, - chasm.TaskAttributes{ScheduledTime: env.TimeSource.Now().Add(-time.Hour)}, + chasm.TaskInvocation{TaskAttributes: chasm.TaskAttributes{ScheduledTime: env.TimeSource.Now().Add(-time.Hour)}}, &schedulerpb.BackfillerTask{}) require.NoError(t, err) require.False(t, valid) @@ -184,7 +184,7 @@ func TestScheduleInvokerExecuteTask_ConsistentLabels(t *testing.T) { invoker.LastProcessedTime = timestamppb.New(env.TimeSource.Now()) // invalidated: no terminate/cancel/eligible work records reason=no_work. - valid, err := h.Validate(ctx, invoker, chasm.TaskAttributes{}, &schedulerpb.InvokerExecuteTask{}) + valid, err := h.Validate(ctx, invoker, chasm.TaskInvocation{}, &schedulerpb.InvokerExecuteTask{}) require.NoError(t, err) require.False(t, valid) diff --git a/chasm/lib/scheduler/scheduler_idle_tasks_test.go b/chasm/lib/scheduler/scheduler_idle_tasks_test.go index 155669dc27..17f6dd0376 100644 --- a/chasm/lib/scheduler/scheduler_idle_tasks_test.go +++ b/chasm/lib/scheduler/scheduler_idle_tasks_test.go @@ -52,7 +52,7 @@ func runIdleValidateTestCase(t *testing.T, env *testEnv, c *idleValidateTestCase task := &schedulerpb.SchedulerIdleTask{IdleTimeTotal: durationpb.New(c.taskIdleTimeTotal)} taskAttrs := chasm.TaskAttributes{ScheduledTime: c.scheduledTime} - isValid, err := handler.Validate(ctx, sched, taskAttrs, task) + isValid, err := handler.Validate(ctx, sched, chasm.TaskInvocation{TaskAttributes: taskAttrs}, task) require.NoError(t, err) require.Equal(t, c.expectedValid, isValid) } @@ -193,7 +193,7 @@ func TestIdleTask_Validate_MetricReasons(t *testing.T) { taskAttrs := chasm.TaskAttributes{ScheduledTime: now} c.setup(env.Scheduler, now, &taskAttrs) - isValid, err := handler.Validate(env.MutableContext(), env.Scheduler, taskAttrs, + isValid, err := handler.Validate(env.MutableContext(), env.Scheduler, chasm.TaskInvocation{TaskAttributes: taskAttrs}, &schedulerpb.SchedulerIdleTask{IdleTimeTotal: durationpb.New(10 * time.Minute)}) require.NoError(t, err) require.False(t, isValid) @@ -297,7 +297,7 @@ func TestIdleTask_Validate_SentinelNotHeldOpen(t *testing.T) { ScheduledTime: sentinel.Info.CreateTime.AsTime().Add(scheduler.SentinelIdleTime), } - isValid, err := handler.Validate(ctx, sentinel, taskAttrs, task) + isValid, err := handler.Validate(ctx, sentinel, chasm.TaskInvocation{TaskAttributes: taskAttrs}, task) require.NoError(t, err) require.True(t, isValid, "sentinel must remain eligible to close regardless of paused/backfill/empty-spec state") } diff --git a/chasm/lib/scheduler/scheduler_migrate_task.go b/chasm/lib/scheduler/scheduler_migrate_task.go index d901fc0634..f2b5237e8e 100644 --- a/chasm/lib/scheduler/scheduler_migrate_task.go +++ b/chasm/lib/scheduler/scheduler_migrate_task.go @@ -66,7 +66,7 @@ func NewSchedulerMigrateToWorkflowTaskHandler( func (h *SchedulerMigrateToWorkflowTaskHandler) Validate( _ chasm.Context, scheduler *Scheduler, - _ chasm.TaskAttributes, + _ chasm.TaskInvocation, _ *schedulerpb.SchedulerMigrateToWorkflowTask, ) (bool, error) { if scheduler.Closed { diff --git a/chasm/lib/scheduler/scheduler_tasks.go b/chasm/lib/scheduler/scheduler_tasks.go index 58d9b01e58..4d323880b9 100644 --- a/chasm/lib/scheduler/scheduler_tasks.go +++ b/chasm/lib/scheduler/scheduler_tasks.go @@ -72,7 +72,7 @@ const ( func (r *SchedulerIdleTaskHandler) Validate( ctx chasm.Context, scheduler *Scheduler, - taskAttrs chasm.TaskAttributes, + taskAttrs chasm.TaskInvocation, task *schedulerpb.SchedulerIdleTask, ) (bool, error) { if scheduler.Closed { @@ -329,7 +329,7 @@ func (r *SchedulerCallbacksTaskHandler) watchRunningStart( func (r *SchedulerCallbacksTaskHandler) Validate( ctx chasm.Context, scheduler *Scheduler, - taskAttrs chasm.TaskAttributes, + taskAttrs chasm.TaskInvocation, task *schedulerpb.SchedulerCallbacksTask, ) (bool, error) { invoker := scheduler.Invoker.Get(ctx) diff --git a/chasm/lib/scheduler/sentinel_test.go b/chasm/lib/scheduler/sentinel_test.go index c00fd034fe..a86c75e8cc 100644 --- a/chasm/lib/scheduler/sentinel_test.go +++ b/chasm/lib/scheduler/sentinel_test.go @@ -31,7 +31,7 @@ func TestSentinelIdleTask_Validate_Valid(t *testing.T) { ScheduledTime: sentinel.Info.CreateTime.AsTime().Add(scheduler.SentinelIdleTime), } - isValid, err := executor.Validate(ctx, sentinel, taskAttrs, task) + isValid, err := executor.Validate(ctx, sentinel, chasm.TaskInvocation{TaskAttributes: taskAttrs}, task) require.NoError(t, err) require.True(t, isValid) } @@ -46,7 +46,7 @@ func TestSentinelIdleTask_Validate_InvalidAfterClosed(t *testing.T) { ScheduledTime: sentinel.Info.CreateTime.AsTime().Add(scheduler.SentinelIdleTime), } - isValid, err := executor.Validate(ctx, sentinel, taskAttrs, task) + isValid, err := executor.Validate(ctx, sentinel, chasm.TaskInvocation{TaskAttributes: taskAttrs}, task) require.NoError(t, err) require.False(t, isValid) } @@ -62,7 +62,7 @@ func TestSentinelIdleTask_Validate_ExpirationShiftedLater(t *testing.T) { ScheduledTime: sentinel.Info.CreateTime.AsTime(), } - isValid, err := executor.Validate(ctx, sentinel, taskAttrs, task) + isValid, err := executor.Validate(ctx, sentinel, chasm.TaskInvocation{TaskAttributes: taskAttrs}, task) require.NoError(t, err) require.False(t, isValid) } diff --git a/chasm/lib/tests/tasks.go b/chasm/lib/tests/tasks.go index d501a11472..f9e952a35d 100644 --- a/chasm/lib/tests/tasks.go +++ b/chasm/lib/tests/tasks.go @@ -26,10 +26,10 @@ func (h *PayloadTTLPureTaskHandler) Execute( func (h *PayloadTTLPureTaskHandler) Validate( chasmContext chasm.Context, store *PayloadStore, - attributes chasm.TaskAttributes, + attributes chasm.TaskInvocation, task *testspb.TestPayloadTTLPureTask, ) (bool, error) { - return validateTask(chasmContext, store, attributes, task.PayloadKey) + return validateTask(chasmContext, store, attributes.TaskAttributes, task.PayloadKey) } type PayloadTTLSideEffectTaskHandler struct { @@ -54,10 +54,10 @@ func (h *PayloadTTLSideEffectTaskHandler) Execute( func (h *PayloadTTLSideEffectTaskHandler) Validate( chasmContext chasm.Context, store *PayloadStore, - attributes chasm.TaskAttributes, + attributes chasm.TaskInvocation, task *testspb.TestPayloadTTLSideEffectTask, ) (bool, error) { - return validateTask(chasmContext, store, attributes, task.PayloadKey) + return validateTask(chasmContext, store, attributes.TaskAttributes, task.PayloadKey) } func validateTask( diff --git a/chasm/registrable_task.go b/chasm/registrable_task.go index 65a0977539..5af74cfc40 100644 --- a/chasm/registrable_task.go +++ b/chasm/registrable_task.go @@ -37,7 +37,7 @@ type ( RegistrableTaskOption func(*RegistrableTask) - validateFn func(Context, any, TaskAttributes, any, *Registry) (bool, error) + validateFn func(Context, any, TaskInvocation, any, *Registry) (bool, error) pureTaskExecuteFn func(MutableContext, any, TaskAttributes, any, *Registry) error sideEffectTaskExecuteFn func(context.Context, ComponentRef, TaskAttributes, any) error sideEffectTaskDiscardFn func(context.Context, ComponentRef, TaskAttributes, any) error @@ -57,14 +57,14 @@ func NewRegistrableSideEffectTask[C any, T any]( func( ctx Context, component any, - taskAttrs TaskAttributes, + taskInvocation TaskInvocation, taskData any, registry *Registry, ) (bool, error) { return handler.Validate( ctx, component.(C), - taskAttrs, + taskInvocation, taskData.(T), ) }, @@ -97,14 +97,14 @@ func NewRegistrablePureTask[C any, T any]( func( ctx Context, component any, - taskAttrs TaskAttributes, + taskInvocation TaskInvocation, taskData any, registry *Registry, ) (bool, error) { return handler.Validate( ctx, component.(C), - taskAttrs, + taskInvocation, taskData.(T), ) }, diff --git a/chasm/task.go b/chasm/task.go index 43698837b3..5f0a43632f 100644 --- a/chasm/task.go +++ b/chasm/task.go @@ -13,7 +13,8 @@ import ( var ErrTaskDiscarded = errors.New("standby task pending for too long") type ( - // TaskAttributes specifies scheduling metadata for a task. + // TaskAttributes specifies scheduling metadata for a task, supplied by the component author when + // the task is added via [MutableContext.AddTask]. TaskAttributes struct { // ScheduledTime is when the task should fire. Use [TaskScheduledTimeImmediate] (zero value) // for tasks that should execute as soon as possible. @@ -24,6 +25,18 @@ type ( Destination string } + // TaskInvocation is passed to a task's Validate callback. It carries the task's [TaskAttributes] + // together with framework-supplied state for the current processing attempt. + TaskInvocation struct { + TaskAttributes + // Attempt is the current processing attempt for this task, starting at 1. It comes from the + // task executable and is not persisted; it resets to 1 on shard reload and on active or + // standby failover. It is 0 when the task is validated outside of task processing, such as + // during transaction close. A best effort validator may compare it against a threshold and + // return false to give up on a task that would otherwise never become invalid on its own. + Attempt int + } + // SideEffectTaskHandler handles side effect tasks that run outside of the state lock and have access to a Go // context to perform I/O and access chasm engine methods such as [UpdateComponent]. Implementations must embed // [SideEffectTaskHandlerBase]. @@ -69,7 +82,7 @@ type ( // - (true, nil) if the task is valid and should be executed // - (false, nil) if the task should be silently dropped (it's no longer relevant) // - (anything, error) if validation fails with an error - Validate(Context, C, TaskAttributes, T) (bool, error) + Validate(Context, C, TaskInvocation, T) (bool, error) } ) diff --git a/chasm/task_mock.go b/chasm/task_mock.go index 4b41f5f50a..5f55deb2e6 100644 --- a/chasm/task_mock.go +++ b/chasm/task_mock.go @@ -69,7 +69,7 @@ func (mr *MockSideEffectTaskHandlerMockRecorder[C, T]) Execute(arg0, arg1, arg2, } // Validate mocks base method. -func (m *MockSideEffectTaskHandler[C, T]) Validate(arg0 Context, arg1 C, arg2 TaskAttributes, arg3 T) (bool, error) { +func (m *MockSideEffectTaskHandler[C, T]) Validate(arg0 Context, arg1 C, arg2 TaskInvocation, arg3 T) (bool, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "Validate", arg0, arg1, arg2, arg3) ret0, _ := ret[0].(bool) @@ -134,7 +134,7 @@ func (mr *MockPureTaskHandlerMockRecorder[C, T]) Execute(arg0, arg1, arg2, arg3 } // Validate mocks base method. -func (m *MockPureTaskHandler[C, T]) Validate(arg0 Context, arg1 C, arg2 TaskAttributes, arg3 T) (bool, error) { +func (m *MockPureTaskHandler[C, T]) Validate(arg0 Context, arg1 C, arg2 TaskInvocation, arg3 T) (bool, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "Validate", arg0, arg1, arg2, arg3) ret0, _ := ret[0].(bool) @@ -185,7 +185,7 @@ func (m *MockTaskValidator[C, T]) EXPECT() *MockTaskValidatorMockRecorder[C, T] } // Validate mocks base method. -func (m *MockTaskValidator[C, T]) Validate(arg0 Context, arg1 C, arg2 TaskAttributes, arg3 T) (bool, error) { +func (m *MockTaskValidator[C, T]) Validate(arg0 Context, arg1 C, arg2 TaskInvocation, arg3 T) (bool, error) { m.ctrl.T.Helper() ret := m.ctrl.Call(m, "Validate", arg0, arg1, arg2, arg3) ret0, _ := ret[0].(bool) diff --git a/chasm/tree.go b/chasm/tree.go index 1968eb9e4c..5aafdbe0e6 100644 --- a/chasm/tree.go +++ b/chasm/tree.go @@ -2130,7 +2130,7 @@ func (n *Node) deserializeComponentTask( // This method assumes component value is already hydrated. func (n *Node) validateTask( validateContext Context, - taskAttributes TaskAttributes, + taskInvocation TaskInvocation, taskInstance any, ) (_ bool, retErr error) { registableTask, ok := n.registry.taskFor(taskInstance) @@ -2155,7 +2155,7 @@ func (n *Node) validateTask( return registableTask.validateFn( validateContext, n.value, - taskAttributes, + taskInvocation, taskInstance, n.registry, ) @@ -2176,9 +2176,11 @@ func (n *Node) closeTransactionCleanupInvalidTasks( valid, err := n.validateTask( validateContext, - TaskAttributes{ - ScheduledTime: existingTask.ScheduledTime.AsTime(), - Destination: existingTask.Destination, + TaskInvocation{ + TaskAttributes: TaskAttributes{ + ScheduledTime: existingTask.ScheduledTime.AsTime(), + Destination: existingTask.Destination, + }, }, existingTaskInstance, ) @@ -2258,7 +2260,7 @@ func (n *Node) closeTransactionHandleNewTasks( valid, err := n.validateTask( validateContext, - newTask.attributes, + TaskInvocation{TaskAttributes: newTask.attributes}, newTask.task, ) if err != nil { @@ -3513,7 +3515,7 @@ func (n *Node) ExecutePureTask( } // Run the task's registered value before execution. - valid, err := n.validateTask(validationContext, taskAttributes, taskInstance) + valid, err := n.validateTask(validationContext, TaskInvocation{TaskAttributes: taskAttributes}, taskInstance) if err != nil { return false, err } @@ -3672,9 +3674,12 @@ func (n *Node) ValidateSideEffectTask( isValidByComponent, retErr = node.validateTask( validateCtx, - TaskAttributes{ - ScheduledTime: chasmTask.GetVisibilityTime(), - Destination: chasmTask.Destination, + TaskInvocation{ + TaskAttributes: TaskAttributes{ + ScheduledTime: chasmTask.GetVisibilityTime(), + Destination: chasmTask.Destination, + }, + Attempt: chasmTask.Attempt, }, chasmTask.DeserializedTask.Interface(), ) @@ -3787,7 +3792,7 @@ func (n *Node) invokeSideEffectTaskFn( componentInitialVT: taskInfo.ComponentInitialVersionedTransition, // Validate the Ref only once it is accessed by the task's handler. - validationFn: makeValidationFn(registrableTask, validate, taskAttributes, taskValue), + validationFn: makeValidationFn(registrableTask, validate, chasmTask.Attempt, taskAttributes, taskValue), } ctx = newContextWithOperationIntent(ctx, OperationIntentProgress) @@ -3828,6 +3833,7 @@ func (n *Node) ComponentByPath( func makeValidationFn( registrableTask *RegistrableTask, validate func(NodeBackend, Context, Component) error, + attempt int, taskAttributes TaskAttributes, taskValue reflect.Value, ) func(NodeBackend, Context, Component, *Registry) error { @@ -3845,7 +3851,7 @@ func makeValidationFn( valid, err := registrableTask.validateFn( ctx, component, - taskAttributes, + TaskInvocation{TaskAttributes: taskAttributes, Attempt: attempt}, taskValue.Interface(), registry, ) diff --git a/chasm/tree_test.go b/chasm/tree_test.go index 53550d2bf5..06100bc47e 100644 --- a/chasm/tree_test.go +++ b/chasm/tree_test.go @@ -2719,7 +2719,7 @@ func (s *nodeSuite) TestCloseTransaction_CleanupTasksAfterInvalidTask() { s.NotNil(root) s.testLibrary.mockPureTaskHandler.EXPECT(). - Validate(gomock.Any(), gomock.Any(), gomock.Eq(task1Attributes), gomock.Eq(task1)). + Validate(gomock.Any(), gomock.Any(), gomock.Eq(TaskInvocation{TaskAttributes: task1Attributes}), gomock.Eq(task1)). Return(false, nil). Times(1) executed, err := root.ExecutePureTask(s.T().Context(), task1Attributes, task1) @@ -3817,9 +3817,9 @@ func (s *nodeSuite) TestExecuteImmediatePureTask() { // One valid task, one invalid task s.testLibrary.mockPureTaskHandler.EXPECT(). - Validate(gomock.Any(), gomock.Any(), gomock.Eq(taskAttributes), gomock.Any()).Return(false, nil).Times(1) + Validate(gomock.Any(), gomock.Any(), gomock.Eq(TaskInvocation{TaskAttributes: taskAttributes}), gomock.Any()).Return(false, nil).Times(1) s.testLibrary.mockPureTaskHandler.EXPECT(). - Validate(gomock.Any(), gomock.Any(), gomock.Eq(taskAttributes), gomock.Any()).Return(true, nil).Times(1) + Validate(gomock.Any(), gomock.Any(), gomock.Eq(TaskInvocation{TaskAttributes: taskAttributes}), gomock.Any()).Return(true, nil).Times(1) s.testLibrary.mockPureTaskHandler.EXPECT(). Execute( gomock.AssignableToTypeOf(&mutableCtx{}), @@ -3864,7 +3864,7 @@ func (s *nodeSuite) TestImmediatePureTaskNowStableWithinTaskOnly() { ) s.testLibrary.mockPureTaskHandler.EXPECT(). - Validate(gomock.Any(), gomock.Any(), gomock.Eq(taskAttributes), gomock.Any()).Return(true, nil).Times(2) + Validate(gomock.Any(), gomock.Any(), gomock.Eq(TaskInvocation{TaskAttributes: taskAttributes}), gomock.Any()).Return(true, nil).Times(2) var observedTimes []time.Time s.testLibrary.mockPureTaskHandler.EXPECT(). @@ -4108,7 +4108,7 @@ func (s *nodeSuite) TestExecutePureTask() { expectValidate := func(retValue bool, errValue error) { s.testLibrary.mockPureTaskHandler.EXPECT(). - Validate(gomock.Any(), gomock.Any(), gomock.Eq(taskAttributes), gomock.Eq(pureTask)). + Validate(gomock.Any(), gomock.Any(), gomock.Eq(TaskInvocation{TaskAttributes: taskAttributes}), gomock.Eq(pureTask)). Return(retValue, errValue). Times(1) } @@ -4234,8 +4234,8 @@ func (s *nodeSuite) TestExecuteSideEffectTask() { gomock.Any(), gomock.Any(), gomock.Eq(TaskAttributes{ - chasmTask.GetVisibilityTime(), - chasmTask.Destination, + ScheduledTime: chasmTask.GetVisibilityTime(), + Destination: chasmTask.Destination, }), gomock.Any(), ).DoAndReturn( @@ -4496,9 +4496,11 @@ func (s *nodeSuite) TestValidateSideEffectTask() { Validate( gomock.AssignableToTypeOf((*immutableCtx)(nil)), gomock.AssignableToTypeOf(componentType), - gomock.Eq(TaskAttributes{ - ScheduledTime: chasmTask.GetVisibilityTime(), - Destination: chasmTask.Destination, + gomock.Eq(TaskInvocation{ + TaskAttributes: TaskAttributes{ + ScheduledTime: chasmTask.GetVisibilityTime(), + Destination: chasmTask.Destination, + }, }), gomock.AssignableToTypeOf(&TestSideEffectTask{}), ).Return(retValue, errValue).Times(1) @@ -4512,6 +4514,24 @@ func (s *nodeSuite) TestValidateSideEffectTask() { s.NoError(err) s.True(chasmTask.DeserializedTask.IsValid()) + // The physical task's attempt is threaded into the validator's TaskInvocation. + chasmTask.Attempt = 7 + s.testLibrary.mockSideEffectTaskHandler.EXPECT(). + Validate( + gomock.AssignableToTypeOf((*immutableCtx)(nil)), + gomock.AssignableToTypeOf((*TestComponent)(nil)), + gomock.Any(), + gomock.AssignableToTypeOf(&TestSideEffectTask{}), + ).DoAndReturn(func(_ Context, _ any, inv TaskInvocation, _ *TestSideEffectTask) (bool, error) { + s.Equal(7, inv.Attempt) + return true, nil + }).Times(1) + isTaskInTree, isValidByComponent, err = root.ValidateSideEffectTask(ctx, chasmTask) + s.True(isTaskInTree) + s.True(isValidByComponent) + s.NoError(err) + chasmTask.Attempt = 0 + // Task is in tree but component says invalid. expectValidate((*TestComponent)(nil), false, nil) isTaskInTree, isValidByComponent, err = root.ValidateSideEffectTask(ctx, chasmTask) diff --git a/chasm/visibility.go b/chasm/visibility.go index 7b81dc64aa..23799c4f95 100644 --- a/chasm/visibility.go +++ b/chasm/visibility.go @@ -355,7 +355,7 @@ var defaultVisibilityTaskHandler = &visibilityTaskHandler{} func (v *visibilityTaskHandler) Validate( _ Context, component *Visibility, - _ TaskAttributes, + _ TaskInvocation, task *persistencespb.ChasmVisibilityTaskData, ) (bool, error) { return task.TransitionCount == component.Data.TransitionCount, nil diff --git a/chasm/visibility_task_test.go b/chasm/visibility_task_test.go index b991b4e2c4..8a129e3c14 100644 --- a/chasm/visibility_task_test.go +++ b/chasm/visibility_task_test.go @@ -15,12 +15,12 @@ func TestTaskValidator(t *testing.T) { } visibility.Data.TransitionCount = 1 - valid, err := defaultVisibilityTaskHandler.Validate(ctx, visibility, TaskAttributes{}, task) + valid, err := defaultVisibilityTaskHandler.Validate(ctx, visibility, TaskInvocation{}, task) require.NoError(t, err) require.False(t, valid) visibility.Data.TransitionCount = task.TransitionCount - valid, err = defaultVisibilityTaskHandler.Validate(ctx, visibility, TaskAttributes{}, task) + valid, err = defaultVisibilityTaskHandler.Validate(ctx, visibility, TaskInvocation{}, task) require.NoError(t, err) require.True(t, valid) } diff --git a/service/history/chasm_task_util_test.go b/service/history/chasm_task_util_test.go index 30d39e898c..8364e39f23 100644 --- a/service/history/chasm_task_util_test.go +++ b/service/history/chasm_task_util_test.go @@ -29,7 +29,7 @@ type discardableTestTaskHandler struct { chasm.SideEffectTaskHandlerBase[*discardableTestTask] } -func (e *discardableTestTaskHandler) Validate(_ chasm.Context, _ any, _ chasm.TaskAttributes, _ *discardableTestTask) (bool, error) { +func (e *discardableTestTaskHandler) Validate(_ chasm.Context, _ any, _ chasm.TaskInvocation, _ *discardableTestTask) (bool, error) { return true, nil } @@ -64,7 +64,7 @@ type nonDiscardableTestTaskHandler struct { chasm.SideEffectTaskHandlerBase[*nonDiscardableTestTask] } -func (e *nonDiscardableTestTaskHandler) Validate(_ chasm.Context, _ any, _ chasm.TaskAttributes, _ *nonDiscardableTestTask) (bool, error) { +func (e *nonDiscardableTestTaskHandler) Validate(_ chasm.Context, _ any, _ chasm.TaskInvocation, _ *nonDiscardableTestTask) (bool, error) { return true, nil } diff --git a/service/history/outbound_queue_active_task_executor.go b/service/history/outbound_queue_active_task_executor.go index 12f523d51f..8270aa8a89 100644 --- a/service/history/outbound_queue_active_task_executor.go +++ b/service/history/outbound_queue_active_task_executor.go @@ -103,6 +103,7 @@ func (e *outboundQueueActiveTaskExecutor) Execute( case *tasks.StateMachineOutboundTask: return respond(e.executeStateMachineTask(ctx, task)) case *tasks.ChasmTask: + task.Attempt = executable.Attempt() return respond(e.executeChasmSideEffectTask(ctx, task)) case *tasks.WorkerCommandsTask: return respond(e.workerCommandsDispatcher.Execute(ctx, task, executable.Attempt(), namespaceTag.Value)) diff --git a/service/history/outbound_queue_active_task_executor_test.go b/service/history/outbound_queue_active_task_executor_test.go index 7a526e970b..9d8fb4b4a8 100644 --- a/service/history/outbound_queue_active_task_executor_test.go +++ b/service/history/outbound_queue_active_task_executor_test.go @@ -194,6 +194,7 @@ func (s *outboundQueueActiveTaskExecutorSuite) TestExecute_ChasmTask() { tc.setupMocks(task) s.mockExecutable.EXPECT().GetTask().Return(task).AnyTimes() + s.mockExecutable.EXPECT().Attempt().Return(1).AnyTimes() s.mockExecutable.EXPECT().GetWorkflowID().Return("").AnyTimes() result := s.executor.Execute(ctx, s.mockExecutable) @@ -252,6 +253,7 @@ func (s *outboundQueueActiveTaskExecutorSuite) TestExecute_PreValidationFails() task := tc.setupTask() tc.setupMocks(task) s.mockExecutable.EXPECT().GetTask().Return(task) + s.mockExecutable.EXPECT().Attempt().Return(1).AnyTimes() s.mockExecutable.EXPECT().GetWorkflowID().Return("").AnyTimes() result := s.executor.Execute(ctx, s.mockExecutable) diff --git a/service/history/outbound_queue_standby_task_executor.go b/service/history/outbound_queue_standby_task_executor.go index 1099f08dd6..16a5f4fd7f 100644 --- a/service/history/outbound_queue_standby_task_executor.go +++ b/service/history/outbound_queue_standby_task_executor.go @@ -103,6 +103,7 @@ func (e *outboundQueueStandbyTaskExecutor) Execute( case *tasks.StateMachineOutboundTask: return respond(e.executeStateMachineTask(ctx, task, nsName)) case *tasks.ChasmTask: + task.Attempt = executable.Attempt() return respond(e.executeChasmSideEffectTask(ctx, task)) case *tasks.WorkerCommandsTask: // Worker commands are best-effort and only executed on the active cluster. @@ -186,12 +187,13 @@ func (e *outboundQueueStandbyTaskExecutor) executeChasmSideEffectTask( return err } - isTaskInTree, _, err := validateChasmSideEffectTask(ctx, ms, task) + isTaskInTree, isValid, err := validateChasmSideEffectTask(ctx, ms, task) if err != nil { return err } - if !isTaskInTree { - // Replication has removed the logical task — drop the physical task. + if !isTaskInTree || !isValid { + // Replication has removed the logical task, or the component reports it + // invalid — drop the physical task. return nil } diff --git a/service/history/outbound_queue_standby_task_executor_test.go b/service/history/outbound_queue_standby_task_executor_test.go index 718f36e08f..8a7d82a299 100644 --- a/service/history/outbound_queue_standby_task_executor_test.go +++ b/service/history/outbound_queue_standby_task_executor_test.go @@ -173,7 +173,7 @@ func (s *outboundQueueStandbyTaskExecutorSuite) TestExecute_ChasmTask() { expectedError: consts.ErrTaskRetry.Error(), }, { - name: "in tree but component invalid (e.g. code-deployment) - retries", + name: "in tree but component invalid (e.g. code-deployment) - drop physical task", setupMocks: func(task *tasks.ChasmTask) { s.mockWorkflowCache.EXPECT(). GetOrCreateChasmExecution(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), tests.ArchetypeID, gomock.Any()). @@ -191,8 +191,8 @@ func (s *outboundQueueStandbyTaskExecutorSuite) TestExecute_ChasmTask() { ValidateSideEffectTask(gomock.Any(), gomock.Any()). Return(true, false, nil) }, - expectHandlerCalled: true, - expectedError: consts.ErrTaskRetry.Error(), + expectHandlerCalled: false, + expectedError: "", }, { name: "not in tree - replication removed it, drop physical task", @@ -250,6 +250,7 @@ func (s *outboundQueueStandbyTaskExecutorSuite) TestExecute_ChasmTask() { tc.setupMocks(task) s.mockExecutable.EXPECT().GetTask().Return(task).AnyTimes() + s.mockExecutable.EXPECT().Attempt().Return(1).AnyTimes() s.mockExecutable.EXPECT().GetWorkflowID().Return("").AnyTimes() result := s.executor.Execute(ctx, s.mockExecutable) @@ -311,6 +312,7 @@ func (s *outboundQueueStandbyTaskExecutorSuite) TestExecute_PreValidationFails() task := tc.setupTask() tc.setupMocks(task) s.mockExecutable.EXPECT().GetTask().Return(task) + s.mockExecutable.EXPECT().Attempt().Return(1).AnyTimes() s.mockExecutable.EXPECT().GetWorkflowID().Return("").AnyTimes() result := s.executor.Execute(ctx, s.mockExecutable) @@ -367,6 +369,7 @@ func (s *outboundQueueStandbyTaskExecutorSuite) TestExecute_ChasmTask_Discard() executable := queues.NewMockExecutable(s.controller) executable.EXPECT().GetTask().Return(task).AnyTimes() + executable.EXPECT().Attempt().Return(1).AnyTimes() executable.EXPECT().GetWorkflowID().Return(task.WorkflowKey.WorkflowID).AnyTimes() executor := newOutboundQueueStandbyTaskExecutor( diff --git a/service/history/tasks/chasm_task.go b/service/history/tasks/chasm_task.go index c99fac6e2d..d8ed72f266 100644 --- a/service/history/tasks/chasm_task.go +++ b/service/history/tasks/chasm_task.go @@ -70,6 +70,10 @@ type ChasmTask struct { // In-memory only outboundTaskGroup string // set to the registered task's taskgroup after deserialization for outbound tasks DeserializedTask reflect.Value + // Attempt is the current processing attempt for this physical task, starting at 1. It is copied + // from the task executable before execution or validation and is not persisted. Surfaced to + // CHASM handlers via chasm.TaskAttributes.Attempt. + Attempt int } var _ Task = &ChasmTask{} diff --git a/service/history/timer_queue_active_task_executor.go b/service/history/timer_queue_active_task_executor.go index dc07530e4f..86bee41bde 100644 --- a/service/history/timer_queue_active_task_executor.go +++ b/service/history/timer_queue_active_task_executor.go @@ -123,6 +123,7 @@ func (t *timerQueueActiveTaskExecutor) Execute( case *tasks.ChasmTaskPure: err = t.executeChasmPureTimerTask(ctx, task) case *tasks.ChasmTask: + task.Attempt = executable.Attempt() err = t.executeChasmSideEffectTimerTask(ctx, task) case *tasks.TimeSkippingTimerTask: err = t.executeTimeSkippingTimerTask(ctx, task) diff --git a/service/history/timer_queue_standby_task_executor.go b/service/history/timer_queue_standby_task_executor.go index 2714d2fba2..078aabef1a 100644 --- a/service/history/timer_queue_standby_task_executor.go +++ b/service/history/timer_queue_standby_task_executor.go @@ -107,6 +107,7 @@ func (t *timerQueueStandbyTaskExecutor) Execute( case *tasks.ChasmTaskPure: err = t.executeChasmPureTimerTask(ctx, task) case *tasks.ChasmTask: + task.Attempt = executable.Attempt() err = t.executeChasmSideEffectTimerTask(ctx, task) case *tasks.TimeSkippingTimerTask: err = t.executeTimeSkippingTimerTask(ctx, task) @@ -169,12 +170,13 @@ func (t *timerQueueStandbyTaskExecutor) executeChasmSideEffectTimerTask( ms historyi.MutableState, _ historyi.ReleaseWorkflowContextFunc, ) (any, error) { - isTaskInTree, _, err := validateChasmSideEffectTask(ctx, ms, task) + isTaskInTree, isValid, err := validateChasmSideEffectTask(ctx, ms, task) if err != nil { return nil, err } - if !isTaskInTree { - // Replication has removed the logical task — drop the physical task. + if !isTaskInTree || !isValid { + // Replication has removed the logical task, or the component reports it + // invalid — drop the physical task. return nil, nil } diff --git a/service/history/timer_queue_standby_task_executor_test.go b/service/history/timer_queue_standby_task_executor_test.go index e1325e3900..d70f9dac16 100644 --- a/service/history/timer_queue_standby_task_executor_test.go +++ b/service/history/timer_queue_standby_task_executor_test.go @@ -2358,11 +2358,11 @@ func (s *timerQueueStandbyTaskExecutorSuite) TestExecuteChasmSideEffectTimerTask s.NotNil(resp) s.ErrorIs(consts.ErrTaskRetry, resp.ExecutionErr) - // Task in tree but component says invalid (e.g. code-deployment) — still retry. + // Task in tree but component says invalid (e.g. code-deployment) — drop the physical task. expectValidate(true, false, nil) resp = timerQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(timerTask)) s.NotNil(resp) - s.ErrorIs(consts.ErrTaskRetry, resp.ExecutionErr) + s.NoError(resp.ExecutionErr) // Task not in tree — replication removed it, drop the physical task. expectValidate(false, false, nil) diff --git a/service/history/transfer_queue_active_task_executor.go b/service/history/transfer_queue_active_task_executor.go index d4f1f80b8d..fe961c760f 100644 --- a/service/history/transfer_queue_active_task_executor.go +++ b/service/history/transfer_queue_active_task_executor.go @@ -171,6 +171,7 @@ func (t *transferQueueActiveTaskExecutor) execute( case *tasks.DeleteExecutionTask: err = t.processDeleteExecutionTask(ctx, task) case *tasks.ChasmTask: + task.Attempt = executable.Attempt() err = t.executeChasmSideEffectTransferTask(ctx, task) default: err = errUnknownTransferTask diff --git a/service/history/transfer_queue_standby_task_executor.go b/service/history/transfer_queue_standby_task_executor.go index b854dc3b7e..a6fc09d831 100644 --- a/service/history/transfer_queue_standby_task_executor.go +++ b/service/history/transfer_queue_standby_task_executor.go @@ -107,6 +107,7 @@ func (t *transferQueueStandbyTaskExecutor) Execute( case *tasks.DeleteExecutionTask: err = t.processDeleteExecutionTask(ctx, task, false) case *tasks.ChasmTask: + task.Attempt = executable.Attempt() err = t.executeChasmSideEffectTransferTask(ctx, task) default: err = errUnknownTransferTask @@ -129,12 +130,13 @@ func (t *transferQueueStandbyTaskExecutor) executeChasmSideEffectTransferTask( ms historyi.MutableState, _ historyi.ReleaseWorkflowContextFunc, ) (any, error) { - isTaskInTree, _, err := validateChasmSideEffectTask(ctx, ms, task) + isTaskInTree, isValid, err := validateChasmSideEffectTask(ctx, ms, task) if err != nil { return nil, err } - if !isTaskInTree { - // Replication has removed the logical task — drop the physical task. + if !isTaskInTree || !isValid { + // Replication has removed the logical task, or the component reports it + // invalid — drop the physical task. return nil, nil } diff --git a/service/history/transfer_queue_standby_task_executor_test.go b/service/history/transfer_queue_standby_task_executor_test.go index 69936f9083..993e992884 100644 --- a/service/history/transfer_queue_standby_task_executor_test.go +++ b/service/history/transfer_queue_standby_task_executor_test.go @@ -355,11 +355,11 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestExecuteChasmSideEffectTransf s.NotNil(resp) s.ErrorIs(consts.ErrTaskRetry, resp.ExecutionErr) - // Task in tree but component says invalid (e.g. code-deployment) — still retry. + // Task in tree but component says invalid (e.g. code-deployment) — drop the physical task. expectValidate(true, false, nil) resp = transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask)) s.NotNil(resp) - s.ErrorIs(consts.ErrTaskRetry, resp.ExecutionErr) + s.NoError(resp.ExecutionErr) // Task not in tree — replication removed it, drop the physical task. expectValidate(false, false, nil) diff --git a/service/history/visibility_queue_task_executor.go b/service/history/visibility_queue_task_executor.go index dec9e6a3b4..32b1e56d24 100644 --- a/service/history/visibility_queue_task_executor.go +++ b/service/history/visibility_queue_task_executor.go @@ -113,6 +113,7 @@ func (t *visibilityQueueTaskExecutor) Execute( case *tasks.DeleteExecutionVisibilityTask: err = t.processDeleteExecution(ctx, task) case *tasks.ChasmTask: + task.Attempt = executable.Attempt() err = t.processChasmTask(ctx, task) default: err = errUnknownVisibilityTask