diff --git a/chasm/lib/activity/activity.go b/chasm/lib/activity/activity.go index 9c4c15bbbe7..d88370692cd 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" @@ -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 diff --git a/chasm/lib/activity/config.go b/chasm/lib/activity/config.go index 421757c1e52..04b7f829aba 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 f786cb674b3..4516f5bb2e1 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 09c1cc3d715..67d1cb4ec71 100644 --- a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go +++ b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go @@ -462,7 +462,13 @@ 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"` + 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 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 } @@ -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) GetComponentRef() []byte { + if x != nil { + return x.ComponentRef + } + 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\"\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" + @@ -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\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/gen/activitypb/v1/tasks.go-helpers.pb.go b/chasm/lib/activity/gen/activitypb/v1/tasks.go-helpers.pb.go index d7628a6e9e6..a4173d9659f 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 796574e7db2..23fc96a8db5 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 8f939df2fca..ac514e00a8c 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 26465402039..405f7dd5576 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 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/proto/v1/tasks.proto b/chasm/lib/activity/proto/v1/tasks.proto index 9a1996e3dd2..70dd3ea992a 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 77672eb43a1..f60d3316fec 100644 --- a/chasm/lib/activity/statemachine.go +++ b/chasm/lib/activity/statemachine.go @@ -144,11 +144,19 @@ var TransitionStarted = chasm.NewTransition( activitypb.ACTIVITY_EXECUTION_STATUS_STARTED, func(a *Activity, ctx chasm.MutableContext, request *historyservice.RecordActivityTaskStartedRequest) error { attempt := a.LastAttempt.Get(ctx) + + // 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() 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/chasm/lib/activity/statemachine_test.go b/chasm/lib/activity/statemachine_test.go index ababbd05eba..ddf97972b57 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.go b/chasm/lib/activity/worker_command_task_handlers.go new file mode 100644 index 00000000000..4032ff0fb09 --- /dev/null +++ b/chasm/lib/activity/worker_command_task_handlers.go @@ -0,0 +1,118 @@ +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/log/tag" + "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] + matchingClient resource.MatchingClient + namespaceRegistry namespace.Registry + config *configs.Config + metricsHandler metrics.Handler + logger log.Logger +} + +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{ + matchingClient: opts.MatchingClient, + namespaceRegistry: opts.NamespaceRegistry, + config: opts.Config, + metricsHandler: opts.MetricsHandler, + logger: opts.Logger, + } +} + +func (h *cancelCommandDispatchTaskHandler) Validate( + 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 + // (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.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.matchingClient, + h.config, + h.metricsHandler, + h.logger, + ) + + 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 new file mode 100644 index 00000000000..25b0f482486 --- /dev/null +++ b/chasm/lib/activity/worker_command_task_handlers_test.go @@ -0,0 +1,89 @@ +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/log" + "go.temporal.io/server/common/workercommands" +) + +func TestCancelCommandDispatchTaskHandler_Validate(t *testing.T) { + handler := &cancelCommandDispatchTaskHandler{ + logger: log.NewNoopLogger(), + } + + 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, + }, + } + 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(ctx, activity, invocation, nil) + require.NoError(t, err) + require.Equal(t, tc.expected, valid) + }) + } +} diff --git a/common/tasktoken/token.go b/common/tasktoken/token.go index 4212b4f93c8..6cdac9471f4 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,35 @@ func NewWorkflowTaskToken( } } +// NewStandaloneActivityTaskToken builds a task token for a standalone activity. +func NewStandaloneActivityTaskToken( + namespaceID string, + activityID string, + activityType string, + attempt int32, + componentRef []byte, +) *tokenspb.Task { + return NewActivityTaskToken( + namespaceID, + "", // workflowId — not applicable for standalone activities + "", // runId — not applicable for standalone activities + 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 +73,7 @@ func NewActivityTaskToken( ScheduledEventId: scheduledEventID, ActivityType: activityType, Attempt: attempt, - ActivityId: activityId, + ActivityId: activityID, Clock: clock, Version: version, StartVersion: startVersion, diff --git a/tests/activity_standalone_test.go b/tests/activity_standalone_test.go index 263d0aaaa72..a2e1f129b1a 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,162 @@ 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's task token must match what was sent to the worker in the poll response. + 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()