From 629e63149c2df15a1ce05029ceec667d43b0296c Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Thu, 25 Jun 2026 17:48:32 -0700 Subject: [PATCH 01/22] Move worker commands dispatcher to common/workercommands Move the entire workerCommandsTaskDispatcher from service/history to common/workercommands so it can be reused by CHASM standalone activities (which cannot import service/history due to circular dependency). Co-Authored-By: Claude Opus 4.6 --- .../workercommands/dispatcher.go | 42 +++++++++---------- .../workercommands/dispatcher_test.go | 38 ++++++++--------- .../outbound_queue_active_task_executor.go | 7 ++-- 3 files changed, 44 insertions(+), 43 deletions(-) rename service/history/worker_commands_task_dispatcher.go => common/workercommands/dispatcher.go (84%) rename service/history/worker_commands_task_dispatcher_test.go => common/workercommands/dispatcher_test.go (91%) diff --git a/service/history/worker_commands_task_dispatcher.go b/common/workercommands/dispatcher.go similarity index 84% rename from service/history/worker_commands_task_dispatcher.go rename to common/workercommands/dispatcher.go index baaf9f2564..17b43c8ff6 100644 --- a/service/history/worker_commands_task_dispatcher.go +++ b/common/workercommands/dispatcher.go @@ -1,4 +1,4 @@ -package history +package workercommands import ( "context" @@ -26,18 +26,18 @@ import ( ) const ( - workerCommandsTaskTimeout = time.Second * 10 * debug.TimeoutMultiplier - workerCommandsMaxTaskAttempt = 3 + DispatchTimeout = time.Second * 10 * debug.TimeoutMultiplier + MaxTaskAttempts = 3 // Nexus service and operation names for worker commands. // TODO: Replace with workerservicepb.WorkerService.ServiceName and // workerservicepb.WorkerService.ExecuteCommands.Name() once the Nexus service // descriptor is published in go.temporal.io/api. - workerCommandsServiceName = "temporal.api.nexusservices.workerservice.v1.WorkerService" - workerCommandsOperationName = "ExecuteCommands" + ServiceName = "temporal.api.nexusservices.workerservice.v1.WorkerService" + OperationName = "ExecuteCommands" ) -// workerCommandsTaskDispatcher dispatches worker commands to workers via Nexus. +// Dispatcher dispatches worker commands to workers via Nexus. // // Failure scenarios: // - No worker polling: matching returns RequestTimeout -> *nexus.HandlerError{Type: UpstreamTimeout}. @@ -51,23 +51,23 @@ const ( // *temporal.CanceledError. Permanent — the worker contract requires success for all // defined commands, so this indicates a bug or version incompatibility. // -// Retryable errors are capped at workerCommandsMaxTaskAttempt attempts (in-memory). These +// Retryable errors are capped at MaxTaskAttempts attempts (in-memory). These // commands are best-effort — the activity will eventually time out anyway — so excessive // retries waste resources. The counter resets on shard movement, which is acceptable. -type workerCommandsTaskDispatcher struct { +type Dispatcher struct { matchingClient resource.MatchingClient config *configs.Config metricsHandler metrics.Handler logger log.Logger } -func newWorkerCommandsTaskDispatcher( +func NewDispatcher( matchingClient resource.MatchingClient, config *configs.Config, metricsHandler metrics.Handler, logger log.Logger, -) *workerCommandsTaskDispatcher { - return &workerCommandsTaskDispatcher{ +) *Dispatcher { + return &Dispatcher{ matchingClient: matchingClient, config: config, metricsHandler: metricsHandler, @@ -75,13 +75,13 @@ func newWorkerCommandsTaskDispatcher( } } -func (d *workerCommandsTaskDispatcher) execute( +func (d *Dispatcher) Execute( ctx context.Context, task *tasks.WorkerCommandsTask, attempt int, namespaceName string, ) error { - if attempt > workerCommandsMaxTaskAttempt { + if attempt > MaxTaskAttempts { d.logger.Info("Worker commands task exceeded max attempts, dropping", tag.WorkflowID(task.WorkflowID), tag.WorkflowRunID(task.RunID), @@ -107,13 +107,13 @@ func (d *workerCommandsTaskDispatcher) execute( return nil } - ctx, cancel := context.WithTimeout(ctx, workerCommandsTaskTimeout) + ctx, cancel := context.WithTimeout(ctx, DispatchTimeout) defer cancel() return d.dispatchToWorker(ctx, task, namespaceName) } -func (d *workerCommandsTaskDispatcher) dispatchToWorker( +func (d *Dispatcher) dispatchToWorker( ctx context.Context, task *tasks.WorkerCommandsTask, namespaceName string, @@ -141,8 +141,8 @@ func (d *workerCommandsTaskDispatcher) dispatchToWorker( Header: map[string]string{}, Variant: &nexuspb.Request_StartOperation{ StartOperation: &nexuspb.StartOperationRequest{ - Service: workerCommandsServiceName, - Operation: workerCommandsOperationName, + Service: ServiceName, + Operation: OperationName, Payload: requestPayload, }, }, @@ -170,7 +170,7 @@ func (d *workerCommandsTaskDispatcher) dispatchToWorker( return d.handleError(nexusErr, task, namespaceName) } -func (d *workerCommandsTaskDispatcher) handleError(nexusErr error, task *tasks.WorkerCommandsTask, namespaceName string) error { +func (d *Dispatcher) handleError(nexusErr error, task *tasks.WorkerCommandsTask, namespaceName string) error { var handlerErr *nexus.HandlerError if errors.As(nexusErr, &handlerErr) { // Handler-level error (transport, timeout, internal). These are constructed by @@ -211,18 +211,18 @@ func (d *workerCommandsTaskDispatcher) handleError(nexusErr error, task *tasks.W return nil } -func (d *workerCommandsTaskDispatcher) recordCommandMetrics(commands []*workerpb.WorkerCommand, namespaceName string, outcome string) { +func (d *Dispatcher) recordCommandMetrics(commands []*workerpb.WorkerCommand, namespaceName string, outcome string) { for _, cmd := range commands { metrics.WorkerCommandsSent.With(d.metricsHandler).Record( 1, metrics.NamespaceTag(namespaceName), metrics.OutcomeTag(outcome), - metrics.StringTag("command_type", workerCommandTypeName(cmd)), + metrics.StringTag("command_type", CommandTypeName(cmd)), ) } } -func workerCommandTypeName(cmd *workerpb.WorkerCommand) string { +func CommandTypeName(cmd *workerpb.WorkerCommand) string { switch cmd.GetType().(type) { case *workerpb.WorkerCommand_CancelActivity: return "cancel_activity" diff --git a/service/history/worker_commands_task_dispatcher_test.go b/common/workercommands/dispatcher_test.go similarity index 91% rename from service/history/worker_commands_task_dispatcher_test.go rename to common/workercommands/dispatcher_test.go index 16e7c846fc..6da6346269 100644 --- a/service/history/worker_commands_task_dispatcher_test.go +++ b/common/workercommands/dispatcher_test.go @@ -1,4 +1,4 @@ -package history +package workercommands import ( "context" @@ -44,7 +44,7 @@ func requireMetricValue(t *testing.T, snap map[string][]*metricstest.CapturedRec } func TestExecute_FeatureFlagOff_DropsTask(t *testing.T) { - d := &workerCommandsTaskDispatcher{ + d := &Dispatcher{ config: &configs.Config{ EnableCancelActivityWorkerCommand: func(string) bool { return false }, }, @@ -52,12 +52,12 @@ func TestExecute_FeatureFlagOff_DropsTask(t *testing.T) { } task := testWorkerCommandsTask() - err := d.execute(context.Background(), task, 1 /* attempt */, "test-namespace") + err := d.Execute(context.Background(), task, 1 /* attempt */, "test-namespace") require.NoError(t, err, "task should be silently dropped when feature flag is off") } func TestExecute_EmptyCommands_DropsTask(t *testing.T) { - d := &workerCommandsTaskDispatcher{ + d := &Dispatcher{ config: &configs.Config{ EnableCancelActivityWorkerCommand: func(string) bool { return true }, }, @@ -66,7 +66,7 @@ func TestExecute_EmptyCommands_DropsTask(t *testing.T) { task := testWorkerCommandsTask() task.Commands = nil - err := d.execute(context.Background(), task, 1 /* attempt */, "test-namespace") + err := d.Execute(context.Background(), task, 1 /* attempt */, "test-namespace") require.NoError(t, err, "task with no commands should be dropped") } @@ -75,7 +75,7 @@ func TestExecute_ExceedsMaxAttempts_DropsTask(t *testing.T) { capture := metricsHandler.StartCapture() defer metricsHandler.StopCapture(capture) - d := &workerCommandsTaskDispatcher{ + d := &Dispatcher{ config: &configs.Config{ EnableCancelActivityWorkerCommand: func(string) bool { return true }, }, @@ -84,7 +84,7 @@ func TestExecute_ExceedsMaxAttempts_DropsTask(t *testing.T) { } task := testWorkerCommandsTask() - err := d.execute(context.Background(), task, workerCommandsMaxTaskAttempt+1, "test-namespace") + err := d.Execute(context.Background(), task, MaxTaskAttempts+1, "test-namespace") require.NoError(t, err, "task should be dropped when max attempts exceeded") requireMetricValue(t, capture.Snapshot(), "max_attempts_exceeded") @@ -97,7 +97,7 @@ func TestExecute_AtMaxAttempt_StillExecutes(t *testing.T) { capture := metricsHandler.StartCapture() defer metricsHandler.StopCapture(capture) - d := &workerCommandsTaskDispatcher{ + d := &Dispatcher{ matchingClient: mockClient, config: &configs.Config{ EnableCancelActivityWorkerCommand: func(string) bool { return true }, @@ -122,7 +122,7 @@ func TestExecute_AtMaxAttempt_StillExecutes(t *testing.T) { }, nil) task := testWorkerCommandsTask() - err := d.execute(context.Background(), task, workerCommandsMaxTaskAttempt, "test-namespace") + err := d.Execute(context.Background(), task, MaxTaskAttempts, "test-namespace") require.NoError(t, err, "task at exactly max attempt should still execute") requireMetricValue(t, capture.Snapshot(), "success") @@ -135,7 +135,7 @@ func TestExecute_DispatchSuccess(t *testing.T) { capture := metricsHandler.StartCapture() defer metricsHandler.StopCapture(capture) - d := &workerCommandsTaskDispatcher{ + d := &Dispatcher{ matchingClient: mockClient, config: &configs.Config{ EnableCancelActivityWorkerCommand: func(string) bool { return true }, @@ -164,7 +164,7 @@ func TestExecute_DispatchSuccess(t *testing.T) { }) task := testWorkerCommandsTask() - err := d.execute(context.Background(), task, 1 /* attempt */, "test-namespace") + err := d.Execute(context.Background(), task, 1 /* attempt */, "test-namespace") require.NoError(t, err) require.NotNil(t, capturedReq) @@ -182,7 +182,7 @@ func TestExecute_DispatchRPCError(t *testing.T) { capture := metricsHandler.StartCapture() defer metricsHandler.StopCapture(capture) - d := &workerCommandsTaskDispatcher{ + d := &Dispatcher{ matchingClient: mockClient, config: &configs.Config{ EnableCancelActivityWorkerCommand: func(string) bool { return true }, @@ -195,7 +195,7 @@ func TestExecute_DispatchRPCError(t *testing.T) { nil, errors.New("connection refused")) task := testWorkerCommandsTask() - err := d.execute(context.Background(), task, 1 /* attempt */, "test-namespace") + err := d.Execute(context.Background(), task, 1 /* attempt */, "test-namespace") require.Error(t, err) require.Contains(t, err.Error(), "connection refused") @@ -209,7 +209,7 @@ func TestExecute_UpstreamTimeout(t *testing.T) { capture := metricsHandler.StartCapture() defer metricsHandler.StopCapture(capture) - d := &workerCommandsTaskDispatcher{ + d := &Dispatcher{ matchingClient: mockClient, config: &configs.Config{ EnableCancelActivityWorkerCommand: func(string) bool { return true }, @@ -226,7 +226,7 @@ func TestExecute_UpstreamTimeout(t *testing.T) { }, nil) task := testWorkerCommandsTask() - err := d.execute(context.Background(), task, 1 /* attempt */, "test-namespace") + err := d.Execute(context.Background(), task, 1 /* attempt */, "test-namespace") require.Error(t, err) var he *nexus.HandlerError @@ -241,7 +241,7 @@ func TestHandleError_WorkerError_ReturnNil(t *testing.T) { capture := metricsHandler.StartCapture() defer metricsHandler.StopCapture(capture) - d := &workerCommandsTaskDispatcher{ + d := &Dispatcher{ metricsHandler: metricsHandler, logger: log.NewNoopLogger(), } @@ -260,7 +260,7 @@ func TestHandleError_UpstreamTimeout_ReturnRetryable(t *testing.T) { capture := metricsHandler.StartCapture() defer metricsHandler.StopCapture(capture) - d := &workerCommandsTaskDispatcher{ + d := &Dispatcher{ metricsHandler: metricsHandler, logger: log.NewNoopLogger(), } @@ -282,7 +282,7 @@ func TestHandleError_NonRetryableHandlerError_ReturnNil(t *testing.T) { capture := metricsHandler.StartCapture() defer metricsHandler.StopCapture(capture) - d := &workerCommandsTaskDispatcher{ + d := &Dispatcher{ metricsHandler: metricsHandler, logger: log.NewNoopLogger(), } @@ -300,7 +300,7 @@ func TestHandleError_OtherHandlerError_ReturnRetryable(t *testing.T) { capture := metricsHandler.StartCapture() defer metricsHandler.StopCapture(capture) - d := &workerCommandsTaskDispatcher{ + d := &Dispatcher{ metricsHandler: metricsHandler, logger: log.NewNoopLogger(), } diff --git a/service/history/outbound_queue_active_task_executor.go b/service/history/outbound_queue_active_task_executor.go index a5cfe6844d..e8476d8d9d 100644 --- a/service/history/outbound_queue_active_task_executor.go +++ b/service/history/outbound_queue_active_task_executor.go @@ -11,6 +11,7 @@ import ( "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/resource" + "go.temporal.io/server/common/workercommands" "go.temporal.io/server/service/history/consts" historyi "go.temporal.io/server/service/history/interfaces" "go.temporal.io/server/service/history/queues" @@ -26,7 +27,7 @@ const ( type outboundQueueActiveTaskExecutor struct { stateMachineEnvironment chasmEngine chasm.Engine - workerCommandsTaskDispatcher *workerCommandsTaskDispatcher + workerCommandsDispatcher *workercommands.Dispatcher } var _ queues.Executor = &outboundQueueActiveTaskExecutor{} @@ -50,7 +51,7 @@ func newOutboundQueueActiveTaskExecutor( metricsHandler: scopedMetricsHandler, }, chasmEngine: chasmEngine, - workerCommandsTaskDispatcher: newWorkerCommandsTaskDispatcher( + workerCommandsDispatcher: workercommands.NewDispatcher( matchingClient, shardCtx.GetConfig(), scopedMetricsHandler, @@ -104,7 +105,7 @@ func (e *outboundQueueActiveTaskExecutor) Execute( case *tasks.ChasmTask: return respond(e.executeChasmSideEffectTask(ctx, task)) case *tasks.WorkerCommandsTask: - return respond(e.workerCommandsTaskDispatcher.execute(ctx, task, executable.Attempt(), namespaceTag.Value)) + return respond(e.workerCommandsDispatcher.Execute(ctx, task, executable.Attempt(), namespaceTag.Value)) } return respond(queueserrors.NewUnprocessableTaskError(fmt.Sprintf("unknown task type '%T'", task))) From ce0370d03c303546dbbd089e6561b3ba3296c6a1 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Thu, 25 Jun 2026 17:49:00 -0700 Subject: [PATCH 02/22] Fix const alignment Co-Authored-By: Claude Opus 4.6 --- common/workercommands/dispatcher.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/common/workercommands/dispatcher.go b/common/workercommands/dispatcher.go index 17b43c8ff6..79c3f2c714 100644 --- a/common/workercommands/dispatcher.go +++ b/common/workercommands/dispatcher.go @@ -26,7 +26,7 @@ import ( ) const ( - DispatchTimeout = time.Second * 10 * debug.TimeoutMultiplier + DispatchTimeout = time.Second * 10 * debug.TimeoutMultiplier MaxTaskAttempts = 3 // Nexus service and operation names for worker commands. From 81855db99c3695730a4d6798e5726e72c9a57cd6 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Thu, 25 Jun 2026 17:57:07 -0700 Subject: [PATCH 03/22] Fix struct field alignment Co-Authored-By: Claude Opus 4.6 --- service/history/outbound_queue_active_task_executor.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/service/history/outbound_queue_active_task_executor.go b/service/history/outbound_queue_active_task_executor.go index e8476d8d9d..12f523d51f 100644 --- a/service/history/outbound_queue_active_task_executor.go +++ b/service/history/outbound_queue_active_task_executor.go @@ -26,7 +26,7 @@ const ( type outboundQueueActiveTaskExecutor struct { stateMachineEnvironment - chasmEngine chasm.Engine + chasmEngine chasm.Engine workerCommandsDispatcher *workercommands.Dispatcher } From 77d9d489e62f8e8db53ee5ce21567b5482216987 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Thu, 25 Jun 2026 18:32:10 -0700 Subject: [PATCH 04/22] Add cancel command dispatch for standalone activities MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a standalone activity is cancelled or terminated while running on a worker, proactively notify the worker via a cancel command dispatched through the Nexus worker commands control queue. This is best-effort — the activity will eventually time out if the worker doesn't respond. Key changes: - Add CancelCommandDispatchTask side-effect task and handler - Store started-time ComponentRef in ActivityAttemptState so cancel tokens are byte-identical to poll tokens (SDK does exact byte match) - Set Execution field in AddActivityTaskRequest for correct WorkflowId/RunId - Add NewStandaloneActivityTaskToken to common/tasktoken - Wire cancel dispatch on cancel-requested and terminate transitions Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/activity.go | 54 ++++++++++- chasm/lib/activity/activity_tasks.go | 90 +++++++++++++++++++ chasm/lib/activity/config.go | 50 ++++++----- chasm/lib/activity/fx.go | 1 + .../gen/activitypb/v1/activity_state.pb.go | 32 +++++-- .../gen/activitypb/v1/tasks.go-helpers.pb.go | 37 ++++++++ .../activity/gen/activitypb/v1/tasks.pb.go | 46 +++++++++- chasm/lib/activity/library.go | 7 ++ .../activity/proto/v1/activity_state.proto | 8 ++ chasm/lib/activity/proto/v1/tasks.proto | 4 + chasm/lib/activity/statemachine.go | 10 +++ common/tasktoken/token.go | 30 ++++++- common/tasktoken/token_test.go | 50 +++++++++++ tests/activity_standalone_test.go | 87 ++++++++++++++++++ 14 files changed, 469 insertions(+), 37 deletions(-) create mode 100644 common/tasktoken/token_test.go diff --git a/chasm/lib/activity/activity.go b/chasm/lib/activity/activity.go index 3efce2027c..f231acd529 100644 --- a/chasm/lib/activity/activity.go +++ b/chasm/lib/activity/activity.go @@ -32,6 +32,7 @@ import ( "go.temporal.io/server/common/nexus/nexusrpc" "go.temporal.io/server/common/payload" serviceerrors "go.temporal.io/server/common/serviceerror" + "go.temporal.io/server/common/tasktoken" "go.temporal.io/server/common/tqid" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" @@ -201,10 +202,16 @@ func (a *Activity) createAddActivityTaskRequest(ctx chasm.Context, namespaceID s return nil, err } + key := ctx.ExecutionKey() + // Note: No need to set the vector clock here, as the components track version conflicts for read/write // TODO: Need to fill in VersionDirective once we decide how to handle versioning for standalone activities return &matchingservice.AddActivityTaskRequest{ - NamespaceId: namespaceID, + NamespaceId: namespaceID, + Execution: &commonpb.WorkflowExecution{ + WorkflowId: key.BusinessID, + RunId: key.RunID, + }, ScheduleToStartTimeout: a.ScheduleToStartTimeout, TaskQueue: a.GetTaskQueue(), Priority: a.GetPriority(), @@ -213,6 +220,25 @@ func (a *Activity) createAddActivityTaskRequest(ctx chasm.Context, namespaceID s }, nil } +// buildCancelCommandTaskToken builds the serialized task token for a cancel command. +// Uses the ComponentRef captured at start time so the token is byte-identical to the poll token. +func (a *Activity) buildCancelCommandTaskToken(ctx chasm.Context, activityRef chasm.ComponentRef) ([]byte, error) { + attempt := a.LastAttempt.Get(ctx) + key := ctx.ExecutionKey() + + token := tasktoken.NewStandaloneActivityTaskToken( + key.NamespaceID, + key.BusinessID, // workflowID — for standalone activities, BusinessID is the ActivityId + key.RunID, + key.BusinessID, // activityId + a.GetActivityType().GetName(), + attempt.GetCount(), + attempt.GetStartedComponentRef(), + ) + + return token.Marshal() +} + // HandleStarted updates the activity on recording activity task started and populates the response. func (a *Activity) HandleStarted(ctx chasm.MutableContext, request *historyservice.RecordActivityTaskStartedRequest) ( *historyservice.RecordActivityTaskStartedResponse, error, @@ -571,6 +597,13 @@ func (a *Activity) Terminate( return chasm.TerminateComponentResponse{}, nil } + // If the activity is running on a worker, proactively notify the worker via Nexus. + // Must be done before the transition since it checks current status. + if a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_STARTED || + a.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED { + a.addCancelCommandDispatchTask(ctx) + } + metricsHandler, err := a.enrichMetricsHandler(ctx, metrics.ActivityTerminatedScope) if err != nil { return chasm.TerminateComponentResponse{}, err @@ -593,6 +626,23 @@ func (a *Activity) getOrCreateLastHeartbeat(ctx chasm.MutableContext) *activityp return heartbeat } +// addCancelCommandDispatchTask schedules a side-effect task to dispatch a cancel command to the +// worker via the Nexus worker commands control queue. No-op if the worker doesn't support worker +// commands (i.e., has no control queue). +func (a *Activity) addCancelCommandDispatchTask(ctx chasm.MutableContext) { + controlQueue := a.LastAttempt.Get(ctx).GetWorkerControlTaskQueue() + if controlQueue == "" { + return + } + ctx.AddTask( + a, + chasm.TaskAttributes{ + Destination: controlQueue, + }, + &activitypb.CancelCommandDispatchTask{}, + ) +} + func (a *Activity) handleCancellationRequested(ctx chasm.MutableContext, request *activitypb.RequestCancelActivityExecutionRequest) ( *activitypb.RequestCancelActivityExecutionResponse, error, ) { @@ -636,6 +686,8 @@ func (a *Activity) handleCancellationRequested(ctx chasm.MutableContext, request if err != nil { return nil, err } + } else { + a.addCancelCommandDispatchTask(ctx) } return &activitypb.RequestCancelActivityExecutionResponse{}, nil diff --git a/chasm/lib/activity/activity_tasks.go b/chasm/lib/activity/activity_tasks.go index e22b2f586a..089dcf9f2b 100644 --- a/chasm/lib/activity/activity_tasks.go +++ b/chasm/lib/activity/activity_tasks.go @@ -4,11 +4,18 @@ import ( "context" enumspb "go.temporal.io/api/enums/v1" + workerpb "go.temporal.io/api/worker/v1" "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" + "go.temporal.io/server/common/definition" + "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" + "go.temporal.io/server/common/namespace" "go.temporal.io/server/common/resource" "go.temporal.io/server/common/util" + "go.temporal.io/server/common/workercommands" + "go.temporal.io/server/service/history/configs" + "go.temporal.io/server/service/history/tasks" "go.uber.org/fx" ) @@ -277,3 +284,86 @@ func (h *heartbeatTimeoutTaskHandler) Execute( fromStatus: activity.GetStatus(), }) } + +// cancelCommandDispatchTaskHandler dispatches a cancel command to the worker via the Nexus +// worker commands control queue. This is a best-effort mechanism — the activity will eventually +// time out if the worker doesn't respond. +type cancelCommandDispatchTaskHandler struct { + chasm.SideEffectTaskHandlerBase[*activitypb.CancelCommandDispatchTask] + opts cancelCommandDispatchTaskHandlerOptions +} + +type cancelCommandDispatchTaskHandlerOptions struct { + fx.In + + MatchingClient resource.MatchingClient + NamespaceRegistry namespace.Registry + Config *configs.Config + MetricsHandler metrics.Handler + Logger log.Logger +} + +func newCancelCommandDispatchTaskHandler(opts cancelCommandDispatchTaskHandlerOptions) *cancelCommandDispatchTaskHandler { + return &cancelCommandDispatchTaskHandler{opts: opts} +} + +func (h *cancelCommandDispatchTaskHandler) Validate( + _ chasm.Context, + activity *Activity, + _ chasm.TaskAttributes, + _ *activitypb.CancelCommandDispatchTask, +) (bool, error) { + // Valid if the activity is in a state where it has been requested to cancel or terminated + // (meaning it was running on a worker when the cancel/terminate was issued). + return activity.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED || + activity.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_TERMINATED, nil +} + +func (h *cancelCommandDispatchTaskHandler) Execute( + ctx context.Context, + activityRef chasm.ComponentRef, + taskAttrs chasm.TaskAttributes, + _ *activitypb.CancelCommandDispatchTask, +) error { + // Read the activity to build the task token for the cancel command. + taskToken, err := chasm.ReadComponent( + ctx, + activityRef, + (*Activity).buildCancelCommandTaskToken, + activityRef, + ) + if err != nil { + return err + } + + nsEntry, err := h.opts.NamespaceRegistry.GetNamespaceByID(namespace.ID(activityRef.NamespaceID)) + if err != nil { + return err + } + + command := &workerpb.WorkerCommand{ + Type: &workerpb.WorkerCommand_CancelActivity{ + CancelActivity: &workerpb.CancelActivityCommand{ + TaskToken: taskToken, + }, + }, + } + + task := &tasks.WorkerCommandsTask{ + WorkflowKey: definition.NewWorkflowKey(activityRef.NamespaceID, "", ""), + Commands: []*workerpb.WorkerCommand{command}, + Destination: taskAttrs.Destination, + } + + dispatcher := workercommands.NewDispatcher( + h.opts.MatchingClient, + h.opts.Config, + h.opts.MetricsHandler, + h.opts.Logger, + ) + + // TODO: CHASM's SideEffectTaskHandler interface doesn't expose an attempt count. The + // dispatcher's max attempts check is effectively bypassed here. We need to either expose + // attempt count in the CHASM task interface or handle retry limiting differently. + return dispatcher.Execute(ctx, task, 1, nsEntry.Name().String()) +} diff --git a/chasm/lib/activity/config.go b/chasm/lib/activity/config.go index b21d7df65d..6adae8a595 100644 --- a/chasm/lib/activity/config.go +++ b/chasm/lib/activity/config.go @@ -42,34 +42,36 @@ var ( ) type Config struct { - BlobSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter - BlobSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter - BreakdownMetricsByTaskQueue dynamicconfig.TypedPropertyFnWithTaskQueueFilter[bool] - EnableCallbacks dynamicconfig.BoolPropertyFnWithNamespaceFilter - Enabled dynamicconfig.BoolPropertyFnWithNamespaceFilter - LongPollBuffer dynamicconfig.DurationPropertyFnWithNamespaceFilter - LongPollTimeout dynamicconfig.DurationPropertyFnWithNamespaceFilter - MaxIDLengthLimit dynamicconfig.IntPropertyFn - MaxCallbacksPerExecution dynamicconfig.IntPropertyFnWithNamespaceFilter - DefaultActivityRetryPolicy dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings] - StartDelayEnabled dynamicconfig.BoolPropertyFnWithNamespaceFilter - VisibilityMaxPageSize dynamicconfig.IntPropertyFnWithNamespaceFilter + BlobSizeLimitError dynamicconfig.IntPropertyFnWithNamespaceFilter + BlobSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter + BreakdownMetricsByTaskQueue dynamicconfig.TypedPropertyFnWithTaskQueueFilter[bool] + EnableCallbacks dynamicconfig.BoolPropertyFnWithNamespaceFilter + EnableCancelActivityWorkerCommand dynamicconfig.BoolPropertyFnWithNamespaceFilter + Enabled dynamicconfig.BoolPropertyFnWithNamespaceFilter + LongPollBuffer dynamicconfig.DurationPropertyFnWithNamespaceFilter + LongPollTimeout dynamicconfig.DurationPropertyFnWithNamespaceFilter + MaxIDLengthLimit dynamicconfig.IntPropertyFn + MaxCallbacksPerExecution dynamicconfig.IntPropertyFnWithNamespaceFilter + DefaultActivityRetryPolicy dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings] + StartDelayEnabled dynamicconfig.BoolPropertyFnWithNamespaceFilter + VisibilityMaxPageSize dynamicconfig.IntPropertyFnWithNamespaceFilter } func ConfigProvider(dc *dynamicconfig.Collection) *Config { return &Config{ - BlobSizeLimitError: dynamicconfig.BlobSizeLimitError.Get(dc), - BlobSizeLimitWarn: dynamicconfig.BlobSizeLimitWarn.Get(dc), - BreakdownMetricsByTaskQueue: dynamicconfig.MetricsBreakdownByTaskQueue.Get(dc), - DefaultActivityRetryPolicy: dynamicconfig.DefaultActivityRetryPolicy.Get(dc), - EnableCallbacks: EnableCallbacks.Get(dc), - Enabled: Enabled.Get(dc), - LongPollBuffer: LongPollBuffer.Get(dc), - LongPollTimeout: LongPollTimeout.Get(dc), - MaxIDLengthLimit: dynamicconfig.MaxIDLengthLimit.Get(dc), - StartDelayEnabled: StartDelayEnabled.Get(dc), - MaxCallbacksPerExecution: callback.MaxPerExecution.Get(dc), - VisibilityMaxPageSize: dynamicconfig.FrontendVisibilityMaxPageSize.Get(dc), + BlobSizeLimitError: dynamicconfig.BlobSizeLimitError.Get(dc), + BlobSizeLimitWarn: dynamicconfig.BlobSizeLimitWarn.Get(dc), + BreakdownMetricsByTaskQueue: dynamicconfig.MetricsBreakdownByTaskQueue.Get(dc), + DefaultActivityRetryPolicy: dynamicconfig.DefaultActivityRetryPolicy.Get(dc), + EnableCallbacks: EnableCallbacks.Get(dc), + EnableCancelActivityWorkerCommand: dynamicconfig.EnableCancelActivityWorkerCommand.Get(dc), + Enabled: Enabled.Get(dc), + LongPollBuffer: LongPollBuffer.Get(dc), + LongPollTimeout: LongPollTimeout.Get(dc), + MaxIDLengthLimit: dynamicconfig.MaxIDLengthLimit.Get(dc), + StartDelayEnabled: StartDelayEnabled.Get(dc), + MaxCallbacksPerExecution: callback.MaxPerExecution.Get(dc), + VisibilityMaxPageSize: dynamicconfig.FrontendVisibilityMaxPageSize.Get(dc), } } diff --git a/chasm/lib/activity/fx.go b/chasm/lib/activity/fx.go index f786cb674b..4516f5bb2e 100644 --- a/chasm/lib/activity/fx.go +++ b/chasm/lib/activity/fx.go @@ -13,6 +13,7 @@ var HistoryModule = fx.Module( ConfigProvider, linkValidatorProvider, newActivityDispatchTaskHandler, + newCancelCommandDispatchTaskHandler, newScheduleToStartTimeoutTaskHandler, newScheduleToCloseTimeoutTaskHandler, newStartToCloseTimeoutTaskHandler, diff --git a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go index 09c1cc3d71..b5a6c8312e 100644 --- a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go +++ b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go @@ -462,9 +462,15 @@ type ActivityAttemptState struct { SdkName string `protobuf:"bytes,10,opt,name=sdk_name,json=sdkName,proto3" json:"sdk_name,omitempty"` // The version of the SDK of the worker that most recently picked up an attempt of this activity (from the gRPC // `client-version` header on PollActivityTaskQueue). Same overwrite semantics as sdk_name. - SdkVersion string `protobuf:"bytes,11,opt,name=sdk_version,json=sdkVersion,proto3" json:"sdk_version,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + SdkVersion string `protobuf:"bytes,11,opt,name=sdk_version,json=sdkVersion,proto3" json:"sdk_version,omitempty"` + // The worker's control task queue for sending commands (e.g. cancel) via Nexus. + // Set when the worker reports it during poll. Empty if the worker doesn't support worker commands. + WorkerControlTaskQueue string `protobuf:"bytes,12,opt,name=worker_control_task_queue,json=workerControlTaskQueue,proto3" json:"worker_control_task_queue,omitempty"` + // The serialized ComponentRef captured at start time, used to construct cancel command + // task tokens that are byte-identical to poll tokens. + StartedComponentRef []byte `protobuf:"bytes,13,opt,name=started_component_ref,json=startedComponentRef,proto3" json:"started_component_ref,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ActivityAttemptState) Reset() { @@ -574,6 +580,20 @@ func (x *ActivityAttemptState) GetSdkVersion() string { return "" } +func (x *ActivityAttemptState) GetWorkerControlTaskQueue() string { + if x != nil { + return x.WorkerControlTaskQueue + } + return "" +} + +func (x *ActivityAttemptState) GetStartedComponentRef() []byte { + if x != nil { + return x.StartedComponentRef + } + return nil +} + type ActivityHeartbeatState struct { state protoimpl.MessageState `protogen:"open.v1"` // Details provided in the last recorded activity heartbeat. @@ -960,7 +980,7 @@ const file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawD "\x06reason\x18\x04 \x01(\tR\x06reason\"7\n" + "\x16ActivityTerminateState\x12\x1d\n" + "\n" + - "request_id\x18\x01 \x01(\tR\trequestId\"\xa4\x06\n" + + "request_id\x18\x01 \x01(\tR\trequestId\"\x93\a\n" + "\x14ActivityAttemptState\x12\x14\n" + "\x05count\x18\x01 \x01(\x05R\x05count\x12O\n" + "\x16current_retry_interval\x18\x02 \x01(\v2\x19.google.protobuf.DurationR\x14currentRetryInterval\x12=\n" + @@ -974,7 +994,9 @@ const file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawD "\bsdk_name\x18\n" + " \x01(\tR\asdkName\x12\x1f\n" + "\vsdk_version\x18\v \x01(\tR\n" + - "sdkVersion\x1a\x80\x01\n" + + "sdkVersion\x129\n" + + "\x19worker_control_task_queue\x18\f \x01(\tR\x16workerControlTaskQueue\x122\n" + + "\x15started_component_ref\x18\r \x01(\fR\x13startedComponentRef\x1a\x80\x01\n" + "\x12LastFailureDetails\x12.\n" + "\x04time\x18\x01 \x01(\v2\x1a.google.protobuf.TimestampR\x04time\x12:\n" + "\afailure\x18\x02 \x01(\v2 .temporal.api.failure.v1.FailureR\afailure\"\xc9\x01\n" + diff --git a/chasm/lib/activity/gen/activitypb/v1/tasks.go-helpers.pb.go b/chasm/lib/activity/gen/activitypb/v1/tasks.go-helpers.pb.go index d7628a6e9e..a4173d9659 100644 --- a/chasm/lib/activity/gen/activitypb/v1/tasks.go-helpers.pb.go +++ b/chasm/lib/activity/gen/activitypb/v1/tasks.go-helpers.pb.go @@ -189,3 +189,40 @@ func (this *HeartbeatTimeoutTask) Equal(that interface{}) bool { return proto.Equal(this, that1) } + +// Marshal an object of type CancelCommandDispatchTask to the protobuf v3 wire format +func (val *CancelCommandDispatchTask) Marshal() ([]byte, error) { + return proto.Marshal(val) +} + +// Unmarshal an object of type CancelCommandDispatchTask from the protobuf v3 wire format +func (val *CancelCommandDispatchTask) Unmarshal(buf []byte) error { + return proto.Unmarshal(buf, val) +} + +// Size returns the size of the object, in bytes, once serialized +func (val *CancelCommandDispatchTask) Size() int { + return proto.Size(val) +} + +// Equal returns whether two CancelCommandDispatchTask values are equivalent by recursively +// comparing the message's fields. +// For more information see the documentation for +// https://pkg.go.dev/google.golang.org/protobuf/proto#Equal +func (this *CancelCommandDispatchTask) Equal(that interface{}) bool { + if that == nil { + return this == nil + } + + var that1 *CancelCommandDispatchTask + switch t := that.(type) { + case *CancelCommandDispatchTask: + that1 = t + case CancelCommandDispatchTask: + that1 = &t + default: + return false + } + + return proto.Equal(this, that1) +} diff --git a/chasm/lib/activity/gen/activitypb/v1/tasks.pb.go b/chasm/lib/activity/gen/activitypb/v1/tasks.pb.go index 796574e7db..23fc96a8db 100644 --- a/chasm/lib/activity/gen/activitypb/v1/tasks.pb.go +++ b/chasm/lib/activity/gen/activitypb/v1/tasks.pb.go @@ -239,6 +239,44 @@ func (x *HeartbeatTimeoutTask) GetStamp() int32 { return 0 } +// CancelCommandDispatchTask is a side-effect task that dispatches a cancel command to the worker +// via the Nexus worker commands control queue. +type CancelCommandDispatchTask struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CancelCommandDispatchTask) Reset() { + *x = CancelCommandDispatchTask{} + mi := &file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CancelCommandDispatchTask) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CancelCommandDispatchTask) ProtoMessage() {} + +func (x *CancelCommandDispatchTask) ProtoReflect() protoreflect.Message { + mi := &file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use CancelCommandDispatchTask.ProtoReflect.Descriptor instead. +func (*CancelCommandDispatchTask) Descriptor() ([]byte, []int) { + return file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_rawDescGZIP(), []int{5} +} + var File_temporal_server_chasm_lib_activity_proto_v1_tasks_proto protoreflect.FileDescriptor const file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_rawDesc = "" + @@ -252,7 +290,8 @@ const file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_rawDesc = "" "\x17StartToCloseTimeoutTask\x12\x14\n" + "\x05stamp\x18\x01 \x01(\x05R\x05stamp\",\n" + "\x14HeartbeatTimeoutTask\x12\x14\n" + - "\x05stamp\x18\x01 \x01(\x05R\x05stampBDZBgo.temporal.io/server/chasm/lib/activity/gen/activitypb;activitypbb\x06proto3" + "\x05stamp\x18\x01 \x01(\x05R\x05stamp\"\x1b\n" + + "\x19CancelCommandDispatchTaskBDZBgo.temporal.io/server/chasm/lib/activity/gen/activitypb;activitypbb\x06proto3" var ( file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_rawDescOnce sync.Once @@ -266,13 +305,14 @@ func file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_rawDescGZIP() return file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_rawDescData } -var file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_msgTypes = make([]protoimpl.MessageInfo, 5) +var file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_msgTypes = make([]protoimpl.MessageInfo, 6) var file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_goTypes = []any{ (*ActivityDispatchTask)(nil), // 0: temporal.server.chasm.lib.activity.proto.v1.ActivityDispatchTask (*ScheduleToStartTimeoutTask)(nil), // 1: temporal.server.chasm.lib.activity.proto.v1.ScheduleToStartTimeoutTask (*ScheduleToCloseTimeoutTask)(nil), // 2: temporal.server.chasm.lib.activity.proto.v1.ScheduleToCloseTimeoutTask (*StartToCloseTimeoutTask)(nil), // 3: temporal.server.chasm.lib.activity.proto.v1.StartToCloseTimeoutTask (*HeartbeatTimeoutTask)(nil), // 4: temporal.server.chasm.lib.activity.proto.v1.HeartbeatTimeoutTask + (*CancelCommandDispatchTask)(nil), // 5: temporal.server.chasm.lib.activity.proto.v1.CancelCommandDispatchTask } var file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_depIdxs = []int32{ 0, // [0:0] is the sub-list for method output_type @@ -293,7 +333,7 @@ func file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_activity_proto_v1_tasks_proto_rawDesc)), NumEnums: 0, - NumMessages: 5, + NumMessages: 6, NumExtensions: 0, NumServices: 0, }, diff --git a/chasm/lib/activity/library.go b/chasm/lib/activity/library.go index 8f939df2fc..ac514e00a8 100644 --- a/chasm/lib/activity/library.go +++ b/chasm/lib/activity/library.go @@ -87,6 +87,7 @@ type library struct { handler *handler activityDispatchTaskHandler *activityDispatchTaskHandler + cancelCommandDispatchTaskHandler *cancelCommandDispatchTaskHandler scheduleToStartTimeoutTaskHandler *scheduleToStartTimeoutTaskHandler scheduleToCloseTimeoutTaskHandler *scheduleToCloseTimeoutTaskHandler startToCloseTimeoutTaskHandler *startToCloseTimeoutTaskHandler @@ -96,6 +97,7 @@ type library struct { func newLibrary( handler *handler, activityDispatchTaskHandler *activityDispatchTaskHandler, + cancelCommandDispatchTaskHandler *cancelCommandDispatchTaskHandler, scheduleToStartTimeoutTaskHandler *scheduleToStartTimeoutTaskHandler, scheduleToCloseTimeoutTaskHandler *scheduleToCloseTimeoutTaskHandler, startToCloseTimeoutTaskHandler *startToCloseTimeoutTaskHandler, @@ -107,6 +109,7 @@ func newLibrary( componentOnlyLibrary: *newComponentOnlyLibrary(config, namespaceRegistry), handler: handler, activityDispatchTaskHandler: activityDispatchTaskHandler, + cancelCommandDispatchTaskHandler: cancelCommandDispatchTaskHandler, scheduleToStartTimeoutTaskHandler: scheduleToStartTimeoutTaskHandler, scheduleToCloseTimeoutTaskHandler: scheduleToCloseTimeoutTaskHandler, startToCloseTimeoutTaskHandler: startToCloseTimeoutTaskHandler, @@ -140,5 +143,9 @@ func (l *library) Tasks() []*chasm.RegistrableTask { "heartbeatTimer", l.heartbeatTimeoutTaskHandler, ), + chasm.NewRegistrableSideEffectTask( + "cancelCommandDispatch", + l.cancelCommandDispatchTaskHandler, + ), } } diff --git a/chasm/lib/activity/proto/v1/activity_state.proto b/chasm/lib/activity/proto/v1/activity_state.proto index 2646540203..b3cdb98153 100644 --- a/chasm/lib/activity/proto/v1/activity_state.proto +++ b/chasm/lib/activity/proto/v1/activity_state.proto @@ -164,6 +164,14 @@ message ActivityAttemptState { // The version of the SDK of the worker that most recently picked up an attempt of this activity (from the gRPC // `client-version` header on PollActivityTaskQueue). Same overwrite semantics as sdk_name. string sdk_version = 11; + + // The worker's control task queue for sending commands (e.g. cancel) via Nexus. + // Set when the worker reports it during poll. Empty if the worker doesn't support worker commands. + string worker_control_task_queue = 12; + + // The serialized ComponentRef captured at start time, used to construct cancel command + // task tokens that are byte-identical to poll tokens. + bytes started_component_ref = 13; } message ActivityHeartbeatState { diff --git a/chasm/lib/activity/proto/v1/tasks.proto b/chasm/lib/activity/proto/v1/tasks.proto index 9a1996e3dd..70dd3ea992 100644 --- a/chasm/lib/activity/proto/v1/tasks.proto +++ b/chasm/lib/activity/proto/v1/tasks.proto @@ -26,3 +26,7 @@ message HeartbeatTimeoutTask { // The current stamp for this activity execution. Used for task validation. See also [ActivityAttemptState]. int32 stamp = 1; } + +// CancelCommandDispatchTask is a side-effect task that dispatches a cancel command to the worker +// via the Nexus worker commands control queue. +message CancelCommandDispatchTask {} diff --git a/chasm/lib/activity/statemachine.go b/chasm/lib/activity/statemachine.go index c1c19e8248..3d9d01abc5 100644 --- a/chasm/lib/activity/statemachine.go +++ b/chasm/lib/activity/statemachine.go @@ -144,11 +144,21 @@ var TransitionStarted = chasm.NewTransition( activitypb.ACTIVITY_EXECUTION_STATUS_STARTED, func(a *Activity, ctx chasm.MutableContext, request *historyservice.RecordActivityTaskStartedRequest) error { attempt := a.LastAttempt.Get(ctx) + + // Capture the ComponentRef at start time for constructing cancel command task tokens + // that are byte-identical to poll tokens. + startedRef, err := ctx.Ref(a) + if err != nil { + return err + } + attempt.StartedComponentRef = startedRef + attempt.StartedTime = timestamppb.New(ctx.Now(a)) attempt.StartRequestId = request.GetRequestId() attempt.LastWorkerIdentity = request.GetPollRequest().GetIdentity() attempt.SdkName = ctx.RequestHeader(headers.ClientNameHeaderName) attempt.SdkVersion = ctx.RequestHeader(headers.ClientVersionHeaderName) + attempt.WorkerControlTaskQueue = request.GetPollRequest().GetWorkerControlTaskQueue() if versionDirective := request.GetVersionDirective().GetDeploymentVersion(); versionDirective != nil { attempt.LastDeploymentVersion = &deploymentpb.WorkerDeploymentVersion{ BuildId: versionDirective.GetBuildId(), diff --git a/common/tasktoken/token.go b/common/tasktoken/token.go index 4212b4f93c..ad03754d35 100644 --- a/common/tasktoken/token.go +++ b/common/tasktoken/token.go @@ -11,7 +11,7 @@ func NewWorkflowTaskToken( workflowID string, runID string, scheduledEventID int64, - startedEventId int64, + startedEventID int64, startedTime *timestamppb.Timestamp, attempt int32, clock *clockspb.VectorClock, @@ -22,7 +22,7 @@ func NewWorkflowTaskToken( WorkflowId: workflowID, RunId: runID, ScheduledEventId: scheduledEventID, - StartedEventId: startedEventId, + StartedEventId: startedEventID, StartedTime: startedTime, Attempt: attempt, Clock: clock, @@ -30,12 +30,34 @@ func NewWorkflowTaskToken( } } +// NewStandaloneActivityTaskToken builds a task token for a standalone activity. +// Standalone activities don't use ScheduledEventId, Clock, Version, or StartVersion. +func NewStandaloneActivityTaskToken( + namespaceID string, + workflowID string, + runID string, + activityID string, + activityType string, + attempt int32, + componentRef []byte, +) *tokenspb.Task { + return NewActivityTaskToken( + namespaceID, workflowID, runID, + 0, // scheduledEventId + activityID, activityType, attempt, + nil, // clock + 0, // version + 0, // startVersion + componentRef, + ) +} + func NewActivityTaskToken( namespaceID string, workflowID string, runID string, scheduledEventID int64, - activityId string, + activityID string, activityType string, attempt int32, clock *clockspb.VectorClock, @@ -50,7 +72,7 @@ func NewActivityTaskToken( ScheduledEventId: scheduledEventID, ActivityType: activityType, Attempt: attempt, - ActivityId: activityId, + ActivityId: activityID, Clock: clock, Version: version, StartVersion: startVersion, diff --git a/common/tasktoken/token_test.go b/common/tasktoken/token_test.go new file mode 100644 index 0000000000..626a48dd34 --- /dev/null +++ b/common/tasktoken/token_test.go @@ -0,0 +1,50 @@ +package tasktoken + +import ( + "testing" + + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +// TestStandaloneActivityTokenMatchesPollToken verifies that NewStandaloneActivityTaskToken +// produces a token byte-identical to what matching builds via NewActivityTaskToken during poll. +// Matching uses NewActivityTaskToken with scheduledEventId=0, clock=nil, version=0, startVersion=0 +// for standalone activities (see matching_engine.go createPollActivityTaskQueueResponse). +// If NewActivityTaskToken gains new fields, this test will fail, reminding us to update +// NewStandaloneActivityTaskToken to match. +func TestStandaloneActivityTokenMatchesPollToken(t *testing.T) { + namespaceID := "ns-id" + workflowID := "wf-id" + runID := "run-id" + activityID := "act-id" + activityType := "MyActivity" + attempt := int32(3) + componentRef := []byte("some-component-ref") + + // This is what matching builds for a standalone activity poll response. + pollToken := NewActivityTaskToken( + namespaceID, workflowID, runID, + 0, // scheduledEventId — always 0 for standalone + activityID, activityType, attempt, + nil, // clock — always nil for standalone + 0, // version — always 0 for standalone + 0, // startVersion — always 0 for standalone + componentRef, + ) + + // This is what the cancel command handler builds. + cancelToken := NewStandaloneActivityTaskToken( + namespaceID, workflowID, runID, + activityID, activityType, attempt, + componentRef, + ) + + pollBytes, err := proto.Marshal(pollToken) + require.NoError(t, err) + cancelBytes, err := proto.Marshal(cancelToken) + require.NoError(t, err) + + require.Equal(t, pollBytes, cancelBytes, + "cancel command token must be byte-identical to poll token") +} diff --git a/tests/activity_standalone_test.go b/tests/activity_standalone_test.go index 7d89302867..e43f8897e3 100644 --- a/tests/activity_standalone_test.go +++ b/tests/activity_standalone_test.go @@ -15,6 +15,7 @@ import ( commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" failurepb "go.temporal.io/api/failure/v1" + workerservicepb "go.temporal.io/api/nexusservices/workerservice/v1" "go.temporal.io/api/operatorservice/v1" sdkpb "go.temporal.io/api/sdk/v1" "go.temporal.io/api/serviceerror" @@ -32,11 +33,13 @@ import ( "go.temporal.io/server/common/payloads" "go.temporal.io/server/common/searchattribute/sadefs" "go.temporal.io/server/common/tasktoken" + "go.temporal.io/server/common/testing/await" "go.temporal.io/server/common/testing/parallelsuite" "go.temporal.io/server/common/testing/protorequire" "go.temporal.io/server/tests/testcore" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" + "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -2616,6 +2619,90 @@ func (s *standaloneActivityTestSuite) TestRequestCancel() { }) } +// TestDispatchCancelCommandToWorker tests that when a standalone activity is cancelled, +// the server dispatches a cancel command to the worker's control queue via Nexus. +func (s *standaloneActivityTestSuite) TestDispatchCancelCommandToWorker() { + env := s.newTestEnv() + t := s.T() + ctx := s.Context() + + // Enable cancel command dispatch. Set globally (unconstrained) because the CHASM task handler + // resolves the feature flag using namespace ID, not namespace name. + env.GetTestCluster().OverrideDynamicConfig( + t, dynamicconfig.EnableCancelActivityWorkerCommand, + true, + ) + + activityID := testcore.RandomizeStr(t.Name()) + taskQueue := testcore.RandomizeStr(t.Name()) + tv := env.Tv() + controlQueueName := tv.ControlQueueName(env.Namespace().String()) + + // Start standalone activity. + startResp := env.startAndValidateActivity(ctx, t, activityID, taskQueue) + runID := startResp.RunId + + // Poll for the activity task with WorkerInstanceKey + WorkerControlTaskQueue so the server + // knows this worker supports cancel commands. + activityPollResp, err := env.FrontendClient().PollActivityTaskQueue(ctx, &workflowservice.PollActivityTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{ + Name: taskQueue, + Kind: enumspb.TASK_QUEUE_KIND_NORMAL, + }, + Identity: tv.WorkerIdentity(), + WorkerInstanceKey: tv.WorkerInstanceKey(), + WorkerControlTaskQueue: controlQueueName, + }) + require.NoError(t, err) + require.Equal(t, activityID, activityPollResp.GetActivityId()) + require.NotEmpty(t, activityPollResp.TaskToken) + + // Request cancellation of the standalone activity. + _, err = env.FrontendClient().RequestCancelActivityExecution(ctx, &workflowservice.RequestCancelActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: runID, + Identity: "cancelling-client", + Reason: "test cancel command dispatch", + }) + require.NoError(t, err) + + // Poll the Nexus control queue — should receive the cancel command. + var nexusPollResp *workflowservice.PollNexusTaskQueueResponse + await.RequireTrue(t, func() bool { + pollCtx, pollCancel := context.WithTimeout(ctx, 5*time.Second) + defer pollCancel() + resp, err := env.FrontendClient().PollNexusTaskQueue(pollCtx, &workflowservice.PollNexusTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: controlQueueName, Kind: enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS}, + Identity: tv.WorkerIdentity(), + }) + if err == nil && resp != nil && resp.Request != nil { + nexusPollResp = resp + return true + } + return false + }, 30*time.Second, 200*time.Millisecond) + + // Verify the Nexus request contains an ExecuteCommands operation with a CancelActivity command. + startOp := nexusPollResp.Request.GetStartOperation() + require.NotNil(t, startOp, "expected StartOperation in Nexus request") + require.Equal(t, "temporal.api.nexusservices.workerservice.v1.WorkerService", startOp.Service) + require.Equal(t, "ExecuteCommands", startOp.Operation) + + require.NotNil(t, startOp.Payload) + var executeReq workerservicepb.ExecuteCommandsRequest + require.NoError(t, proto.Unmarshal(startOp.Payload.Data, &executeReq)) + require.Len(t, executeReq.Commands, 1, "expected exactly 1 command") + cancelCmd := executeReq.Commands[0].GetCancelActivity() + require.NotNil(t, cancelCmd, "expected CancelActivity command") + + // The cancel command token must be byte-identical to the poll token because the SDK + // uses exact byte match to look up the activity to cancel. + require.Equal(t, activityPollResp.TaskToken, cancelCmd.TaskToken) +} + func (s *standaloneActivityTestSuite) TestTerminate() { env := s.newTestEnv() t := s.T() From d0e8eb07049ec6436ba134c3df700b10fd5b2de9 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Thu, 25 Jun 2026 19:34:43 -0700 Subject: [PATCH 05/22] Remove redundant token sync unit test The functional test TestDispatchCancelCommandToWorker already validates that the cancel command token matches the poll token end-to-end. Co-Authored-By: Claude Opus 4.6 --- common/tasktoken/token_test.go | 50 ---------------------------------- 1 file changed, 50 deletions(-) delete mode 100644 common/tasktoken/token_test.go diff --git a/common/tasktoken/token_test.go b/common/tasktoken/token_test.go deleted file mode 100644 index 626a48dd34..0000000000 --- a/common/tasktoken/token_test.go +++ /dev/null @@ -1,50 +0,0 @@ -package tasktoken - -import ( - "testing" - - "github.com/stretchr/testify/require" - "google.golang.org/protobuf/proto" -) - -// TestStandaloneActivityTokenMatchesPollToken verifies that NewStandaloneActivityTaskToken -// produces a token byte-identical to what matching builds via NewActivityTaskToken during poll. -// Matching uses NewActivityTaskToken with scheduledEventId=0, clock=nil, version=0, startVersion=0 -// for standalone activities (see matching_engine.go createPollActivityTaskQueueResponse). -// If NewActivityTaskToken gains new fields, this test will fail, reminding us to update -// NewStandaloneActivityTaskToken to match. -func TestStandaloneActivityTokenMatchesPollToken(t *testing.T) { - namespaceID := "ns-id" - workflowID := "wf-id" - runID := "run-id" - activityID := "act-id" - activityType := "MyActivity" - attempt := int32(3) - componentRef := []byte("some-component-ref") - - // This is what matching builds for a standalone activity poll response. - pollToken := NewActivityTaskToken( - namespaceID, workflowID, runID, - 0, // scheduledEventId — always 0 for standalone - activityID, activityType, attempt, - nil, // clock — always nil for standalone - 0, // version — always 0 for standalone - 0, // startVersion — always 0 for standalone - componentRef, - ) - - // This is what the cancel command handler builds. - cancelToken := NewStandaloneActivityTaskToken( - namespaceID, workflowID, runID, - activityID, activityType, attempt, - componentRef, - ) - - pollBytes, err := proto.Marshal(pollToken) - require.NoError(t, err) - cancelBytes, err := proto.Marshal(cancelToken) - require.NoError(t, err) - - require.Equal(t, pollBytes, cancelBytes, - "cancel command token must be byte-identical to poll token") -} From d16e3d6c5131695a3ef577e7ed8501637a8cc8aa Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Fri, 26 Jun 2026 11:57:11 -0700 Subject: [PATCH 06/22] Move cancel command handler to worker_command_task_handlers.go Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/activity_tasks.go | 89 --------------- .../activity/worker_command_task_handlers.go | 101 ++++++++++++++++++ 2 files changed, 101 insertions(+), 89 deletions(-) create mode 100644 chasm/lib/activity/worker_command_task_handlers.go diff --git a/chasm/lib/activity/activity_tasks.go b/chasm/lib/activity/activity_tasks.go index 089dcf9f2b..20d8a46d43 100644 --- a/chasm/lib/activity/activity_tasks.go +++ b/chasm/lib/activity/activity_tasks.go @@ -4,18 +4,11 @@ import ( "context" enumspb "go.temporal.io/api/enums/v1" - workerpb "go.temporal.io/api/worker/v1" "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" - "go.temporal.io/server/common/definition" - "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" - "go.temporal.io/server/common/namespace" "go.temporal.io/server/common/resource" "go.temporal.io/server/common/util" - "go.temporal.io/server/common/workercommands" - "go.temporal.io/server/service/history/configs" - "go.temporal.io/server/service/history/tasks" "go.uber.org/fx" ) @@ -285,85 +278,3 @@ func (h *heartbeatTimeoutTaskHandler) Execute( }) } -// cancelCommandDispatchTaskHandler dispatches a cancel command to the worker via the Nexus -// worker commands control queue. This is a best-effort mechanism — the activity will eventually -// time out if the worker doesn't respond. -type cancelCommandDispatchTaskHandler struct { - chasm.SideEffectTaskHandlerBase[*activitypb.CancelCommandDispatchTask] - opts cancelCommandDispatchTaskHandlerOptions -} - -type cancelCommandDispatchTaskHandlerOptions struct { - fx.In - - MatchingClient resource.MatchingClient - NamespaceRegistry namespace.Registry - Config *configs.Config - MetricsHandler metrics.Handler - Logger log.Logger -} - -func newCancelCommandDispatchTaskHandler(opts cancelCommandDispatchTaskHandlerOptions) *cancelCommandDispatchTaskHandler { - return &cancelCommandDispatchTaskHandler{opts: opts} -} - -func (h *cancelCommandDispatchTaskHandler) Validate( - _ chasm.Context, - activity *Activity, - _ chasm.TaskAttributes, - _ *activitypb.CancelCommandDispatchTask, -) (bool, error) { - // Valid if the activity is in a state where it has been requested to cancel or terminated - // (meaning it was running on a worker when the cancel/terminate was issued). - return activity.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED || - activity.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_TERMINATED, nil -} - -func (h *cancelCommandDispatchTaskHandler) Execute( - ctx context.Context, - activityRef chasm.ComponentRef, - taskAttrs chasm.TaskAttributes, - _ *activitypb.CancelCommandDispatchTask, -) error { - // Read the activity to build the task token for the cancel command. - taskToken, err := chasm.ReadComponent( - ctx, - activityRef, - (*Activity).buildCancelCommandTaskToken, - activityRef, - ) - if err != nil { - return err - } - - nsEntry, err := h.opts.NamespaceRegistry.GetNamespaceByID(namespace.ID(activityRef.NamespaceID)) - if err != nil { - return err - } - - command := &workerpb.WorkerCommand{ - Type: &workerpb.WorkerCommand_CancelActivity{ - CancelActivity: &workerpb.CancelActivityCommand{ - TaskToken: taskToken, - }, - }, - } - - task := &tasks.WorkerCommandsTask{ - WorkflowKey: definition.NewWorkflowKey(activityRef.NamespaceID, "", ""), - Commands: []*workerpb.WorkerCommand{command}, - Destination: taskAttrs.Destination, - } - - dispatcher := workercommands.NewDispatcher( - h.opts.MatchingClient, - h.opts.Config, - h.opts.MetricsHandler, - h.opts.Logger, - ) - - // TODO: CHASM's SideEffectTaskHandler interface doesn't expose an attempt count. The - // dispatcher's max attempts check is effectively bypassed here. We need to either expose - // attempt count in the CHASM task interface or handle retry limiting differently. - return dispatcher.Execute(ctx, task, 1, nsEntry.Name().String()) -} diff --git a/chasm/lib/activity/worker_command_task_handlers.go b/chasm/lib/activity/worker_command_task_handlers.go new file mode 100644 index 0000000000..6d7aa482f6 --- /dev/null +++ b/chasm/lib/activity/worker_command_task_handlers.go @@ -0,0 +1,101 @@ +package activity + +import ( + "context" + + workerpb "go.temporal.io/api/worker/v1" + "go.temporal.io/server/chasm" + "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" + "go.temporal.io/server/common/definition" + "go.temporal.io/server/common/log" + "go.temporal.io/server/common/metrics" + "go.temporal.io/server/common/namespace" + "go.temporal.io/server/common/resource" + "go.temporal.io/server/common/workercommands" + "go.temporal.io/server/service/history/configs" + "go.temporal.io/server/service/history/tasks" + "go.uber.org/fx" +) + +// cancelCommandDispatchTaskHandler dispatches a cancel command to the worker via the Nexus +// worker commands control queue. This is a best-effort mechanism — the activity will eventually +// time out if the worker doesn't respond. +type cancelCommandDispatchTaskHandler struct { + chasm.SideEffectTaskHandlerBase[*activitypb.CancelCommandDispatchTask] + opts cancelCommandDispatchTaskHandlerOptions +} + +type cancelCommandDispatchTaskHandlerOptions struct { + fx.In + + MatchingClient resource.MatchingClient + NamespaceRegistry namespace.Registry + Config *configs.Config + MetricsHandler metrics.Handler + Logger log.Logger +} + +func newCancelCommandDispatchTaskHandler(opts cancelCommandDispatchTaskHandlerOptions) *cancelCommandDispatchTaskHandler { + return &cancelCommandDispatchTaskHandler{opts: opts} +} + +func (h *cancelCommandDispatchTaskHandler) Validate( + _ chasm.Context, + activity *Activity, + _ chasm.TaskAttributes, + _ *activitypb.CancelCommandDispatchTask, +) (bool, error) { + // Valid if the activity is in a state where it has been requested to cancel or terminated + // (meaning it was running on a worker when the cancel/terminate was issued). + return activity.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED || + activity.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_TERMINATED, nil +} + +func (h *cancelCommandDispatchTaskHandler) Execute( + ctx context.Context, + activityRef chasm.ComponentRef, + taskAttrs chasm.TaskAttributes, + _ *activitypb.CancelCommandDispatchTask, +) error { + // Read the activity to build the task token for the cancel command. + taskToken, err := chasm.ReadComponent( + ctx, + activityRef, + (*Activity).buildCancelCommandTaskToken, + activityRef, + ) + if err != nil { + return err + } + + nsEntry, err := h.opts.NamespaceRegistry.GetNamespaceByID(namespace.ID(activityRef.NamespaceID)) + if err != nil { + return err + } + + command := &workerpb.WorkerCommand{ + Type: &workerpb.WorkerCommand_CancelActivity{ + CancelActivity: &workerpb.CancelActivityCommand{ + TaskToken: taskToken, + }, + }, + } + + task := &tasks.WorkerCommandsTask{ + WorkflowKey: definition.NewWorkflowKey(activityRef.NamespaceID, "", ""), + Commands: []*workerpb.WorkerCommand{command}, + Destination: taskAttrs.Destination, + } + + dispatcher := workercommands.NewDispatcher( + h.opts.MatchingClient, + h.opts.Config, + h.opts.MetricsHandler, + h.opts.Logger, + ) + + // TODO: CHASM's SideEffectTaskHandler interface doesn't expose an attempt count. The + // dispatcher's max attempts check is effectively bypassed here. We need to either expose + // attempt count in the CHASM task interface or handle retry limiting differently. + return dispatcher.Execute(ctx, task, 1, nsEntry.Name().String()) +} From c9c7da3ca93808e4e1d0ea881b6c7f54bf12480c Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Fri, 26 Jun 2026 11:58:53 -0700 Subject: [PATCH 07/22] Clarify token comments: must match the poll response token Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/activity.go | 2 +- chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go | 5 +++-- chasm/lib/activity/proto/v1/activity_state.proto | 5 +++-- chasm/lib/activity/statemachine.go | 5 +++-- tests/activity_standalone_test.go | 3 +-- 5 files changed, 11 insertions(+), 9 deletions(-) diff --git a/chasm/lib/activity/activity.go b/chasm/lib/activity/activity.go index f231acd529..9c422a5c62 100644 --- a/chasm/lib/activity/activity.go +++ b/chasm/lib/activity/activity.go @@ -221,7 +221,7 @@ func (a *Activity) createAddActivityTaskRequest(ctx chasm.Context, namespaceID s } // buildCancelCommandTaskToken builds the serialized task token for a cancel command. -// Uses the ComponentRef captured at start time so the token is byte-identical to the poll token. +// The token must match what was sent to the worker in the poll response. func (a *Activity) buildCancelCommandTaskToken(ctx chasm.Context, activityRef chasm.ComponentRef) ([]byte, error) { attempt := a.LastAttempt.Get(ctx) key := ctx.ExecutionKey() diff --git a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go index b5a6c8312e..b1985abe08 100644 --- a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go +++ b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go @@ -466,8 +466,9 @@ type ActivityAttemptState struct { // The worker's control task queue for sending commands (e.g. cancel) via Nexus. // Set when the worker reports it during poll. Empty if the worker doesn't support worker commands. WorkerControlTaskQueue string `protobuf:"bytes,12,opt,name=worker_control_task_queue,json=workerControlTaskQueue,proto3" json:"worker_control_task_queue,omitempty"` - // The serialized ComponentRef captured at start time, used to construct cancel command - // task tokens that are byte-identical to poll tokens. + // The serialized ComponentRef captured when the task was started. Used to construct + // the task token for cancel commands. The token must match what was sent to the + // worker in the poll response. StartedComponentRef []byte `protobuf:"bytes,13,opt,name=started_component_ref,json=startedComponentRef,proto3" json:"started_component_ref,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache diff --git a/chasm/lib/activity/proto/v1/activity_state.proto b/chasm/lib/activity/proto/v1/activity_state.proto index b3cdb98153..d9e8901f7f 100644 --- a/chasm/lib/activity/proto/v1/activity_state.proto +++ b/chasm/lib/activity/proto/v1/activity_state.proto @@ -169,8 +169,9 @@ message ActivityAttemptState { // Set when the worker reports it during poll. Empty if the worker doesn't support worker commands. string worker_control_task_queue = 12; - // The serialized ComponentRef captured at start time, used to construct cancel command - // task tokens that are byte-identical to poll tokens. + // The serialized ComponentRef captured when the task was started. Used to construct + // the task token for cancel commands. The token must match what was sent to the + // worker in the poll response. bytes started_component_ref = 13; } diff --git a/chasm/lib/activity/statemachine.go b/chasm/lib/activity/statemachine.go index 3d9d01abc5..e50cd70392 100644 --- a/chasm/lib/activity/statemachine.go +++ b/chasm/lib/activity/statemachine.go @@ -145,8 +145,9 @@ var TransitionStarted = chasm.NewTransition( func(a *Activity, ctx chasm.MutableContext, request *historyservice.RecordActivityTaskStartedRequest) error { attempt := a.LastAttempt.Get(ctx) - // Capture the ComponentRef at start time for constructing cancel command task tokens - // that are byte-identical to poll tokens. + // Capture the ComponentRef at start time. Used to construct the task token + // for cancel commands. The token must match what was sent to the worker in + // the poll response. startedRef, err := ctx.Ref(a) if err != nil { return err diff --git a/tests/activity_standalone_test.go b/tests/activity_standalone_test.go index e43f8897e3..27df7a420f 100644 --- a/tests/activity_standalone_test.go +++ b/tests/activity_standalone_test.go @@ -2698,8 +2698,7 @@ func (s *standaloneActivityTestSuite) TestDispatchCancelCommandToWorker() { cancelCmd := executeReq.Commands[0].GetCancelActivity() require.NotNil(t, cancelCmd, "expected CancelActivity command") - // The cancel command token must be byte-identical to the poll token because the SDK - // uses exact byte match to look up the activity to cancel. + // The cancel command's task token must match what was sent to the worker in the poll response. require.Equal(t, activityPollResp.TaskToken, cancelCmd.TaskToken) } From 69813edfeda63cd4bcf4eb6e73f6e2598f0d6c41 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Fri, 26 Jun 2026 12:10:06 -0700 Subject: [PATCH 08/22] Remove unnecessary comments and trailing newline Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/activity.go | 2 -- chasm/lib/activity/activity_tasks.go | 1 - 2 files changed, 3 deletions(-) diff --git a/chasm/lib/activity/activity.go b/chasm/lib/activity/activity.go index 9c422a5c62..117fbe99fd 100644 --- a/chasm/lib/activity/activity.go +++ b/chasm/lib/activity/activity.go @@ -204,8 +204,6 @@ func (a *Activity) createAddActivityTaskRequest(ctx chasm.Context, namespaceID s key := ctx.ExecutionKey() - // Note: No need to set the vector clock here, as the components track version conflicts for read/write - // TODO: Need to fill in VersionDirective once we decide how to handle versioning for standalone activities return &matchingservice.AddActivityTaskRequest{ NamespaceId: namespaceID, Execution: &commonpb.WorkflowExecution{ diff --git a/chasm/lib/activity/activity_tasks.go b/chasm/lib/activity/activity_tasks.go index 20d8a46d43..e22b2f586a 100644 --- a/chasm/lib/activity/activity_tasks.go +++ b/chasm/lib/activity/activity_tasks.go @@ -277,4 +277,3 @@ func (h *heartbeatTimeoutTaskHandler) Execute( fromStatus: activity.GetStatus(), }) } - From 383a0e4921945cdbe4a20b8655c0ed3e008878fb Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Mon, 6 Jul 2026 22:18:48 -0700 Subject: [PATCH 09/22] Record cancel command dispatch state for standby cluster invalidation After successfully dispatching the cancel command to the worker, record cancel_command_dispatched=true via chasm.UpdateComponent(). This state replicates to standby clusters, allowing Validate() to return false and drop the task instead of retrying until the discard delay expires. During failover, if active never dispatched, the flag remains unset so the new active picks up and executes the task. Co-Authored-By: Claude Opus 4.6 --- .../gen/activitypb/v1/activity_state.pb.go | 22 ++++++++++++---- .../activity/proto/v1/activity_state.proto | 5 ++++ .../activity/worker_command_task_handlers.go | 25 ++++++++++++++++++- 3 files changed, 46 insertions(+), 6 deletions(-) diff --git a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go index b1985abe08..bf76195aa3 100644 --- a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go +++ b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go @@ -185,9 +185,13 @@ type ActivityState struct { TerminateState *ActivityTerminateState `protobuf:"bytes,12,opt,name=terminate_state,json=terminateState,proto3" json:"terminate_state,omitempty"` // Amount of time to wait before dispatching the activity task to the task queue for the first time. If the activity // has a retry policy, retry attempts will not have start delay applied. - StartDelay *durationpb.Duration `protobuf:"bytes,13,opt,name=start_delay,json=startDelay,proto3" json:"start_delay,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + StartDelay *durationpb.Duration `protobuf:"bytes,13,opt,name=start_delay,json=startDelay,proto3" json:"start_delay,omitempty"` + // Set to true after the cancel command has been successfully dispatched to the worker + // via the Nexus control queue. Used by standby clusters to determine whether the + // dispatch task can be safely discarded. + CancelCommandDispatched bool `protobuf:"varint,14,opt,name=cancel_command_dispatched,json=cancelCommandDispatched,proto3" json:"cancel_command_dispatched,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ActivityState) Reset() { @@ -311,6 +315,13 @@ func (x *ActivityState) GetStartDelay() *durationpb.Duration { return nil } +func (x *ActivityState) GetCancelCommandDispatched() bool { + if x != nil { + return x.CancelCommandDispatched + } + return false +} + type ActivityCancelState struct { state protoimpl.MessageState `protogen:"open.v1"` RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` @@ -955,7 +966,7 @@ var File_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto protor const file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawDesc = "" + "\n" + - "@temporal/server/chasm/lib/activity/proto/v1/activity_state.proto\x12+temporal.server.chasm.lib.activity.proto.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\x1a(temporal/api/deployment/v1/message.proto\x1a%temporal/api/failure/v1/message.proto\x1a'temporal/api/sdk/v1/user_metadata.proto\x1a'temporal/api/taskqueue/v1/message.proto\"\x97\b\n" + + "@temporal/server/chasm/lib/activity/proto/v1/activity_state.proto\x12+temporal.server.chasm.lib.activity.proto.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\x1a(temporal/api/deployment/v1/message.proto\x1a%temporal/api/failure/v1/message.proto\x1a'temporal/api/sdk/v1/user_metadata.proto\x1a'temporal/api/taskqueue/v1/message.proto\"\xd3\b\n" + "\rActivityState\x12I\n" + "\ractivity_type\x18\x01 \x01(\v2$.temporal.api.common.v1.ActivityTypeR\factivityType\x12C\n" + "\n" + @@ -972,7 +983,8 @@ const file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawD "\fcancel_state\x18\v \x01(\v2@.temporal.server.chasm.lib.activity.proto.v1.ActivityCancelStateR\vcancelState\x12l\n" + "\x0fterminate_state\x18\f \x01(\v2C.temporal.server.chasm.lib.activity.proto.v1.ActivityTerminateStateR\x0eterminateState\x12:\n" + "\vstart_delay\x18\r \x01(\v2\x19.google.protobuf.DurationR\n" + - "startDelay\"\xa7\x01\n" + + "startDelay\x12:\n" + + "\x19cancel_command_dispatched\x18\x0e \x01(\bR\x17cancelCommandDispatched\"\xa7\x01\n" + "\x13ActivityCancelState\x12\x1d\n" + "\n" + "request_id\x18\x01 \x01(\tR\trequestId\x12=\n" + diff --git a/chasm/lib/activity/proto/v1/activity_state.proto b/chasm/lib/activity/proto/v1/activity_state.proto index d9e8901f7f..6f4adc53f2 100644 --- a/chasm/lib/activity/proto/v1/activity_state.proto +++ b/chasm/lib/activity/proto/v1/activity_state.proto @@ -94,6 +94,11 @@ message ActivityState { // Amount of time to wait before dispatching the activity task to the task queue for the first time. If the activity // has a retry policy, retry attempts will not have start delay applied. google.protobuf.Duration start_delay = 13; + + // Set to true after the cancel command has been successfully dispatched to the worker + // via the Nexus control queue. Used by standby clusters to determine whether the + // dispatch task can be safely discarded. + bool cancel_command_dispatched = 14; } message ActivityCancelState { diff --git a/chasm/lib/activity/worker_command_task_handlers.go b/chasm/lib/activity/worker_command_task_handlers.go index 6d7aa482f6..a4dd52a163 100644 --- a/chasm/lib/activity/worker_command_task_handlers.go +++ b/chasm/lib/activity/worker_command_task_handlers.go @@ -45,6 +45,10 @@ func (h *cancelCommandDispatchTaskHandler) Validate( _ chasm.TaskAttributes, _ *activitypb.CancelCommandDispatchTask, ) (bool, error) { + // Invalid if the cancel command was already dispatched (replicated from the active cluster). + if activity.GetCancelCommandDispatched() { + return false, nil + } // Valid if the activity is in a state where it has been requested to cancel or terminated // (meaning it was running on a worker when the cancel/terminate was issued). return activity.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED || @@ -97,5 +101,24 @@ func (h *cancelCommandDispatchTaskHandler) Execute( // TODO: CHASM's SideEffectTaskHandler interface doesn't expose an attempt count. The // dispatcher's max attempts check is effectively bypassed here. We need to either expose // attempt count in the CHASM task interface or handle retry limiting differently. - return dispatcher.Execute(ctx, task, 1, nsEntry.Name().String()) + if err := dispatcher.Execute(ctx, task, 1, nsEntry.Name().String()); err != nil { + return err + } + + // Record that the cancel command was dispatched. This state is replicated to standby + // clusters, allowing them to invalidate the task via Validate() instead of waiting + // for the discard delay. + _, _, err = chasm.UpdateComponent( + ctx, + activityRef, + (*Activity).recordCancelCommandDispatched, + nil, + ) + return err +} + +// recordCancelCommandDispatched marks that the cancel command has been dispatched. +func (a *Activity) recordCancelCommandDispatched(_ chasm.MutableContext, _ any) (any, error) { + a.CancelCommandDispatched = true + return nil, nil } From d9ddecdd4642671de4cb4c36398b8f1cf434dd15 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Tue, 7 Jul 2026 13:47:17 -0700 Subject: [PATCH 10/22] Remove implementation details from comments Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go | 5 ++--- chasm/lib/activity/proto/v1/activity_state.proto | 5 ++--- chasm/lib/activity/worker_command_task_handlers.go | 5 ++--- 3 files changed, 6 insertions(+), 9 deletions(-) diff --git a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go index bf76195aa3..c1059079cb 100644 --- a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go +++ b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go @@ -186,9 +186,8 @@ type ActivityState struct { // Amount of time to wait before dispatching the activity task to the task queue for the first time. If the activity // has a retry policy, retry attempts will not have start delay applied. StartDelay *durationpb.Duration `protobuf:"bytes,13,opt,name=start_delay,json=startDelay,proto3" json:"start_delay,omitempty"` - // Set to true after the cancel command has been successfully dispatched to the worker - // via the Nexus control queue. Used by standby clusters to determine whether the - // dispatch task can be safely discarded. + // Set to true after the cancel command has been successfully dispatched to the worker. + // Used by standby clusters to determine whether the dispatch task can be safely discarded. CancelCommandDispatched bool `protobuf:"varint,14,opt,name=cancel_command_dispatched,json=cancelCommandDispatched,proto3" json:"cancel_command_dispatched,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache diff --git a/chasm/lib/activity/proto/v1/activity_state.proto b/chasm/lib/activity/proto/v1/activity_state.proto index 6f4adc53f2..6191555cba 100644 --- a/chasm/lib/activity/proto/v1/activity_state.proto +++ b/chasm/lib/activity/proto/v1/activity_state.proto @@ -95,9 +95,8 @@ message ActivityState { // has a retry policy, retry attempts will not have start delay applied. google.protobuf.Duration start_delay = 13; - // Set to true after the cancel command has been successfully dispatched to the worker - // via the Nexus control queue. Used by standby clusters to determine whether the - // dispatch task can be safely discarded. + // Set to true after the cancel command has been successfully dispatched to the worker. + // Used by standby clusters to determine whether the dispatch task can be safely discarded. bool cancel_command_dispatched = 14; } diff --git a/chasm/lib/activity/worker_command_task_handlers.go b/chasm/lib/activity/worker_command_task_handlers.go index a4dd52a163..12ddf9e780 100644 --- a/chasm/lib/activity/worker_command_task_handlers.go +++ b/chasm/lib/activity/worker_command_task_handlers.go @@ -45,7 +45,7 @@ func (h *cancelCommandDispatchTaskHandler) Validate( _ chasm.TaskAttributes, _ *activitypb.CancelCommandDispatchTask, ) (bool, error) { - // Invalid if the cancel command was already dispatched (replicated from the active cluster). + // Invalid if the cancel command was already dispatched. if activity.GetCancelCommandDispatched() { return false, nil } @@ -106,8 +106,7 @@ func (h *cancelCommandDispatchTaskHandler) Execute( } // Record that the cancel command was dispatched. This state is replicated to standby - // clusters, allowing them to invalidate the task via Validate() instead of waiting - // for the discard delay. + // clusters so they can discard the task. _, _, err = chasm.UpdateComponent( ctx, activityRef, From 8406cd098616fe472a843306bf4fbf52e017c04b Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Tue, 7 Jul 2026 13:57:03 -0700 Subject: [PATCH 11/22] Address review comments: unpack handler deps, simplify token API Unpack fx.In options into individual handler struct fields, matching the callback invocationTaskHandler pattern. Simplify NewStandaloneActivityTaskToken to take a single activityID since workflowID and activityID are always identical for standalone activities. Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/activity.go | 3 +-- .../activity/worker_command_task_handlers.go | 24 +++++++++++++------ common/tasktoken/token.go | 8 +++---- 3 files changed, 22 insertions(+), 13 deletions(-) diff --git a/chasm/lib/activity/activity.go b/chasm/lib/activity/activity.go index 117fbe99fd..6ee0d9fe1f 100644 --- a/chasm/lib/activity/activity.go +++ b/chasm/lib/activity/activity.go @@ -226,9 +226,8 @@ func (a *Activity) buildCancelCommandTaskToken(ctx chasm.Context, activityRef ch token := tasktoken.NewStandaloneActivityTaskToken( key.NamespaceID, - key.BusinessID, // workflowID — for standalone activities, BusinessID is the ActivityId + key.BusinessID, // activityID key.RunID, - key.BusinessID, // activityId a.GetActivityType().GetName(), attempt.GetCount(), attempt.GetStartedComponentRef(), diff --git a/chasm/lib/activity/worker_command_task_handlers.go b/chasm/lib/activity/worker_command_task_handlers.go index 12ddf9e780..d5ec60077f 100644 --- a/chasm/lib/activity/worker_command_task_handlers.go +++ b/chasm/lib/activity/worker_command_task_handlers.go @@ -22,7 +22,11 @@ import ( // time out if the worker doesn't respond. type cancelCommandDispatchTaskHandler struct { chasm.SideEffectTaskHandlerBase[*activitypb.CancelCommandDispatchTask] - opts cancelCommandDispatchTaskHandlerOptions + matchingClient resource.MatchingClient + namespaceRegistry namespace.Registry + config *configs.Config + metricsHandler metrics.Handler + logger log.Logger } type cancelCommandDispatchTaskHandlerOptions struct { @@ -36,7 +40,13 @@ type cancelCommandDispatchTaskHandlerOptions struct { } func newCancelCommandDispatchTaskHandler(opts cancelCommandDispatchTaskHandlerOptions) *cancelCommandDispatchTaskHandler { - return &cancelCommandDispatchTaskHandler{opts: opts} + return &cancelCommandDispatchTaskHandler{ + matchingClient: opts.MatchingClient, + namespaceRegistry: opts.NamespaceRegistry, + config: opts.Config, + metricsHandler: opts.MetricsHandler, + logger: opts.Logger, + } } func (h *cancelCommandDispatchTaskHandler) Validate( @@ -72,7 +82,7 @@ func (h *cancelCommandDispatchTaskHandler) Execute( return err } - nsEntry, err := h.opts.NamespaceRegistry.GetNamespaceByID(namespace.ID(activityRef.NamespaceID)) + nsEntry, err := h.namespaceRegistry.GetNamespaceByID(namespace.ID(activityRef.NamespaceID)) if err != nil { return err } @@ -92,10 +102,10 @@ func (h *cancelCommandDispatchTaskHandler) Execute( } dispatcher := workercommands.NewDispatcher( - h.opts.MatchingClient, - h.opts.Config, - h.opts.MetricsHandler, - h.opts.Logger, + h.matchingClient, + h.config, + h.metricsHandler, + h.logger, ) // TODO: CHASM's SideEffectTaskHandler interface doesn't expose an attempt count. The diff --git a/common/tasktoken/token.go b/common/tasktoken/token.go index ad03754d35..7e693a05bf 100644 --- a/common/tasktoken/token.go +++ b/common/tasktoken/token.go @@ -31,18 +31,18 @@ func NewWorkflowTaskToken( } // NewStandaloneActivityTaskToken builds a task token for a standalone activity. -// Standalone activities don't use ScheduledEventId, Clock, Version, or StartVersion. +// For standalone activities, the activity ID is used as both the workflow ID and activity ID +// in the token. Standalone activities don't use ScheduledEventId, Clock, Version, or StartVersion. func NewStandaloneActivityTaskToken( namespaceID string, - workflowID string, - runID string, activityID string, + runID string, activityType string, attempt int32, componentRef []byte, ) *tokenspb.Task { return NewActivityTaskToken( - namespaceID, workflowID, runID, + namespaceID, activityID, runID, 0, // scheduledEventId activityID, activityType, attempt, nil, // clock From 581ee8ab1c9b5c9cb04841d8f70e0eba5e6b5095 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Wed, 8 Jul 2026 15:46:02 -0700 Subject: [PATCH 12/22] Use matching's component ref for cancel command token Capture the ComponentRef from RecordActivityTaskStartedRequest instead of ctx.Ref(a). This is the exact ref matching used to build the poll token, so the cancel command token is guaranteed to match even if the component is mutated between dispatch and start. Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/activity.go | 3 +-- .../gen/activitypb/v1/activity_state.pb.go | 21 +++++++++---------- .../activity/proto/v1/activity_state.proto | 7 +++---- chasm/lib/activity/statemachine.go | 13 +++++------- 4 files changed, 19 insertions(+), 25 deletions(-) diff --git a/chasm/lib/activity/activity.go b/chasm/lib/activity/activity.go index 6ee0d9fe1f..5419cc4469 100644 --- a/chasm/lib/activity/activity.go +++ b/chasm/lib/activity/activity.go @@ -196,7 +196,6 @@ func NewEmbeddedActivity( } func (a *Activity) createAddActivityTaskRequest(ctx chasm.Context, namespaceID string) (*matchingservice.AddActivityTaskRequest, error) { - // Get latest component ref and unmarshal into proto ref componentRef, err := ctx.Ref(a) if err != nil { return nil, err @@ -230,7 +229,7 @@ func (a *Activity) buildCancelCommandTaskToken(ctx chasm.Context, activityRef ch key.RunID, a.GetActivityType().GetName(), attempt.GetCount(), - attempt.GetStartedComponentRef(), + attempt.GetComponentRef(), ) return token.Marshal() diff --git a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go index c1059079cb..7cd3b1ffb6 100644 --- a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go +++ b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go @@ -476,12 +476,11 @@ type ActivityAttemptState struct { // The worker's control task queue for sending commands (e.g. cancel) via Nexus. // Set when the worker reports it during poll. Empty if the worker doesn't support worker commands. WorkerControlTaskQueue string `protobuf:"bytes,12,opt,name=worker_control_task_queue,json=workerControlTaskQueue,proto3" json:"worker_control_task_queue,omitempty"` - // The serialized ComponentRef captured when the task was started. Used to construct - // the task token for cancel commands. The token must match what was sent to the - // worker in the poll response. - StartedComponentRef []byte `protobuf:"bytes,13,opt,name=started_component_ref,json=startedComponentRef,proto3" json:"started_component_ref,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // The serialized ComponentRef captured when the attempt was scheduled. Used to + // construct the task token for both dispatch to matching and cancel commands. + ComponentRef []byte `protobuf:"bytes,13,opt,name=component_ref,json=componentRef,proto3" json:"component_ref,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ActivityAttemptState) Reset() { @@ -598,9 +597,9 @@ func (x *ActivityAttemptState) GetWorkerControlTaskQueue() string { return "" } -func (x *ActivityAttemptState) GetStartedComponentRef() []byte { +func (x *ActivityAttemptState) GetComponentRef() []byte { if x != nil { - return x.StartedComponentRef + return x.ComponentRef } return nil } @@ -992,7 +991,7 @@ const file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawD "\x06reason\x18\x04 \x01(\tR\x06reason\"7\n" + "\x16ActivityTerminateState\x12\x1d\n" + "\n" + - "request_id\x18\x01 \x01(\tR\trequestId\"\x93\a\n" + + "request_id\x18\x01 \x01(\tR\trequestId\"\x84\a\n" + "\x14ActivityAttemptState\x12\x14\n" + "\x05count\x18\x01 \x01(\x05R\x05count\x12O\n" + "\x16current_retry_interval\x18\x02 \x01(\v2\x19.google.protobuf.DurationR\x14currentRetryInterval\x12=\n" + @@ -1007,8 +1006,8 @@ const file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawD " \x01(\tR\asdkName\x12\x1f\n" + "\vsdk_version\x18\v \x01(\tR\n" + "sdkVersion\x129\n" + - "\x19worker_control_task_queue\x18\f \x01(\tR\x16workerControlTaskQueue\x122\n" + - "\x15started_component_ref\x18\r \x01(\fR\x13startedComponentRef\x1a\x80\x01\n" + + "\x19worker_control_task_queue\x18\f \x01(\tR\x16workerControlTaskQueue\x12#\n" + + "\rcomponent_ref\x18\r \x01(\fR\fcomponentRef\x1a\x80\x01\n" + "\x12LastFailureDetails\x12.\n" + "\x04time\x18\x01 \x01(\v2\x1a.google.protobuf.TimestampR\x04time\x12:\n" + "\afailure\x18\x02 \x01(\v2 .temporal.api.failure.v1.FailureR\afailure\"\xc9\x01\n" + diff --git a/chasm/lib/activity/proto/v1/activity_state.proto b/chasm/lib/activity/proto/v1/activity_state.proto index 6191555cba..c94a055d1f 100644 --- a/chasm/lib/activity/proto/v1/activity_state.proto +++ b/chasm/lib/activity/proto/v1/activity_state.proto @@ -173,10 +173,9 @@ message ActivityAttemptState { // Set when the worker reports it during poll. Empty if the worker doesn't support worker commands. string worker_control_task_queue = 12; - // The serialized ComponentRef captured when the task was started. Used to construct - // the task token for cancel commands. The token must match what was sent to the - // worker in the poll response. - bytes started_component_ref = 13; + // The serialized ComponentRef captured when the attempt was scheduled. Used to + // construct the task token for both dispatch to matching and cancel commands. + bytes component_ref = 13; } message ActivityHeartbeatState { diff --git a/chasm/lib/activity/statemachine.go b/chasm/lib/activity/statemachine.go index e50cd70392..d64d4e8dc7 100644 --- a/chasm/lib/activity/statemachine.go +++ b/chasm/lib/activity/statemachine.go @@ -145,14 +145,11 @@ var TransitionStarted = chasm.NewTransition( func(a *Activity, ctx chasm.MutableContext, request *historyservice.RecordActivityTaskStartedRequest) error { attempt := a.LastAttempt.Get(ctx) - // Capture the ComponentRef at start time. Used to construct the task token - // for cancel commands. The token must match what was sent to the worker in - // the poll response. - startedRef, err := ctx.Ref(a) - if err != nil { - return err - } - attempt.StartedComponentRef = startedRef + // Store the ComponentRef that matching used to build the poll token. This is + // the ref from the dispatch task data, which matching also embeds in the task + // token sent to the worker. Using this ref (rather than ctx.Ref) guarantees + // the cancel command token matches the poll token. + attempt.ComponentRef = request.GetComponentRef() attempt.StartedTime = timestamppb.New(ctx.Now(a)) attempt.StartRequestId = request.GetRequestId() From d21dd91fe49e86f5b865f931b00abe82a943410d Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Wed, 8 Jul 2026 15:47:48 -0700 Subject: [PATCH 13/22] Revert cancel_command_dispatched standby optimization Remove the cancel_command_dispatched flag and associated logic. Standby handling will be addressed via a new CHASM abstraction instead. Co-Authored-By: Claude Opus 4.6 --- .../gen/activitypb/v1/activity_state.pb.go | 21 ++++------------ .../activity/proto/v1/activity_state.proto | 4 ---- .../activity/worker_command_task_handlers.go | 24 +------------------ 3 files changed, 6 insertions(+), 43 deletions(-) diff --git a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go index 7cd3b1ffb6..67d1cb4ec7 100644 --- a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go +++ b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go @@ -185,12 +185,9 @@ type ActivityState struct { TerminateState *ActivityTerminateState `protobuf:"bytes,12,opt,name=terminate_state,json=terminateState,proto3" json:"terminate_state,omitempty"` // Amount of time to wait before dispatching the activity task to the task queue for the first time. If the activity // has a retry policy, retry attempts will not have start delay applied. - StartDelay *durationpb.Duration `protobuf:"bytes,13,opt,name=start_delay,json=startDelay,proto3" json:"start_delay,omitempty"` - // Set to true after the cancel command has been successfully dispatched to the worker. - // Used by standby clusters to determine whether the dispatch task can be safely discarded. - CancelCommandDispatched bool `protobuf:"varint,14,opt,name=cancel_command_dispatched,json=cancelCommandDispatched,proto3" json:"cancel_command_dispatched,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + StartDelay *durationpb.Duration `protobuf:"bytes,13,opt,name=start_delay,json=startDelay,proto3" json:"start_delay,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ActivityState) Reset() { @@ -314,13 +311,6 @@ func (x *ActivityState) GetStartDelay() *durationpb.Duration { return nil } -func (x *ActivityState) GetCancelCommandDispatched() bool { - if x != nil { - return x.CancelCommandDispatched - } - return false -} - type ActivityCancelState struct { state protoimpl.MessageState `protogen:"open.v1"` RequestId string `protobuf:"bytes,1,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"` @@ -964,7 +954,7 @@ var File_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto protor const file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawDesc = "" + "\n" + - "@temporal/server/chasm/lib/activity/proto/v1/activity_state.proto\x12+temporal.server.chasm.lib.activity.proto.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\x1a(temporal/api/deployment/v1/message.proto\x1a%temporal/api/failure/v1/message.proto\x1a'temporal/api/sdk/v1/user_metadata.proto\x1a'temporal/api/taskqueue/v1/message.proto\"\xd3\b\n" + + "@temporal/server/chasm/lib/activity/proto/v1/activity_state.proto\x12+temporal.server.chasm.lib.activity.proto.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\x1a(temporal/api/deployment/v1/message.proto\x1a%temporal/api/failure/v1/message.proto\x1a'temporal/api/sdk/v1/user_metadata.proto\x1a'temporal/api/taskqueue/v1/message.proto\"\x97\b\n" + "\rActivityState\x12I\n" + "\ractivity_type\x18\x01 \x01(\v2$.temporal.api.common.v1.ActivityTypeR\factivityType\x12C\n" + "\n" + @@ -981,8 +971,7 @@ const file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawD "\fcancel_state\x18\v \x01(\v2@.temporal.server.chasm.lib.activity.proto.v1.ActivityCancelStateR\vcancelState\x12l\n" + "\x0fterminate_state\x18\f \x01(\v2C.temporal.server.chasm.lib.activity.proto.v1.ActivityTerminateStateR\x0eterminateState\x12:\n" + "\vstart_delay\x18\r \x01(\v2\x19.google.protobuf.DurationR\n" + - "startDelay\x12:\n" + - "\x19cancel_command_dispatched\x18\x0e \x01(\bR\x17cancelCommandDispatched\"\xa7\x01\n" + + "startDelay\"\xa7\x01\n" + "\x13ActivityCancelState\x12\x1d\n" + "\n" + "request_id\x18\x01 \x01(\tR\trequestId\x12=\n" + diff --git a/chasm/lib/activity/proto/v1/activity_state.proto b/chasm/lib/activity/proto/v1/activity_state.proto index c94a055d1f..405f7dd557 100644 --- a/chasm/lib/activity/proto/v1/activity_state.proto +++ b/chasm/lib/activity/proto/v1/activity_state.proto @@ -94,10 +94,6 @@ message ActivityState { // Amount of time to wait before dispatching the activity task to the task queue for the first time. If the activity // has a retry policy, retry attempts will not have start delay applied. google.protobuf.Duration start_delay = 13; - - // Set to true after the cancel command has been successfully dispatched to the worker. - // Used by standby clusters to determine whether the dispatch task can be safely discarded. - bool cancel_command_dispatched = 14; } message ActivityCancelState { diff --git a/chasm/lib/activity/worker_command_task_handlers.go b/chasm/lib/activity/worker_command_task_handlers.go index d5ec60077f..468afbfce5 100644 --- a/chasm/lib/activity/worker_command_task_handlers.go +++ b/chasm/lib/activity/worker_command_task_handlers.go @@ -55,10 +55,6 @@ func (h *cancelCommandDispatchTaskHandler) Validate( _ chasm.TaskAttributes, _ *activitypb.CancelCommandDispatchTask, ) (bool, error) { - // Invalid if the cancel command was already dispatched. - if activity.GetCancelCommandDispatched() { - return false, nil - } // Valid if the activity is in a state where it has been requested to cancel or terminated // (meaning it was running on a worker when the cancel/terminate was issued). return activity.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED || @@ -111,23 +107,5 @@ func (h *cancelCommandDispatchTaskHandler) Execute( // TODO: CHASM's SideEffectTaskHandler interface doesn't expose an attempt count. The // dispatcher's max attempts check is effectively bypassed here. We need to either expose // attempt count in the CHASM task interface or handle retry limiting differently. - if err := dispatcher.Execute(ctx, task, 1, nsEntry.Name().String()); err != nil { - return err - } - - // Record that the cancel command was dispatched. This state is replicated to standby - // clusters so they can discard the task. - _, _, err = chasm.UpdateComponent( - ctx, - activityRef, - (*Activity).recordCancelCommandDispatched, - nil, - ) - return err -} - -// recordCancelCommandDispatched marks that the cancel command has been dispatched. -func (a *Activity) recordCancelCommandDispatched(_ chasm.MutableContext, _ any) (any, error) { - a.CancelCommandDispatched = true - return nil, nil + return dispatcher.Execute(ctx, task, 1, nsEntry.Name().String()) } From ffd7055ff2d1824201bafbcdb370a31dad60cd6b Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Thu, 9 Jul 2026 10:25:39 -0700 Subject: [PATCH 14/22] Use task attempt count for cancel command dispatch retry limiting Now that CHASM exposes attempt count via TaskAttributes, use it to: - Drop the cancel command dispatch task after MaxTaskAttempts in Validate - Pass the real attempt count to the dispatcher instead of hardcoded 1 Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/worker_command_task_handlers.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/chasm/lib/activity/worker_command_task_handlers.go b/chasm/lib/activity/worker_command_task_handlers.go index 468afbfce5..ebc4ce83ae 100644 --- a/chasm/lib/activity/worker_command_task_handlers.go +++ b/chasm/lib/activity/worker_command_task_handlers.go @@ -52,9 +52,12 @@ func newCancelCommandDispatchTaskHandler(opts cancelCommandDispatchTaskHandlerOp func (h *cancelCommandDispatchTaskHandler) Validate( _ chasm.Context, activity *Activity, - _ chasm.TaskAttributes, + taskAttrs chasm.TaskAttributes, _ *activitypb.CancelCommandDispatchTask, ) (bool, error) { + if taskAttrs.Attempt > workercommands.MaxTaskAttempts { + return false, nil + } // Valid if the activity is in a state where it has been requested to cancel or terminated // (meaning it was running on a worker when the cancel/terminate was issued). return activity.GetStatus() == activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED || @@ -104,8 +107,5 @@ func (h *cancelCommandDispatchTaskHandler) Execute( h.logger, ) - // TODO: CHASM's SideEffectTaskHandler interface doesn't expose an attempt count. The - // dispatcher's max attempts check is effectively bypassed here. We need to either expose - // attempt count in the CHASM task interface or handle retry limiting differently. - return dispatcher.Execute(ctx, task, 1, nsEntry.Name().String()) + return dispatcher.Execute(ctx, task, taskAttrs.Attempt, nsEntry.Name().String()) } From 686154ddd8f502051a4df6e40ddce072451ee988 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Mon, 13 Jul 2026 14:44:44 -0700 Subject: [PATCH 15/22] Fix cancel command handler to match TaskInvocation interface Use TaskInvocation (not TaskAttributes) in Validate, since Attempt is only available there. Keep dispatcher call with attempt=1 since max attempts are enforced in Validate. Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/worker_command_task_handlers.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/chasm/lib/activity/worker_command_task_handlers.go b/chasm/lib/activity/worker_command_task_handlers.go index ebc4ce83ae..755a4b7d7d 100644 --- a/chasm/lib/activity/worker_command_task_handlers.go +++ b/chasm/lib/activity/worker_command_task_handlers.go @@ -52,10 +52,10 @@ func newCancelCommandDispatchTaskHandler(opts cancelCommandDispatchTaskHandlerOp func (h *cancelCommandDispatchTaskHandler) Validate( _ chasm.Context, activity *Activity, - taskAttrs chasm.TaskAttributes, + invocation chasm.TaskInvocation, _ *activitypb.CancelCommandDispatchTask, ) (bool, error) { - if taskAttrs.Attempt > workercommands.MaxTaskAttempts { + if invocation.Attempt > workercommands.MaxTaskAttempts { return false, nil } // Valid if the activity is in a state where it has been requested to cancel or terminated @@ -107,5 +107,7 @@ func (h *cancelCommandDispatchTaskHandler) Execute( h.logger, ) - return dispatcher.Execute(ctx, task, taskAttrs.Attempt, nsEntry.Name().String()) + // Attempt count is not available in Execute (only in Validate via TaskInvocation). + // Max attempts are enforced in Validate, so this is always within bounds. + return dispatcher.Execute(ctx, task, 1, nsEntry.Name().String()) } From ee3b094d7c327f7fe5be460f7b8ea385c5fe29b2 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Mon, 13 Jul 2026 16:42:15 -0700 Subject: [PATCH 16/22] Remove WorkflowId/RunId from standalone activity token Standalone activities route responses via ComponentRef, not WorkflowId or RunId. Remove the Execution field from AddActivityTaskRequest and simplify NewStandaloneActivityTaskToken to pass empty strings, matching what matching produces in the poll token. Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/activity.go | 15 ++++----------- common/tasktoken/token.go | 13 ++++++------- 2 files changed, 10 insertions(+), 18 deletions(-) diff --git a/chasm/lib/activity/activity.go b/chasm/lib/activity/activity.go index 5419cc4469..a28d3c196e 100644 --- a/chasm/lib/activity/activity.go +++ b/chasm/lib/activity/activity.go @@ -201,14 +201,10 @@ func (a *Activity) createAddActivityTaskRequest(ctx chasm.Context, namespaceID s return nil, err } - key := ctx.ExecutionKey() - + // Note: No need to set the vector clock here, as the components track version conflicts for read/write + // TODO: Need to fill in VersionDirective once we decide how to handle versioning for standalone activities return &matchingservice.AddActivityTaskRequest{ - NamespaceId: namespaceID, - Execution: &commonpb.WorkflowExecution{ - WorkflowId: key.BusinessID, - RunId: key.RunID, - }, + NamespaceId: namespaceID, ScheduleToStartTimeout: a.ScheduleToStartTimeout, TaskQueue: a.GetTaskQueue(), Priority: a.GetPriority(), @@ -221,12 +217,9 @@ func (a *Activity) createAddActivityTaskRequest(ctx chasm.Context, namespaceID s // The token must match what was sent to the worker in the poll response. func (a *Activity) buildCancelCommandTaskToken(ctx chasm.Context, activityRef chasm.ComponentRef) ([]byte, error) { attempt := a.LastAttempt.Get(ctx) - key := ctx.ExecutionKey() token := tasktoken.NewStandaloneActivityTaskToken( - key.NamespaceID, - key.BusinessID, // activityID - key.RunID, + activityRef.NamespaceID, a.GetActivityType().GetName(), attempt.GetCount(), attempt.GetComponentRef(), diff --git a/common/tasktoken/token.go b/common/tasktoken/token.go index 7e693a05bf..0387b0b7b0 100644 --- a/common/tasktoken/token.go +++ b/common/tasktoken/token.go @@ -31,20 +31,19 @@ func NewWorkflowTaskToken( } // NewStandaloneActivityTaskToken builds a task token for a standalone activity. -// For standalone activities, the activity ID is used as both the workflow ID and activity ID -// in the token. Standalone activities don't use ScheduledEventId, Clock, Version, or StartVersion. +// Standalone activities don't use WorkflowId, RunId, ScheduledEventId, Clock, Version, or +// StartVersion. The ComponentRef is the sole identifier for routing responses. func NewStandaloneActivityTaskToken( namespaceID string, - activityID string, - runID string, activityType string, attempt int32, componentRef []byte, ) *tokenspb.Task { return NewActivityTaskToken( - namespaceID, activityID, runID, - 0, // scheduledEventId - activityID, activityType, attempt, + namespaceID, "", "", + 0, // scheduledEventId + "", // activityId + activityType, attempt, nil, // clock 0, // version 0, // startVersion From cde28271d9e30f7a0eb2e86317fb10675ce30edc Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Mon, 13 Jul 2026 21:31:51 -0700 Subject: [PATCH 17/22] Remove WorkflowId/RunId from standalone activity token Standalone activities don't have a parent workflow. Remove the Execution field from AddActivityTaskRequest and leave WorkflowId/RunId empty in NewStandaloneActivityTaskToken. Matching gets ActivityId from the RecordActivityTaskStartedResponse, not from AddActivityTaskRequest, so the poll and cancel command tokens remain consistent. Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/activity.go | 4 +++- common/tasktoken/token.go | 16 +++++++++++----- 2 files changed, 14 insertions(+), 6 deletions(-) diff --git a/chasm/lib/activity/activity.go b/chasm/lib/activity/activity.go index a28d3c196e..3b39c4484e 100644 --- a/chasm/lib/activity/activity.go +++ b/chasm/lib/activity/activity.go @@ -217,9 +217,11 @@ func (a *Activity) createAddActivityTaskRequest(ctx chasm.Context, namespaceID s // The token must match what was sent to the worker in the poll response. func (a *Activity) buildCancelCommandTaskToken(ctx chasm.Context, activityRef chasm.ComponentRef) ([]byte, error) { attempt := a.LastAttempt.Get(ctx) + key := ctx.ExecutionKey() token := tasktoken.NewStandaloneActivityTaskToken( - activityRef.NamespaceID, + key.NamespaceID, + key.BusinessID, // activityID a.GetActivityType().GetName(), attempt.GetCount(), attempt.GetComponentRef(), diff --git a/common/tasktoken/token.go b/common/tasktoken/token.go index 0387b0b7b0..398c4ecc6e 100644 --- a/common/tasktoken/token.go +++ b/common/tasktoken/token.go @@ -31,19 +31,25 @@ func NewWorkflowTaskToken( } // NewStandaloneActivityTaskToken builds a task token for a standalone activity. -// Standalone activities don't use WorkflowId, RunId, ScheduledEventId, Clock, Version, or -// StartVersion. The ComponentRef is the sole identifier for routing responses. +// The token fields must match what matching produces in the poll token. Matching gets WorkflowId +// and RunId from AddActivityTaskRequest.Execution (empty for standalone activities) and ActivityId +// from RecordActivityTaskStartedResponse. ScheduledEventId, Clock, Version, and StartVersion are +// unused for standalone activities. func NewStandaloneActivityTaskToken( namespaceID string, + activityID string, activityType string, attempt int32, componentRef []byte, ) *tokenspb.Task { return NewActivityTaskToken( - namespaceID, "", "", + namespaceID, + "", // workflowId — not applicable for standalone activities + "", // runId — not applicable for standalone activities 0, // scheduledEventId - "", // activityId - activityType, attempt, + activityID, + activityType, + attempt, nil, // clock 0, // version 0, // startVersion From da429a442f2411cf7f1a8bd76d048db99af82b88 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Mon, 13 Jul 2026 21:41:58 -0700 Subject: [PATCH 18/22] Remove unnecessary comment from NewStandaloneActivityTaskToken Co-Authored-By: Claude Opus 4.6 --- common/tasktoken/token.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/common/tasktoken/token.go b/common/tasktoken/token.go index 398c4ecc6e..6cdac9471f 100644 --- a/common/tasktoken/token.go +++ b/common/tasktoken/token.go @@ -31,10 +31,6 @@ func NewWorkflowTaskToken( } // NewStandaloneActivityTaskToken builds a task token for a standalone activity. -// The token fields must match what matching produces in the poll token. Matching gets WorkflowId -// and RunId from AddActivityTaskRequest.Execution (empty for standalone activities) and ActivityId -// from RecordActivityTaskStartedResponse. ScheduledEventId, Clock, Version, and StartVersion are -// unused for standalone activities. func NewStandaloneActivityTaskToken( namespaceID string, activityID string, From 2db056bf2aa923d1f746c0b8d2ee9972287579b7 Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Mon, 13 Jul 2026 23:19:16 -0700 Subject: [PATCH 19/22] Add tests for cancel command dispatch - Unit tests for cancelCommandDispatchTaskHandler.Validate: status checks and max attempt enforcement - TransitionStarted test: verify ComponentRef and WorkerControlTaskQueue are captured from RecordActivityTaskStartedRequest - E2e test: terminate dispatches cancel command to worker control queue Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/statemachine_test.go | 7 +- .../worker_command_task_handlers_test.go | 78 +++++++++++++++++++ tests/activity_standalone_test.go | 73 +++++++++++++++++ 3 files changed, 157 insertions(+), 1 deletion(-) create mode 100644 chasm/lib/activity/worker_command_task_handlers_test.go diff --git a/chasm/lib/activity/statemachine_test.go b/chasm/lib/activity/statemachine_test.go index f5b40927d3..6a350ca82c 100644 --- a/chasm/lib/activity/statemachine_test.go +++ b/chasm/lib/activity/statemachine_test.go @@ -303,10 +303,13 @@ func TestTransitionStarted(t *testing.T) { Outcome: chasm.NewDataField(ctx, outcome), } + componentRef := []byte("test-component-ref") err := TransitionStarted.Apply(activity, ctx, &historyservice.RecordActivityTaskStartedRequest{ PollRequest: &workflowservice.PollActivityTaskQueueRequest{ - Identity: "test-worker", + Identity: "test-worker", + WorkerControlTaskQueue: "test-control-queue", }, + ComponentRef: componentRef, }) require.NoError(t, err) require.Equal(t, activitypb.ACTIVITY_EXECUTION_STATUS_STARTED, activity.Status) @@ -315,6 +318,8 @@ func TestTransitionStarted(t *testing.T) { require.Equal(t, "test-worker", attemptState.LastWorkerIdentity) require.Equal(t, headers.ClientNameGoSDK, attemptState.SdkName) require.Equal(t, temporal.SDKVersion, attemptState.SdkVersion) + require.Equal(t, componentRef, attemptState.ComponentRef) + require.Equal(t, "test-control-queue", attemptState.WorkerControlTaskQueue) // Verify added tasks require.Len(t, ctx.Tasks, 1) diff --git a/chasm/lib/activity/worker_command_task_handlers_test.go b/chasm/lib/activity/worker_command_task_handlers_test.go new file mode 100644 index 0000000000..fe5466d470 --- /dev/null +++ b/chasm/lib/activity/worker_command_task_handlers_test.go @@ -0,0 +1,78 @@ +package activity + +import ( + "testing" + + "github.com/stretchr/testify/require" + "go.temporal.io/server/chasm" + "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" + "go.temporal.io/server/common/workercommands" +) + +func TestCancelCommandDispatchTaskHandler_Validate(t *testing.T) { + handler := &cancelCommandDispatchTaskHandler{} + + testCases := []struct { + name string + status activitypb.ActivityExecutionStatus + attempt int + expected bool + }{ + { + name: "cancel requested", + status: activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED, + attempt: 1, + expected: true, + }, + { + name: "terminated", + status: activitypb.ACTIVITY_EXECUTION_STATUS_TERMINATED, + attempt: 1, + expected: true, + }, + { + name: "scheduled", + status: activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED, + attempt: 1, + expected: false, + }, + { + name: "started", + status: activitypb.ACTIVITY_EXECUTION_STATUS_STARTED, + attempt: 1, + expected: false, + }, + { + name: "completed", + status: activitypb.ACTIVITY_EXECUTION_STATUS_COMPLETED, + attempt: 1, + expected: false, + }, + { + name: "cancel requested at max attempts", + status: activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED, + attempt: workercommands.MaxTaskAttempts, + expected: true, + }, + { + name: "cancel requested exceeds max attempts", + status: activitypb.ACTIVITY_EXECUTION_STATUS_CANCEL_REQUESTED, + attempt: workercommands.MaxTaskAttempts + 1, + expected: false, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + activity := &Activity{ + ActivityState: &activitypb.ActivityState{ + Status: tc.status, + }, + } + invocation := chasm.TaskInvocation{Attempt: tc.attempt} + valid, err := handler.Validate(nil, activity, invocation, nil) + require.NoError(t, err) + require.Equal(t, tc.expected, valid) + }) + } +} diff --git a/tests/activity_standalone_test.go b/tests/activity_standalone_test.go index 27df7a420f..8b9322ba57 100644 --- a/tests/activity_standalone_test.go +++ b/tests/activity_standalone_test.go @@ -2702,6 +2702,79 @@ func (s *standaloneActivityTestSuite) TestDispatchCancelCommandToWorker() { require.Equal(t, activityPollResp.TaskToken, cancelCmd.TaskToken) } +// TestDispatchCancelCommandOnTerminate tests that when a running standalone activity is terminated, +// the server dispatches a cancel command to the worker's control queue. +func (s *standaloneActivityTestSuite) TestDispatchCancelCommandOnTerminate() { + env := s.newTestEnv() + t := s.T() + ctx := s.Context() + + env.GetTestCluster().OverrideDynamicConfig( + t, dynamicconfig.EnableCancelActivityWorkerCommand, + true, + ) + + activityID := testcore.RandomizeStr(t.Name()) + taskQueue := testcore.RandomizeStr(t.Name()) + tv := env.Tv() + controlQueueName := tv.ControlQueueName(env.Namespace().String()) + + startResp := env.startAndValidateActivity(ctx, t, activityID, taskQueue) + runID := startResp.RunId + + // Poll with worker command support. + activityPollResp, err := env.FrontendClient().PollActivityTaskQueue(ctx, &workflowservice.PollActivityTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{ + Name: taskQueue, + Kind: enumspb.TASK_QUEUE_KIND_NORMAL, + }, + Identity: tv.WorkerIdentity(), + WorkerInstanceKey: tv.WorkerInstanceKey(), + WorkerControlTaskQueue: controlQueueName, + }) + require.NoError(t, err) + require.NotEmpty(t, activityPollResp.TaskToken) + + // Terminate the activity. + _, err = env.FrontendClient().TerminateActivityExecution(ctx, &workflowservice.TerminateActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: runID, + Reason: "test terminate cancel command", + Identity: "terminator", + }) + require.NoError(t, err) + + // Poll the Nexus control queue — should receive the cancel command. + var nexusPollResp *workflowservice.PollNexusTaskQueueResponse + await.RequireTrue(t, func() bool { + pollCtx, pollCancel := context.WithTimeout(ctx, 5*time.Second) + defer pollCancel() + resp, err := env.FrontendClient().PollNexusTaskQueue(pollCtx, &workflowservice.PollNexusTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: controlQueueName, Kind: enumspb.TASK_QUEUE_KIND_WORKER_COMMANDS}, + Identity: tv.WorkerIdentity(), + }) + if err == nil && resp != nil && resp.Request != nil { + nexusPollResp = resp + return true + } + return false + }, 30*time.Second, 200*time.Millisecond) + + startOp := nexusPollResp.Request.GetStartOperation() + require.NotNil(t, startOp, "expected StartOperation in Nexus request") + require.Equal(t, "ExecuteCommands", startOp.Operation) + + var executeReq workerservicepb.ExecuteCommandsRequest + require.NoError(t, proto.Unmarshal(startOp.Payload.Data, &executeReq)) + require.Len(t, executeReq.Commands, 1) + cancelCmd := executeReq.Commands[0].GetCancelActivity() + require.NotNil(t, cancelCmd, "expected CancelActivity command") + require.Equal(t, activityPollResp.TaskToken, cancelCmd.TaskToken) +} + func (s *standaloneActivityTestSuite) TestTerminate() { env := s.newTestEnv() t := s.T() From 91a7dcc7fa8cf12818d305d4714ddbff0f7d4cca Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Tue, 14 Jul 2026 14:55:01 -0700 Subject: [PATCH 20/22] Remove stale comment about attempt availability in Execute Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/worker_command_task_handlers.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/chasm/lib/activity/worker_command_task_handlers.go b/chasm/lib/activity/worker_command_task_handlers.go index 755a4b7d7d..3659148009 100644 --- a/chasm/lib/activity/worker_command_task_handlers.go +++ b/chasm/lib/activity/worker_command_task_handlers.go @@ -107,7 +107,5 @@ func (h *cancelCommandDispatchTaskHandler) Execute( h.logger, ) - // Attempt count is not available in Execute (only in Validate via TaskInvocation). - // Max attempts are enforced in Validate, so this is always within bounds. return dispatcher.Execute(ctx, task, 1, nsEntry.Name().String()) } From aee246ff89e71320f29d4a76e86792f9d26ef20b Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Tue, 14 Jul 2026 14:57:30 -0700 Subject: [PATCH 21/22] Log when cancel command dispatch exceeds max attempts Co-Authored-By: Claude Opus 4.6 --- .../lib/activity/worker_command_task_handlers.go | 9 ++++++++- .../activity/worker_command_task_handlers_test.go | 15 +++++++++++++-- 2 files changed, 21 insertions(+), 3 deletions(-) diff --git a/chasm/lib/activity/worker_command_task_handlers.go b/chasm/lib/activity/worker_command_task_handlers.go index 3659148009..a34f0a7e32 100644 --- a/chasm/lib/activity/worker_command_task_handlers.go +++ b/chasm/lib/activity/worker_command_task_handlers.go @@ -8,6 +8,7 @@ import ( "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" "go.temporal.io/server/common/definition" "go.temporal.io/server/common/log" + "go.temporal.io/server/common/log/tag" "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/namespace" "go.temporal.io/server/common/resource" @@ -50,12 +51,18 @@ func newCancelCommandDispatchTaskHandler(opts cancelCommandDispatchTaskHandlerOp } func (h *cancelCommandDispatchTaskHandler) Validate( - _ chasm.Context, + ctx chasm.Context, activity *Activity, invocation chasm.TaskInvocation, _ *activitypb.CancelCommandDispatchTask, ) (bool, error) { if invocation.Attempt > workercommands.MaxTaskAttempts { + key := ctx.ExecutionKey() + h.logger.Info("Cancel command dispatch task exceeded max attempts, dropping", + tag.WorkflowNamespaceID(key.NamespaceID), + tag.ActivityID(key.BusinessID), + tag.Attempt(int32(invocation.Attempt)), + ) return false, nil } // Valid if the activity is in a state where it has been requested to cancel or terminated diff --git a/chasm/lib/activity/worker_command_task_handlers_test.go b/chasm/lib/activity/worker_command_task_handlers_test.go index fe5466d470..e6d918e630 100644 --- a/chasm/lib/activity/worker_command_task_handlers_test.go +++ b/chasm/lib/activity/worker_command_task_handlers_test.go @@ -5,12 +5,15 @@ import ( "github.com/stretchr/testify/require" "go.temporal.io/server/chasm" + "go.temporal.io/server/common/log" "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" "go.temporal.io/server/common/workercommands" ) func TestCancelCommandDispatchTaskHandler_Validate(t *testing.T) { - handler := &cancelCommandDispatchTaskHandler{} + handler := &cancelCommandDispatchTaskHandler{ + logger: log.NewNoopLogger(), + } testCases := []struct { name string @@ -69,8 +72,16 @@ func TestCancelCommandDispatchTaskHandler_Validate(t *testing.T) { Status: tc.status, }, } + ctx := &chasm.MockContext{ + HandleExecutionKey: func() chasm.ExecutionKey { + return chasm.ExecutionKey{ + NamespaceID: "test-ns-id", + BusinessID: "test-activity-id", + } + }, + } invocation := chasm.TaskInvocation{Attempt: tc.attempt} - valid, err := handler.Validate(nil, activity, invocation, nil) + valid, err := handler.Validate(ctx, activity, invocation, nil) require.NoError(t, err) require.Equal(t, tc.expected, valid) }) From 232a32cb127e3d701d3d366d838ea486d244d35c Mon Sep 17 00:00:00 2001 From: Kannan Rajah Date: Wed, 15 Jul 2026 09:25:19 -0700 Subject: [PATCH 22/22] Remove hardcoded attempt parameter from dispatcher.Execute Now that #11066 moved the attempt check to callers, the dispatcher no longer takes an attempt parameter. Co-Authored-By: Claude Opus 4.6 --- chasm/lib/activity/worker_command_task_handlers.go | 2 +- chasm/lib/activity/worker_command_task_handlers_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/chasm/lib/activity/worker_command_task_handlers.go b/chasm/lib/activity/worker_command_task_handlers.go index a34f0a7e32..4032ff0fb0 100644 --- a/chasm/lib/activity/worker_command_task_handlers.go +++ b/chasm/lib/activity/worker_command_task_handlers.go @@ -114,5 +114,5 @@ func (h *cancelCommandDispatchTaskHandler) Execute( h.logger, ) - return dispatcher.Execute(ctx, task, 1, nsEntry.Name().String()) + return dispatcher.Execute(ctx, task, nsEntry.Name().String()) } diff --git a/chasm/lib/activity/worker_command_task_handlers_test.go b/chasm/lib/activity/worker_command_task_handlers_test.go index e6d918e630..25b0f48248 100644 --- a/chasm/lib/activity/worker_command_task_handlers_test.go +++ b/chasm/lib/activity/worker_command_task_handlers_test.go @@ -5,8 +5,8 @@ import ( "github.com/stretchr/testify/require" "go.temporal.io/server/chasm" - "go.temporal.io/server/common/log" "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" + "go.temporal.io/server/common/log" "go.temporal.io/server/common/workercommands" )