Dispatch cancel command for standalone activities

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"
@@ -195,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
@@ -213,6 +213,23 @@ func (a *Activity) createAddActivityTaskRequest(ctx chasm.Context, namespaceID s
 	}, nil
 }
 
+// buildCancelCommandTaskToken builds the serialized task token for a cancel command.
+// 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
+		a.GetActivityType().GetName(),
+		attempt.GetCount(),
+		attempt.GetComponentRef(),
+	)
+
+	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 +588,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 +617,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 +677,8 @@ func (a *Activity) handleCancellationRequested(ctx chasm.MutableContext, request
 		if err != nil {
 			return nil, err
 		}
+	} else {
+		a.addCancelCommandDispatchTask(ctx)
 	}
 
 	return &activitypb.RequestCancelActivityExecutionResponse{}, nil

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),
 	}
 }
 

chasm/lib/activity/fx.go

@@ -13,6 +13,7 @@ var HistoryModule = fx.Module(
 		ConfigProvider,
 		linkValidatorProvider,
 		newActivityDispatchTaskHandler,
+		newCancelCommandDispatchTaskHandler,
 		newScheduleToStartTimeoutTaskHandler,
 		newScheduleToCloseTimeoutTaskHandler,
 		newStartToCloseTimeoutTaskHandler,

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,
+		),
 	}
 }

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 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 {

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 {}