diff --git a/chasm/lib/activity/config.go b/chasm/lib/activity/config.go index a6551f49dd7..8e708980c7e 100644 --- a/chasm/lib/activity/config.go +++ b/chasm/lib/activity/config.go @@ -35,6 +35,12 @@ var ( `Allows non-zero start_delay on StartActivityExecution requests.`, ) + EnableEagerStart = dynamicconfig.NewNamespaceBoolSetting( + "activity.enableEagerStart", + true, + `Allows the first standalone activity task to be returned directly by StartActivityExecution.`, + ) + EnableCallbacks = dynamicconfig.NewNamespaceBoolSetting( "activity.enableCallbacks", false, @@ -61,8 +67,10 @@ type Config struct { BlobSizeLimitWarn dynamicconfig.IntPropertyFnWithNamespaceFilter BreakdownMetricsByTaskQueue dynamicconfig.TypedPropertyFnWithTaskQueueFilter[bool] EnableCallbacks dynamicconfig.BoolPropertyFnWithNamespaceFilter + EnableActivityEagerDispatchCheck dynamicconfig.BoolPropertyFnWithNamespaceFilter EnabledCallbackKinds dynamicconfig.TypedPropertyFnWithNamespaceFilter[[]callbacks.Kind] Enabled dynamicconfig.BoolPropertyFnWithNamespaceFilter + EnableEagerStart dynamicconfig.BoolPropertyFnWithNamespaceFilter EnableStandaloneActivityOperatorCommands dynamicconfig.BoolPropertyFnWithNamespaceFilter LongPollBuffer dynamicconfig.DurationPropertyFnWithNamespaceFilter LongPollTimeout dynamicconfig.DurationPropertyFnWithNamespaceFilter @@ -82,9 +90,11 @@ func ConfigProvider(dc *dynamicconfig.Collection) *Config { BlobSizeLimitWarn: dynamicconfig.BlobSizeLimitWarn.Get(dc), BreakdownMetricsByTaskQueue: dynamicconfig.MetricsBreakdownByTaskQueue.Get(dc), DefaultActivityRetryPolicy: dynamicconfig.DefaultActivityRetryPolicy.Get(dc), + EnableActivityEagerDispatchCheck: dynamicconfig.EnableActivityEagerDispatchCheck.Get(dc), EnableCallbacks: EnableCallbacks.Get(dc), EnabledCallbackKinds: EnabledCallbackKinds.Get(dc), Enabled: Enabled.Get(dc), + EnableEagerStart: EnableEagerStart.Get(dc), EnableStandaloneActivityOperatorCommands: EnableStandaloneActivityOperatorCommands.Get(dc), LongPollBuffer: LongPollBuffer.Get(dc), LongPollTimeout: LongPollTimeout.Get(dc), diff --git a/chasm/lib/activity/frontend.go b/chasm/lib/activity/frontend.go index 695e24dc242..3541fb4f0f4 100644 --- a/chasm/lib/activity/frontend.go +++ b/chasm/lib/activity/frontend.go @@ -391,7 +391,10 @@ func (h *frontendHandler) validateAndPopulateStartRequest( if req.GetStartDelay().AsDuration() > 0 && !h.config.StartDelayEnabled(req.GetNamespace()) { return nil, serviceerror.NewInvalidArgument("start_delay is not enabled for this namespace") } - // TODO(saa): when eager start is supported, deny it if start delay > 0 (same as workflow behavior). + if req.GetRequestEagerExecution() && + (!h.config.EnableEagerStart(req.GetNamespace()) || req.GetStartDelay().AsDuration() > 0) { + req.RequestEagerExecution = false + } opts := activityOptionsFromStartRequest(req) err := ValidateAndNormalizeStandaloneActivity( diff --git a/chasm/lib/activity/frontend_test.go b/chasm/lib/activity/frontend_test.go index 1e105364bf4..bbe3abe6764 100644 --- a/chasm/lib/activity/frontend_test.go +++ b/chasm/lib/activity/frontend_test.go @@ -11,6 +11,7 @@ import ( "go.temporal.io/api/workflowservice/v1" "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/log" + "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/namespace" "google.golang.org/protobuf/types/known/durationpb" ) @@ -215,3 +216,58 @@ func TestRequestIdStableAcrossRetries(t *testing.T) { }) }) } + +func TestEagerStartFallback(t *testing.T) { + newHandler := func(eagerEnabled bool) *frontendHandler { + return &frontendHandler{ + config: &Config{ + BlobSizeLimitError: defaultBlobSizeLimitError, + BlobSizeLimitWarn: defaultBlobSizeLimitWarn, + DefaultActivityRetryPolicy: getDefaultRetrySettings, + EnableEagerStart: dynamicconfig.GetBoolPropertyFnFilteredByNamespace(eagerEnabled), + MaxIDLengthLimit: func() int { return defaultMaxIDLengthLimit }, + MaxUserMetadataDetailsSize: defaultMaxUserMetadataDetailsSize, + MaxUserMetadataSummarySize: defaultMaxUserMetadataSummarySize, + StartDelayEnabled: dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true), + }, + linkValidator: newLinkValidator( + defaultMaxLinksPerRequest, + func(string) int { return 2000 }, + defaultLinkMaxSize, + ), + logger: log.NewNoopLogger(), + metricsHandler: metrics.NoopMetricsHandler, + } + } + + newRequest := func() *workflowservice.StartActivityExecutionRequest { + return &workflowservice.StartActivityExecutionRequest{ + Namespace: "test-namespace", + RequestEagerExecution: true, + ActivityId: "test-activity", + ActivityType: &commonpb.ActivityType{Name: "test-type"}, + TaskQueue: &taskqueuepb.TaskQueue{Name: "test-queue"}, + StartToCloseTimeout: durationpb.New(time.Minute), + } + } + + t.Run("enabled", func(t *testing.T) { + req, err := newHandler(true).validateAndPopulateStartRequest(context.Background(), newRequest(), "test-namespace-id") + require.NoError(t, err) + require.True(t, req.GetRequestEagerExecution()) + }) + + t.Run("namespace disabled", func(t *testing.T) { + req, err := newHandler(false).validateAndPopulateStartRequest(context.Background(), newRequest(), "test-namespace-id") + require.NoError(t, err) + require.False(t, req.GetRequestEagerExecution()) + }) + + t.Run("start delay", func(t *testing.T) { + req := newRequest() + req.StartDelay = durationpb.New(time.Minute) + req, err := newHandler(true).validateAndPopulateStartRequest(context.Background(), req, "test-namespace-id") + require.NoError(t, err) + require.False(t, req.GetRequestEagerExecution()) + }) +} 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 9018035b715..4fe000fabfa 100644 --- a/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go +++ b/chasm/lib/activity/gen/activitypb/v1/activity_state.pb.go @@ -317,8 +317,10 @@ type ActivityState struct { LastResetRequestId string `protobuf:"bytes,22,opt,name=last_reset_request_id,json=lastResetRequestId,proto3" json:"last_reset_request_id,omitempty"` // Used to de-dupe update requests. LastUpdateOptionsRequestId string `protobuf:"bytes,23,opt,name=last_update_options_request_id,json=lastUpdateOptionsRequestId,proto3" json:"last_update_options_request_id,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + // Indicates that eager execution was accepted for this activity's first task. + EagerExecutionAccepted bool `protobuf:"varint,24,opt,name=eager_execution_accepted,json=eagerExecutionAccepted,proto3" json:"eager_execution_accepted,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ActivityState) Reset() { @@ -512,6 +514,13 @@ func (x *ActivityState) GetLastUpdateOptionsRequestId() string { return "" } +func (x *ActivityState) GetEagerExecutionAccepted() bool { + if x != nil { + return x.EagerExecutionAccepted + } + 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"` @@ -1254,7 +1263,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/activity/v1/message.proto\x1a$temporal/api/common/v1/message.proto\x1a(temporal/api/deployment/v1/message.proto\x1a$temporal/api/enums/v1/workflow.proto\x1a%temporal/api/failure/v1/message.proto\x1a#temporal/api/nexus/v1/message.proto\x1a'temporal/api/sdk/v1/user_metadata.proto\x1a'temporal/api/taskqueue/v1/message.proto\"\xb9\r\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/activity/v1/message.proto\x1a$temporal/api/common/v1/message.proto\x1a(temporal/api/deployment/v1/message.proto\x1a$temporal/api/enums/v1/workflow.proto\x1a%temporal/api/failure/v1/message.proto\x1a#temporal/api/nexus/v1/message.proto\x1a'temporal/api/sdk/v1/user_metadata.proto\x1a'temporal/api/taskqueue/v1/message.proto\"\xf3\r\n" + "\rActivityState\x12I\n" + "\ractivity_type\x18\x01 \x01(\v2$.temporal.api.common.v1.ActivityTypeR\factivityType\x12C\n" + "\n" + @@ -1281,7 +1290,8 @@ const file_temporal_server_chasm_lib_activity_proto_v1_activity_state_proto_rawD "\x15reset_restore_options\x18\x14 \x01(\bR\x13resetRestoreOptions\x125\n" + "\x17last_unpause_request_id\x18\x15 \x01(\tR\x14lastUnpauseRequestId\x121\n" + "\x15last_reset_request_id\x18\x16 \x01(\tR\x12lastResetRequestId\x12B\n" + - "\x1elast_update_options_request_id\x18\x17 \x01(\tR\x1alastUpdateOptionsRequestId\"\xa7\x01\n" + + "\x1elast_update_options_request_id\x18\x17 \x01(\tR\x1alastUpdateOptionsRequestId\x128\n" + + "\x18eager_execution_accepted\x18\x18 \x01(\bR\x16eagerExecutionAccepted\"\xa7\x01\n" + "\x13ActivityCancelState\x12\x1d\n" + "\n" + "request_id\x18\x01 \x01(\tR\trequestId\x12=\n" + diff --git a/chasm/lib/activity/handler.go b/chasm/lib/activity/handler.go index b6165e92ba1..f45799c4e1f 100644 --- a/chasm/lib/activity/handler.go +++ b/chasm/lib/activity/handler.go @@ -11,11 +11,13 @@ import ( "go.temporal.io/server/api/historyservice/v1" "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" + "go.temporal.io/server/common" "go.temporal.io/server/common/contextutil" "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/namespace" commonnexus "go.temporal.io/server/common/nexus" + "go.temporal.io/server/common/resource" ) var ( @@ -38,6 +40,7 @@ type handler struct { linkValidator *linkValidator logger log.Logger metricsHandler metrics.Handler + matchingClient resource.MatchingClient namespaceRegistry namespace.Registry } @@ -47,6 +50,7 @@ func newHandler( linkValidator *linkValidator, metricsHandler metrics.Handler, logger log.Logger, + matchingClient resource.MatchingClient, namespaceRegistry namespace.Registry, ) *handler { return &handler{ @@ -55,6 +59,7 @@ func newHandler( linkValidator: linkValidator, logger: logger, metricsHandler: metricsHandler, + matchingClient: matchingClient, namespaceRegistry: namespaceRegistry, } } @@ -64,18 +69,22 @@ func newHandler( // matching being delivered to a worker poll request. func (h *handler) StartActivityExecution(ctx context.Context, req *activitypb.StartActivityExecutionRequest) (*activitypb.StartActivityExecutionResponse, error) { frontendReq := req.GetFrontendRequest() + var eagerTaskData *eagerActivityTaskData - reusePolicy, ok := businessIDReusePolicyMap[frontendReq.GetIdReusePolicy()] - if !ok { - return nil, serviceerror.NewInvalidArgumentf("unsupported ID reuse policy: %v", frontendReq.GetIdReusePolicy()) - } - - conflictPolicy, ok := businessIDConflictPolicyMap[frontendReq.GetIdConflictPolicy()] - if !ok { - return nil, serviceerror.NewInvalidArgumentf("unsupported ID conflict policy: %v", frontendReq.GetIdConflictPolicy()) + reusePolicy, conflictPolicy, err := businessIDPolicies(frontendReq) + if err != nil { + return nil, err } maxCallbacks := h.config.MaxCallbacksPerExecution(frontendReq.GetNamespace()) + // TODO: Reconcile the Matching grant when the subsequent start does not create an eager task. + if frontendReq.GetRequestEagerExecution() && + h.config.EnableActivityEagerDispatchCheck(frontendReq.GetNamespace()) && + !h.grantEagerActivityDispatch(ctx, req.GetNamespaceId(), frontendReq) { + // Matching grants are best-effort. On a denial or any Matching failure, schedule the + // activity normally instead of failing the start request. + frontendReq.RequestEagerExecution = false + } result, err := chasm.StartExecution( ctx, @@ -84,27 +93,11 @@ func (h *handler) StartActivityExecution(ctx context.Context, req *activitypb.St BusinessID: frontendReq.GetActivityId(), }, func(mutableContext chasm.MutableContext, request *workflowservice.StartActivityExecutionRequest) (*Activity, error) { - newActivity, err := NewStandaloneActivity(mutableContext, request) - if err != nil { - return nil, err - } - - if cbs := request.GetCompletionCallbacks(); len(cbs) > 0 { - if err := newActivity.addCompletionCallbacks(mutableContext, request.GetRequestId(), cbs, maxCallbacks); err != nil { - return nil, err - } - } - if len(request.GetLinks()) > 0 { - if err := newActivity.attachLinks(mutableContext, request.GetLinks(), request.GetRequestId(), h.linkValidator, frontendReq.GetNamespace()); err != nil { - return nil, err - } - } - - err = TransitionScheduled.Apply(newActivity, mutableContext, nil) + newActivity, newEagerTaskData, err := h.newActivityForStartExecution(mutableContext, request, req.GetNamespaceId(), maxCallbacks) if err != nil { return nil, err } - + eagerTaskData = newEagerTaskData return newActivity, nil }, frontendReq, @@ -114,9 +107,12 @@ func (h *handler) StartActivityExecution(ctx context.Context, req *activitypb.St if err != nil { if alreadyStartedErr, ok := errors.AsType[*chasm.ExecutionAlreadyStartedError](err); ok { - return nil, serviceerror.NewActivityExecutionAlreadyStarted("activity execution already started", alreadyStartedErr.CurrentRequestID, alreadyStartedErr.CurrentRunID) + return nil, serviceerror.NewActivityExecutionAlreadyStarted( + "activity execution already started", + alreadyStartedErr.CurrentRequestID, + alreadyStartedErr.CurrentRunID, + ) } - return nil, err } @@ -130,6 +126,15 @@ func (h *handler) StartActivityExecution(ctx context.Context, req *activitypb.St ) } + var eagerTask *workflowservice.PollActivityTaskQueueResponse + if result.Created && eagerTaskData != nil { + eagerTask, err = eagerTaskData.response(result.ExecutionRef) + if err != nil { + return nil, err + } + metrics.ActivityEagerExecutionCounter.With(eagerActivityMetricsHandler(h.metricsHandler, h.config, frontendReq)).Record(1) + } + // Apply on_conflict_options to an existing activity. // TODO: Use chasm.UpdateWithStartExecution to avoid a second transaction once the engine supports BusinessIDConflictPolicyFail in the updateFn path. cbs := frontendReq.GetCompletionCallbacks() @@ -176,8 +181,9 @@ func (h *handler) StartActivityExecution(ctx context.Context, req *activitypb.St return &activitypb.StartActivityExecutionResponse{ FrontendResponse: &workflowservice.StartActivityExecutionResponse{ - RunId: result.ExecutionKey.RunID, - Started: result.Created, + RunId: result.ExecutionKey.RunID, + Started: result.Created, + EagerActivityTask: eagerTask, Link: &commonpb.Link{ Variant: &commonpb.Link_Activity_{ Activity: &commonpb.Link_Activity{ @@ -187,11 +193,87 @@ func (h *handler) StartActivityExecution(ctx context.Context, req *activitypb.St }, }, }, - // EagerTask: TODO when supported, need to call the same code that would handle the HandleStarted API }, }, nil } +func (h *handler) newActivityForStartExecution( + mutableContext chasm.MutableContext, + request *workflowservice.StartActivityExecutionRequest, + namespaceID string, + maxCallbacks int, +) (*Activity, *eagerActivityTaskData, error) { + newActivity, err := NewStandaloneActivity(mutableContext, request) + if err != nil { + return nil, nil, err + } + + if cbs := request.GetCompletionCallbacks(); len(cbs) > 0 { + if err := newActivity.addCompletionCallbacks(mutableContext, request.GetRequestId(), cbs, maxCallbacks); err != nil { + return nil, nil, err + } + } + if len(request.GetLinks()) > 0 { + if err := newActivity.attachLinks(mutableContext, request.GetLinks(), request.GetRequestId(), h.linkValidator, request.GetNamespace()); err != nil { + return nil, nil, err + } + } + + if request.GetRequestEagerExecution() { + err = TransitionEagerStarted.Apply(newActivity, mutableContext, eagerStartEvent{ + requestID: request.GetRequestId(), + identity: request.GetIdentity(), + }) + } else { + err = TransitionScheduled.Apply(newActivity, mutableContext, nil) + } + if err != nil { + return nil, nil, err + } + if !request.GetRequestEagerExecution() { + return newActivity, nil, nil + } + + eagerTaskData, err := newActivity.eagerActivityTaskData(mutableContext, eagerActivityTaskRequest{ + namespaceID: namespaceID, + namespace: request.GetNamespace(), + requestID: request.GetRequestId(), + }) + if err != nil { + return nil, nil, err + } + return newActivity, eagerTaskData, nil +} + +func businessIDPolicies(request *workflowservice.StartActivityExecutionRequest) (chasm.BusinessIDReusePolicy, chasm.BusinessIDConflictPolicy, error) { + reusePolicy, ok := businessIDReusePolicyMap[request.GetIdReusePolicy()] + if !ok { + return 0, 0, serviceerror.NewInvalidArgumentf("unsupported ID reuse policy: %v", request.GetIdReusePolicy()) + } + + conflictPolicy, ok := businessIDConflictPolicyMap[request.GetIdConflictPolicy()] + if !ok { + return 0, 0, serviceerror.NewInvalidArgumentf("unsupported ID conflict policy: %v", request.GetIdConflictPolicy()) + } + + return reusePolicy, conflictPolicy, nil +} + +func (h *handler) grantEagerActivityDispatch( + ctx context.Context, + namespaceID string, + request *workflowservice.StartActivityExecutionRequest, +) bool { + return common.GrantEagerDispatch( + ctx, + h.matchingClient, + namespaceID, + request.GetTaskQueue().GetName(), + request.GetPriority(), + nil, + ) +} + // DescribeActivityExecution queries current activity state, optionally as a long-poll that waits // for any state change. When used to long-poll, it returns an empty non-error response on context // deadline expiry, to indicate that the state being waited for was not reached. Callers should diff --git a/chasm/lib/activity/handler_test.go b/chasm/lib/activity/handler_test.go new file mode 100644 index 00000000000..d4f3a571a23 --- /dev/null +++ b/chasm/lib/activity/handler_test.go @@ -0,0 +1,77 @@ +package activity + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/server/api/matchingservice/v1" + "go.temporal.io/server/api/matchingservicemock/v1" + "go.uber.org/mock/gomock" + "google.golang.org/grpc" + "google.golang.org/protobuf/proto" +) + +func TestGrantEagerActivityDispatch(t *testing.T) { + priority := &commonpb.Priority{PriorityKey: 2, FairnessKey: "fairness-key"} + request := &workflowservice.StartActivityExecutionRequest{ + Namespace: "namespace", + TaskQueue: &taskqueuepb.TaskQueue{Name: "activity-task-queue"}, + Priority: priority, + } + + testCases := []struct { + name string + response *matchingservice.GrantEagerDispatchResponse + err error + granted bool + }{ + { + name: "granted", + response: &matchingservice.GrantEagerDispatchResponse{Items: []*matchingservice.GrantEagerDispatchResponse_Item{ + {GrantedCount: 1}, + }}, + granted: true, + }, + { + name: "denied", + response: &matchingservice.GrantEagerDispatchResponse{Items: []*matchingservice.GrantEagerDispatchResponse_Item{ + {}, + }}, + }, + { + name: "matching error", + err: errors.New("matching unavailable"), + }, + { + name: "missing response item", + response: &matchingservice.GrantEagerDispatchResponse{}, + }, + } + + for _, test := range testCases { + t.Run(test.name, func(t *testing.T) { + controller := gomock.NewController(t) + matchingClient := matchingservicemock.NewMockMatchingServiceClient(controller) + matchingClient.EXPECT().GrantEagerDispatch(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, actual *matchingservice.GrantEagerDispatchRequest, _ ...grpc.CallOption) (*matchingservice.GrantEagerDispatchResponse, error) { + require.Equal(t, "namespace-id", actual.GetNamespaceId()) + require.Equal(t, "activity-task-queue", actual.GetTaskQueuePartition().GetTaskQueue()) + require.Equal(t, enumspb.TASK_QUEUE_TYPE_ACTIVITY, actual.GetTaskQueuePartition().GetTaskQueueType()) + require.Len(t, actual.GetItems(), 1) + require.EqualValues(t, 1, actual.GetItems()[0].GetCount()) + require.True(t, proto.Equal(priority, actual.GetItems()[0].GetPriority())) + return test.response, test.err + }, + ) + + handler := &handler{matchingClient: matchingClient} + require.Equal(t, test.granted, handler.grantEagerActivityDispatch(context.Background(), "namespace-id", request)) + }) + } +} diff --git a/chasm/lib/activity/metrics.go b/chasm/lib/activity/metrics.go index 6e680062a9a..20213d28147 100644 --- a/chasm/lib/activity/metrics.go +++ b/chasm/lib/activity/metrics.go @@ -7,6 +7,7 @@ import ( commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" failurepb "go.temporal.io/api/failure/v1" + "go.temporal.io/api/workflowservice/v1" "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" "go.temporal.io/server/common/metrics" @@ -28,6 +29,22 @@ func (a *Activity) taskScheduleToStartMetricsHandler(ctx chasm.Context) metrics. )) } +func eagerActivityMetricsHandler( + handler metrics.Handler, + config *Config, + request *workflowservice.StartActivityExecutionRequest, +) metrics.Handler { + namespaceName := request.GetNamespace() + taskQueueName := request.GetTaskQueue().GetName() + return metrics.GetPerActivityEagerExecutionScope( + handler, + namespaceName, + taskQueueName, + config.BreakdownMetricsByTaskQueue(namespaceName, taskQueueName, enumspb.TASK_QUEUE_TYPE_ACTIVITY), + "StartActivityExecution", + ) +} + // baseMetricsHandler adds only the operation tag. func (a *Activity) baseMetricsHandler(ctx chasm.Context, operation string) metrics.Handler { return ctx.MetricsHandler().WithTags(metrics.OperationTag(operation)) diff --git a/chasm/lib/activity/model/model.go b/chasm/lib/activity/model/model.go index a104b18a1c8..c27d0b4af45 100644 --- a/chasm/lib/activity/model/model.go +++ b/chasm/lib/activity/model/model.go @@ -33,6 +33,10 @@ type Outcome struct { // Initial is the state of a newly created Activity. func Initial(cfg Config) AbstractState { s := AbstractState{Status: Scheduled, AttemptCount: 1} + if cfg.InitialAttemptStarted { + s.Status = Started + return s + } if cfg.HasStartDelay { s.Dispatchability = StartDelayPending } diff --git a/chasm/lib/activity/model/model_test.go b/chasm/lib/activity/model/model_test.go index 39187e104d5..54c82e1a407 100644 --- a/chasm/lib/activity/model/model_test.go +++ b/chasm/lib/activity/model/model_test.go @@ -12,6 +12,27 @@ func TestInitial(t *testing.T) { require.Equal(t, AbstractState{Status: Scheduled, AttemptCount: 1}, Initial(Config{HasScheduleToClose: true})) } +func TestEagerInitialAttempt(t *testing.T) { + cfg := Config{ + InitialAttemptStarted: true, + HasScheduleToClose: true, + HasScheduleToStart: true, + HasHeartbeat: true, + } + initial := Initial(cfg) + require.Equal(t, AbstractState{Status: Started, AttemptCount: 1}, initial) + require.False(t, FindsTask(initial), "an eager task is already started and must not be dispatched again") + require.False(t, Possible(cfg, initial, ScheduleToStartElapsesType), "an eager first attempt has no schedule-to-start window") + require.True(t, Possible(cfg, initial, ScheduleToCloseElapsesType), "schedule-to-close still protects an eager first attempt") + require.True(t, Possible(cfg, initial, StartToCloseElapsesType), "start-to-close still protects an eager first attempt") + require.True(t, Possible(cfg, initial, HeartbeatElapsesType), "heartbeat still protects an eager first attempt") + + retry := Transition(cfg, initial, FailRetryably).Next + require.Equal(t, AbstractState{Status: Scheduled, AttemptCount: 2, Dispatchability: BackoffPending}, retry) + retry = Transition(cfg, retry, BackoffElapses).Next + require.True(t, FindsTask(retry), "an eager first attempt's retry must return to normal dispatch") +} + func TestPollFromScheduledStarts(t *testing.T) { out := Transition(Config{}, Initial(Config{}), Event{Type: PollType}) require.Equal(t, NoError, out.Reject) @@ -40,7 +61,10 @@ func TestPauseWhileStartedIsPauseRequested(t *testing.T) { // backedOffRetry returns a Scheduled state with a pending retry backoff (attempt 2), reached the way // a worker would: poll the first attempt, then fail it retryably. func backedOffRetry(t require.TestingT, cfg Config) AbstractState { - started := Transition(cfg, Initial(cfg), Event{Type: PollType}).Next + started := Initial(cfg) + if started.Status != Started { + started = Transition(cfg, started, Event{Type: PollType}).Next + } s := Transition(cfg, started, Event{Type: RespondFailedType, Failure: &Failure{}}).Next require.Equal(t, Scheduled, s.Status, "a retryable failure must schedule a retry") require.Equal(t, BackoffPending, s.Dispatchability, "a retry must wait for its backoff") diff --git a/chasm/lib/activity/model/vocabulary.go b/chasm/lib/activity/model/vocabulary.go index 02e646a090e..e00c19ca20e 100644 --- a/chasm/lib/activity/model/vocabulary.go +++ b/chasm/lib/activity/model/vocabulary.go @@ -43,11 +43,13 @@ type AbstractState struct { // Config is what the model needs to know about an activity's configuration: which options are set, // and what their durations imply regarding retries. type Config struct { - HasScheduleToClose bool - HasScheduleToStart bool - HasHeartbeat bool - HasStartDelay bool - MaxAttempts int32 // 0 = unlimited + // InitialAttemptStarted means the first attempt was delivered eagerly and is already running. + InitialAttemptStarted bool + HasScheduleToClose bool + HasScheduleToStart bool + HasHeartbeat bool + HasStartDelay bool + MaxAttempts int32 // 0 = unlimited // NonRetryableTimeouts are the timeout elapses whose failure the retry policy refuses to retry, // so that the timeout closes the activity instead of scheduling another attempt. diff --git a/chasm/lib/activity/proto/v1/activity_state.proto b/chasm/lib/activity/proto/v1/activity_state.proto index 700792e6dbd..4b36a1f5e4b 100644 --- a/chasm/lib/activity/proto/v1/activity_state.proto +++ b/chasm/lib/activity/proto/v1/activity_state.proto @@ -162,6 +162,9 @@ message ActivityState { // Used to de-dupe update requests. string last_update_options_request_id = 23; + + // Indicates that eager execution was accepted for this activity's first task. + bool eager_execution_accepted = 24; } message ActivityCancelState { diff --git a/chasm/lib/activity/responses.go b/chasm/lib/activity/responses.go index f581efb26db..eeb2549587e 100644 --- a/chasm/lib/activity/responses.go +++ b/chasm/lib/activity/responses.go @@ -7,14 +7,124 @@ import ( commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" failurepb "go.temporal.io/api/failure/v1" + nexuspb "go.temporal.io/api/nexus/v1" "go.temporal.io/api/workflowservice/v1" "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" + "go.temporal.io/server/common" + "go.temporal.io/server/common/tasktoken" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" ) +type eagerActivityTaskRequest struct { + namespaceID string + namespace string + requestID string +} + +type eagerActivityTaskData struct { + namespaceID string + namespace string + runID string + activityType *commonpb.ActivityType + activityID string + header *commonpb.Header + input *commonpb.Payloads + heartbeatDetails *commonpb.Payloads + scheduledTime *timestamppb.Timestamp + currentAttemptScheduledTime *timestamppb.Timestamp + startedTime *timestamppb.Timestamp + attempt int32 + scheduleToCloseTimeout *durationpb.Duration + startToCloseTimeout *durationpb.Duration + heartbeatTimeout *durationpb.Duration + retryPolicy *commonpb.RetryPolicy + priority *commonpb.Priority + propagatedNexusSerializationContext *nexuspb.PropagatedSerializationContext + activityAttemptStartedStamp int32 +} + +func (a *Activity) eagerActivityTaskData( + ctx chasm.Context, + request eagerActivityTaskRequest, +) (*eagerActivityTaskData, error) { + attempt := a.LastAttempt.Get(ctx) + if !a.hasAttemptInProgress() || attempt.GetCount() != 1 || attempt.GetStartRequestId() != request.requestID { + return nil, nil + } + + key := ctx.ExecutionKey() + requestData := a.RequestData.Get(ctx) + lastHeartbeat, _ := a.LastHeartbeat.TryGet(ctx) + + // This data is used to construct the eager response after the transaction has completed, so + // it must not retain mutable state owned by the transaction. + return &eagerActivityTaskData{ + namespaceID: request.namespaceID, + namespace: request.namespace, + runID: key.RunID, + activityType: common.CloneProto(a.GetActivityType()), + activityID: key.BusinessID, + header: common.CloneProto(requestData.GetHeader()), + input: common.CloneProto(requestData.GetInput()), + heartbeatDetails: common.CloneProto(lastHeartbeat.GetDetails()), + scheduledTime: common.CloneProto(a.GetScheduleTime()), + currentAttemptScheduledTime: common.CloneProto(a.dispatchTimeForAttempt(attempt)), + startedTime: common.CloneProto(attempt.GetStartedTime()), + attempt: attempt.GetCount(), + scheduleToCloseTimeout: common.CloneProto(a.GetScheduleToCloseTimeout()), + startToCloseTimeout: common.CloneProto(a.GetStartToCloseTimeout()), + heartbeatTimeout: common.CloneProto(a.GetHeartbeatTimeout()), + retryPolicy: common.CloneProto(a.GetRetryPolicy()), + priority: common.CloneProto(a.GetPriority()), + propagatedNexusSerializationContext: common.CloneProto(requestData.GetPropagatedNexusSerializationContext()), + activityAttemptStartedStamp: attempt.GetStartedStamp(), + }, nil +} + +func (d *eagerActivityTaskData) response(componentRef []byte) (*workflowservice.PollActivityTaskQueueResponse, error) { + token, err := tasktoken.NewSerializer().Serialize(tasktoken.NewActivityTaskToken( + d.namespaceID, + "", + d.runID, + 0, + d.activityID, + d.activityType.GetName(), + d.attempt, + nil, + 0, + 0, + componentRef, + d.activityAttemptStartedStamp, + )) + if err != nil { + return nil, err + } + + return &workflowservice.PollActivityTaskQueueResponse{ + TaskToken: token, + WorkflowNamespace: d.namespace, + ActivityType: d.activityType, + ActivityId: d.activityID, + Header: d.header, + Input: d.input, + HeartbeatDetails: d.heartbeatDetails, + ScheduledTime: d.scheduledTime, + CurrentAttemptScheduledTime: d.currentAttemptScheduledTime, + StartedTime: d.startedTime, + Attempt: d.attempt, + ScheduleToCloseTimeout: d.scheduleToCloseTimeout, + StartToCloseTimeout: d.startToCloseTimeout, + HeartbeatTimeout: d.heartbeatTimeout, + RetryPolicy: d.retryPolicy, + Priority: d.priority, + PropagatedNexusSerializationContext: d.propagatedNexusSerializationContext, + ActivityRunId: d.runID, + }, nil +} + // Projection of activity state onto the API response protos. // InternalStatusToAPIStatus converts internal activity execution status to API status. @@ -113,6 +223,7 @@ func (a *Activity) buildActivityExecutionInfo( CanceledReason: a.CancelState.GetReason(), CloseTime: closeTime, CurrentRetryInterval: a.currentRetryInterval(ctx, attempt), + EagerExecutionAccepted: a.GetEagerExecutionAccepted(), ExecutionDuration: executionDuration, ExecutionTime: timestamppb.New(a.firstDispatchTime()), ExpirationTime: expirationTime, diff --git a/chasm/lib/activity/responses_test.go b/chasm/lib/activity/responses_test.go index 106b3e69d27..403f6053443 100644 --- a/chasm/lib/activity/responses_test.go +++ b/chasm/lib/activity/responses_test.go @@ -5,11 +5,15 @@ import ( "time" "github.com/stretchr/testify/require" + commonpb "go.temporal.io/api/common/v1" deploymentpb "go.temporal.io/api/deployment/v1" + nexuspb "go.temporal.io/api/nexus/v1" "go.temporal.io/api/workflowservice/v1" "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" "go.temporal.io/server/common/testing/protorequire" + "google.golang.org/protobuf/types/known/durationpb" + "google.golang.org/protobuf/types/known/timestamppb" ) func TestBuildActivityExecutionInfo_IncludeLastDeploymentVersion(t *testing.T) { @@ -43,3 +47,121 @@ func TestBuildActivityExecutionInfo_IncludeLastDeploymentVersion(t *testing.T) { BuildId: "test-build-1", }, resp.FrontendResponse.GetInfo().GetLastDeploymentVersion()) } + +func TestBuildActivityExecutionInfo_EagerExecutionAccepted(t *testing.T) { + ctx := &chasm.MockMutableContext{ + MockContext: chasm.MockContext{ + HandleNow: func(chasm.Component) time.Time { return time.Unix(0, 0) }, + }, + } + + for _, test := range []struct { + name string + accepted bool + }{ + {name: "accepted", accepted: true}, + {name: "not accepted"}, + } { + t.Run(test.name, func(t *testing.T) { + activity := &Activity{ + ActivityState: &activitypb.ActivityState{ + Status: activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED, + EagerExecutionAccepted: test.accepted, + }, + LastAttempt: chasm.NewDataField(ctx, &activitypb.ActivityAttemptState{}), + RequestData: chasm.NewDataField(ctx, &activitypb.ActivityRequestData{}), + Visibility: chasm.NewComponentField(ctx, &chasm.Visibility{}), + } + + info := activity.buildActivityExecutionInfo(ctx, &workflowservice.DescribeActivityExecutionRequest{}) + require.Equal(t, test.accepted, info.GetEagerExecutionAccepted()) + }) + } +} + +func TestEagerActivityTaskDataSnapshotsState(t *testing.T) { + ctx := &chasm.MockMutableContext{ + MockContext: chasm.MockContext{ + HandleExecutionKey: func() chasm.ExecutionKey { + return chasm.ExecutionKey{BusinessID: "activity-id", RunID: "run-id"} + }, + }, + } + + activityType := &commonpb.ActivityType{Name: "activity-type"} + header := &commonpb.Header{Fields: map[string]*commonpb.Payload{"header": {Data: []byte("header")}}} + input := &commonpb.Payloads{Payloads: []*commonpb.Payload{{Data: []byte("input")}}} + heartbeatDetails := &commonpb.Payloads{Payloads: []*commonpb.Payload{{Data: []byte("heartbeat")}}} + scheduledTime := timestamppb.New(time.Unix(10, 0)) + dispatchTime := timestamppb.New(time.Unix(11, 0)) + startedTime := timestamppb.New(time.Unix(12, 0)) + scheduleToCloseTimeout := durationpb.New(time.Minute) + startToCloseTimeout := durationpb.New(2 * time.Minute) + heartbeatTimeout := durationpb.New(3 * time.Minute) + retryPolicy := &commonpb.RetryPolicy{MaximumAttempts: 2} + priority := &commonpb.Priority{PriorityKey: 1} + propagatedNexusSerializationContext := &nexuspb.PropagatedSerializationContext{Endpoint: "endpoint", Service: "service", Operation: "operation"} + + activity := &Activity{ + ActivityState: &activitypb.ActivityState{ + ActivityType: activityType, + Status: activitypb.ACTIVITY_EXECUTION_STATUS_STARTED, + ScheduleTime: scheduledTime, + ScheduleToCloseTimeout: scheduleToCloseTimeout, + StartToCloseTimeout: startToCloseTimeout, + HeartbeatTimeout: heartbeatTimeout, + RetryPolicy: retryPolicy, + Priority: priority, + }, + LastAttempt: chasm.NewDataField(ctx, &activitypb.ActivityAttemptState{ + Count: 1, + StartRequestId: "request-id", + StartedTime: startedTime, + DispatchTime: dispatchTime, + }), + LastHeartbeat: chasm.NewDataField(ctx, &activitypb.ActivityHeartbeatState{Details: heartbeatDetails}), + RequestData: chasm.NewDataField(ctx, &activitypb.ActivityRequestData{ + Header: header, + Input: input, + PropagatedNexusSerializationContext: propagatedNexusSerializationContext, + }), + } + + data, err := activity.eagerActivityTaskData(ctx, eagerActivityTaskRequest{ + namespaceID: "namespace-id", + namespace: "namespace", + requestID: "request-id", + }) + require.NoError(t, err) + require.NotNil(t, data) + + activityType.Name = "changed-activity-type" + header.Fields["header"].Data = []byte("changed-header") + input.Payloads[0].Data = []byte("changed-input") + heartbeatDetails.Payloads[0].Data = []byte("changed-heartbeat") + scheduledTime.Seconds++ + dispatchTime.Seconds++ + startedTime.Seconds++ + scheduleToCloseTimeout.Seconds++ + startToCloseTimeout.Seconds++ + heartbeatTimeout.Seconds++ + retryPolicy.MaximumAttempts++ + priority.PriorityKey++ + propagatedNexusSerializationContext.Endpoint = "changed-endpoint" + + response, err := data.response([]byte("component-ref")) + require.NoError(t, err) + require.Equal(t, "activity-type", response.GetActivityType().GetName()) + require.Equal(t, []byte("header"), response.GetHeader().GetFields()["header"].GetData()) + require.Equal(t, []byte("input"), response.GetInput().GetPayloads()[0].GetData()) + require.Equal(t, []byte("heartbeat"), response.GetHeartbeatDetails().GetPayloads()[0].GetData()) + require.Equal(t, int64(10), response.GetScheduledTime().GetSeconds()) + require.Equal(t, int64(11), response.GetCurrentAttemptScheduledTime().GetSeconds()) + require.Equal(t, int64(12), response.GetStartedTime().GetSeconds()) + require.Equal(t, int64(time.Minute/time.Second), response.GetScheduleToCloseTimeout().GetSeconds()) + require.Equal(t, int64(2*time.Minute/time.Second), response.GetStartToCloseTimeout().GetSeconds()) + require.Equal(t, int64(3*time.Minute/time.Second), response.GetHeartbeatTimeout().GetSeconds()) + require.EqualValues(t, 2, response.GetRetryPolicy().GetMaximumAttempts()) + require.EqualValues(t, 1, response.GetPriority().GetPriorityKey()) + require.Equal(t, "endpoint", response.GetPropagatedNexusSerializationContext().GetEndpoint()) +} diff --git a/chasm/lib/activity/statemachine.go b/chasm/lib/activity/statemachine.go index a9cd9be2d07..b7f1c9ffaab 100644 --- a/chasm/lib/activity/statemachine.go +++ b/chasm/lib/activity/statemachine.go @@ -35,6 +35,11 @@ func (a *Activity) SetStateMachineState(state activitypb.ActivityExecutionStatus } type ( + eagerStartEvent struct { + requestID string + identity string + } + rescheduleEvent struct { retryInterval time.Duration retryIntervalSource activitypb.ActivityRetryIntervalSource @@ -98,7 +103,22 @@ var TransitionScheduled = chasm.NewTransition( }, activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED, func(a *Activity, ctx chasm.MutableContext, _ any) error { - return a.applyScheduled(ctx) + return a.applyScheduled(ctx, true) + }, +) + +// TransitionEagerStarted atomically schedules and starts the first attempt without creating a +// Matching dispatch task. The task itself is returned by StartActivityExecution. +var TransitionEagerStarted = chasm.NewTransition( + []activitypb.ActivityExecutionStatus{ + activitypb.ACTIVITY_EXECUTION_STATUS_UNSPECIFIED, + }, + activitypb.ACTIVITY_EXECUTION_STATUS_STARTED, + func(a *Activity, ctx chasm.MutableContext, event eagerStartEvent) error { + if err := a.applyScheduled(ctx, false); err != nil { + return err + } + return a.applyEagerStarted(ctx, event) }, ) @@ -346,7 +366,7 @@ var TransitionResetAttemptFailedToScheduled = chasm.NewTransition( // The methods below are transition bodies, invoked from the transitions above once the // source state has been validated. Do not call them from outside this file. -func (a *Activity) applyScheduled(ctx chasm.MutableContext) error { +func (a *Activity) applyScheduled(ctx chasm.MutableContext, dispatch bool) error { attempt := a.LastAttempt.Get(ctx) attempt.Count++ @@ -357,7 +377,7 @@ func (a *Activity) applyScheduled(ctx chasm.MutableContext) error { dispatchTime := a.firstDispatchTime() attempt.DispatchTime = timestamppb.New(dispatchTime) - if timeout := a.GetScheduleToStartTimeout().AsDuration(); timeout > 0 { + if timeout := a.GetScheduleToStartTimeout().AsDuration(); dispatch && timeout > 0 { ctx.AddTask( a, chasm.TaskAttributes{ @@ -378,15 +398,45 @@ func (a *Activity) applyScheduled(ctx chasm.MutableContext) error { &activitypb.ScheduleToCloseTimeoutTask{Stamp: a.GetScheduleToCloseStamp()}) } - dispatchAttrs := chasm.TaskAttributes{} - if dispatchTime.After(a.ScheduleTime.AsTime()) { - dispatchAttrs.ScheduledTime = dispatchTime + if dispatch { + dispatchAttrs := chasm.TaskAttributes{} + if dispatchTime.After(a.ScheduleTime.AsTime()) { + dispatchAttrs.ScheduledTime = dispatchTime + } + ctx.AddTask( + a, + dispatchAttrs, + a.newActivityDispatchTask(ctx)) + } + + return nil +} + +func (a *Activity) applyEagerStarted(ctx chasm.MutableContext, event eagerStartEvent) error { + attempt := a.LastAttempt.Get(ctx) + a.EagerExecutionAccepted = true + attempt.StartedTime = timestamppb.New(ctx.Now(a)) + attempt.StartedStamp = attempt.GetStamp() + if a.FirstAttemptStartedTime == nil { + a.FirstAttemptStartedTime = attempt.GetStartedTime() } + attempt.StartRequestId = event.requestID + attempt.LastWorkerIdentity = event.identity + attempt.SdkName = ctx.RequestHeader(headers.ClientNameHeaderName) + attempt.SdkVersion = ctx.RequestHeader(headers.ClientVersionHeaderName) + // TODO: Populate LastDeploymentVersion when standalone eager start supports worker versioning. + startTime := attempt.GetStartedTime().AsTime() ctx.AddTask( a, - dispatchAttrs, - a.newActivityDispatchTask(ctx)) + chasm.TaskAttributes{ScheduledTime: startTime.Add(a.GetStartToCloseTimeout().AsDuration())}, + &activitypb.StartToCloseTimeoutTask{Stamp: attempt.GetStamp()}) + if heartbeatTimeout := a.GetHeartbeatTimeout().AsDuration(); heartbeatTimeout > 0 { + ctx.AddTask( + a, + chasm.TaskAttributes{ScheduledTime: startTime.Add(heartbeatTimeout)}, + &activitypb.HeartbeatTimeoutTask{Stamp: attempt.GetStamp()}) + } return nil } diff --git a/chasm/lib/activity/statemachine_test.go b/chasm/lib/activity/statemachine_test.go index 1583c7b7ff5..c4f8a44529f 100644 --- a/chasm/lib/activity/statemachine_test.go +++ b/chasm/lib/activity/statemachine_test.go @@ -196,6 +196,50 @@ func TestTransitionScheduled(t *testing.T) { } } +func TestTransitionEagerStarted(t *testing.T) { + ctx := &chasm.MockMutableContext{ + MockContext: chasm.MockContext{ + HandleNow: func(chasm.Component) time.Time { return defaultTime }, + GoCtx: headers.SetVersionsForTests(context.Background(), temporal.SDKVersion, headers.ClientNameGoSDK, "", ""), + }, + } + attempt := &activitypb.ActivityAttemptState{} + activity := &Activity{ + ActivityState: &activitypb.ActivityState{ + ActivityType: &commonpb.ActivityType{Name: "test-activity-type"}, + ScheduleTime: timestamppb.New(defaultTime), + ScheduleToCloseTimeout: durationpb.New(defaultScheduleToCloseTimeout), + ScheduleToStartTimeout: durationpb.New(defaultScheduleToStartTimeout), + StartToCloseTimeout: durationpb.New(defaultStartToCloseTimeout), + HeartbeatTimeout: durationpb.New(time.Minute), + Status: activitypb.ACTIVITY_EXECUTION_STATUS_UNSPECIFIED, + TaskQueue: &taskqueuepb.TaskQueue{Name: "test-task-queue"}, + }, + LastAttempt: chasm.NewDataField(ctx, attempt), + Outcome: chasm.NewDataField(ctx, &activitypb.ActivityOutcome{}), + } + + err := TransitionEagerStarted.Apply(activity, ctx, eagerStartEvent{ + requestID: "start-request-id", + identity: "starter", + }) + require.NoError(t, err) + require.Equal(t, activitypb.ACTIVITY_EXECUTION_STATUS_STARTED, activity.GetStatus()) + require.True(t, activity.GetEagerExecutionAccepted()) + require.EqualValues(t, 1, attempt.GetCount()) + require.EqualValues(t, 1, attempt.GetStamp()) + require.EqualValues(t, 1, attempt.GetStartedStamp()) + require.Equal(t, "start-request-id", attempt.GetStartRequestId()) + require.Equal(t, "starter", attempt.GetLastWorkerIdentity()) + require.Equal(t, headers.ClientNameGoSDK, attempt.GetSdkName()) + require.Equal(t, temporal.SDKVersion, attempt.GetSdkVersion()) + + require.Len(t, ctx.Tasks, 3) + require.IsType(t, &activitypb.ScheduleToCloseTimeoutTask{}, ctx.Tasks[0].Payload) + require.IsType(t, &activitypb.StartToCloseTimeoutTask{}, ctx.Tasks[1].Payload) + require.IsType(t, &activitypb.HeartbeatTimeoutTask{}, ctx.Tasks[2].Payload) +} + func TestTransitionRescheduled(t *testing.T) { testCases := []struct { name string diff --git a/chasm/lib/activity/validator_test.go b/chasm/lib/activity/validator_test.go index c3d70500614..105eab0b80b 100644 --- a/chasm/lib/activity/validator_test.go +++ b/chasm/lib/activity/validator_test.go @@ -837,16 +837,16 @@ func TestValidateStandaloneUserMetadata(t *testing.T) { }, } - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { + for _, testCase := range tests { + t.Run(testCase.name, func(t *testing.T) { h := &frontendHandler{ config: &Config{ BlobSizeLimitError: defaultBlobSizeLimitError, BlobSizeLimitWarn: defaultBlobSizeLimitWarn, DefaultActivityRetryPolicy: getDefaultRetrySettings, MaxIDLengthLimit: func() int { return defaultMaxIDLengthLimit }, - MaxUserMetadataSummarySize: func(string) int { return test.summaryLimit }, - MaxUserMetadataDetailsSize: func(string) int { return test.detailsLimit }, + MaxUserMetadataSummarySize: func(string) int { return testCase.summaryLimit }, + MaxUserMetadataDetailsSize: func(string) int { return testCase.detailsLimit }, }, linkValidator: newLinkValidator( defaultMaxLinksPerRequest, @@ -861,16 +861,16 @@ func TestValidateStandaloneUserMetadata(t *testing.T) { ActivityType: &commonpb.ActivityType{Name: defaultActivityType}, TaskQueue: &taskqueuepb.TaskQueue{Name: defaultTaskQueue}, StartToCloseTimeout: durationpb.New(10 * time.Second), - UserMetadata: test.metadata, + UserMetadata: testCase.metadata, } _, err := h.validateAndPopulateStartRequest(t.Context(), req, namespace.ID(defaultNamespaceID)) - if test.errContains == "" { + if testCase.errContains == "" { require.NoError(t, err) return } var invalidArgument *serviceerror.InvalidArgument require.ErrorAs(t, err, &invalidArgument) - require.ErrorContains(t, err, test.errContains) + require.ErrorContains(t, err, testCase.errContains) }) } } diff --git a/common/metrics/activity.go b/common/metrics/activity.go index ca1b377828c..9f9f571b657 100644 --- a/common/metrics/activity.go +++ b/common/metrics/activity.go @@ -27,3 +27,20 @@ func GetPerActivityScope( WorkflowTypeTag(workflowType), ) } + +// GetPerActivityEagerExecutionScope returns the tags shared by eager activity execution metrics. +func GetPerActivityEagerExecutionScope( + handler Handler, + namespaceName string, + taskQueueName string, + taskQueueBreakdown bool, + operation string, +) Handler { + return GetPerTaskQueueFamilyScope( + handler, + namespaceName, + tqid.UnsafeTaskQueueFamily(namespaceName, taskQueueName), + taskQueueBreakdown, + OperationTag(operation), + ) +} diff --git a/common/util.go b/common/util.go index 5af646bbf78..b8728ff4cb6 100644 --- a/common/util.go +++ b/common/util.go @@ -15,8 +15,10 @@ import ( enumspb "go.temporal.io/api/enums/v1" "go.temporal.io/api/serviceerror" "go.temporal.io/api/workflowservice/v1" + deploymentspb "go.temporal.io/server/api/deployment/v1" "go.temporal.io/server/api/historyservice/v1" "go.temporal.io/server/api/matchingservice/v1" + taskqueuespb "go.temporal.io/server/api/taskqueue/v1" workflowspb "go.temporal.io/server/api/workflow/v1" "go.temporal.io/server/common/backoff" "go.temporal.io/server/common/log" @@ -691,6 +693,34 @@ func CloneProto[T proto.Message](v T) T { return proto.Clone(v).(T) } +// GrantEagerDispatch asks Matching whether it may eagerly dispatch one activity task. +func GrantEagerDispatch( + ctx context.Context, + client matchingservice.MatchingServiceClient, + namespaceID string, + taskQueue string, + priority *commonpb.Priority, + version *deploymentspb.WorkerDeploymentVersion, +) bool { + if client == nil { + return false + } + + response, err := client.GrantEagerDispatch(ctx, &matchingservice.GrantEagerDispatchRequest{ + NamespaceId: namespaceID, + TaskQueuePartition: &taskqueuespb.TaskQueuePartition{ + TaskQueue: taskQueue, + TaskQueueType: enumspb.TASK_QUEUE_TYPE_ACTIVITY, + }, + Items: []*matchingservice.GrantEagerDispatchRequest_Item{{ + Count: 1, + Priority: priority, + Version: version, + }}, + }) + return err == nil && len(response.GetItems()) == 1 && response.GetItems()[0].GetGrantedCount() == 1 +} + func CloneProtoMap[K comparable, T proto.Message](src map[K]T) map[K]T { if src == nil { return nil diff --git a/common/util_test.go b/common/util_test.go index b67c46cd477..ce9ed0a3844 100644 --- a/common/util_test.go +++ b/common/util_test.go @@ -648,3 +648,14 @@ func TestCreateHistoryStartWorkflowRequestPayloads(t *testing.T) { // ensure the original request object is unmodified require.Equal(t, startRequestClone, startRequest) } + +func TestGrantEagerDispatchWithoutClient(t *testing.T) { + require.False(t, GrantEagerDispatch( + context.Background(), + nil, + "namespace-id", + "task-queue", + nil, + nil, + )) +} diff --git a/go.mod b/go.mod index 132c905d9e4..8dd315ca873 100644 --- a/go.mod +++ b/go.mod @@ -70,7 +70,7 @@ require ( go.opentelemetry.io/otel/sdk/metric v1.44.0 go.opentelemetry.io/otel/trace v1.44.0 go.opentelemetry.io/proto/otlp v1.10.0 - go.temporal.io/api v1.63.7-0.20261006170555-05ff170da40c + go.temporal.io/api v1.63.7-0.20261008212609-555b97e92c27 go.temporal.io/auto-scaled-workers v0.0.0-20260928183627-719679297cbe go.temporal.io/sdk v1.49.0 go.uber.org/fx v1.24.0 diff --git a/go.sum b/go.sum index f5502173afa..6e4c67dc81f 100644 --- a/go.sum +++ b/go.sum @@ -479,8 +479,8 @@ go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0 h1:R go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0/go.mod h1:I89cynRj8y+383o7tEQVg2SVA6SRgDVIouWPUVXjx0U= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN6Z8jsUeYv8J0lXRvygALXIzsmAeCcZE0= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4= -go.temporal.io/api v1.63.7-0.20261006170555-05ff170da40c h1:obFBvsi8+5laUnGUFB5s3SpNh62CyW+C8A7pTJXgEV8= -go.temporal.io/api v1.63.7-0.20261006170555-05ff170da40c/go.mod h1:acM0I9WPuYg8W3Pd9jOZvEgi7mRUttUQ4+e7fowKVnM= +go.temporal.io/api v1.63.7-0.20261008212609-555b97e92c27 h1:QI4R6Tvtx0s+cCgMnFFsTUuKsw+xDpR/Z9mCx8gs6Os= +go.temporal.io/api v1.63.7-0.20261008212609-555b97e92c27/go.mod h1:acM0I9WPuYg8W3Pd9jOZvEgi7mRUttUQ4+e7fowKVnM= go.temporal.io/auto-scaled-workers v0.0.0-20260928183627-719679297cbe h1:BlshPqVI1uh/AsdfGPPjfKc37HEZdGfcmvx+uW+rbe8= go.temporal.io/auto-scaled-workers v0.0.0-20260928183627-719679297cbe/go.mod h1:6Q4guHwlbJldBOZwvc201E0+Jz9Tly1L1Db6C7RJuLw= go.temporal.io/sdk v1.49.0 h1:CtGI0BUe/SCo3eoqTwuWWtXKueii9GBVus7KrKKH1Vo= diff --git a/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go b/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go index 489e46c3e92..2549cee7ac7 100644 --- a/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go +++ b/service/history/api/respondworkflowtaskcompleted/workflow_task_completed_handler.go @@ -22,7 +22,6 @@ import ( "go.temporal.io/api/workflowservice/v1" "go.temporal.io/server/api/historyservice/v1" "go.temporal.io/server/api/matchingservice/v1" - taskqueuespb "go.temporal.io/server/api/taskqueue/v1" chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/common" "go.temporal.io/server/common/backoff" @@ -595,26 +594,15 @@ func (handler *workflowTaskCompletedHandler) grantEagerActivityDispatch( ctx context.Context, attr *commandpb.ScheduleActivityTaskCommandAttributes, ) bool { - if handler.matchingClient == nil { - return false - } - executionInfo := handler.mutableState.GetExecutionInfo() - response, err := handler.matchingClient.GrantEagerDispatch(ctx, &matchingservice.GrantEagerDispatchRequest{ - NamespaceId: executionInfo.GetNamespaceId(), - TaskQueuePartition: &taskqueuespb.TaskQueuePartition{ - TaskQueue: attr.GetTaskQueue().GetName(), - TaskQueueType: enumspb.TASK_QUEUE_TYPE_ACTIVITY, - }, - Items: []*matchingservice.GrantEagerDispatchRequest_Item{ - { - Count: 1, - Priority: attr.GetPriority(), - Version: worker_versioning.DeploymentVersionFromDeployment(handler.workflowTaskDeployment), - }, - }, - }) - return err == nil && len(response.GetItems()) == 1 && response.GetItems()[0].GetGrantedCount() == 1 + return common.GrantEagerDispatch( + ctx, + handler.matchingClient, + executionInfo.GetNamespaceId(), + attr.GetTaskQueue().GetName(), + attr.GetPriority(), + worker_versioning.DeploymentVersionFromDeployment(handler.workflowTaskDeployment), + ) } func (handler *workflowTaskCompletedHandler) handlePostCommandEagerExecuteActivity( @@ -711,7 +699,17 @@ func (handler *workflowTaskCompletedHandler) handlePostCommandEagerExecuteActivi }, } metrics.ActivityEagerExecutionCounter.With( - workflow.GetPerTaskQueueFamilyScope(handler.metricsHandler, handler.mutableState.GetNamespaceEntry().Name(), ai.TaskQueue, handler.config), + metrics.GetPerActivityEagerExecutionScope( + handler.metricsHandler, + handler.mutableState.GetNamespaceEntry().Name().String(), + ai.TaskQueue, + handler.config.BreakdownMetricsByTaskQueue( + handler.mutableState.GetNamespaceEntry().Name().String(), + ai.TaskQueue, + enumspb.TASK_QUEUE_TYPE_WORKFLOW, + ), + metrics.HistoryRespondWorkflowTaskCompletedScope, + ), ).Record(1) return func(resp *historyservice.RespondWorkflowTaskCompletedResponse) error { diff --git a/tests/activity_driver.go b/tests/activity_driver.go index 360c42f23dc..a563bcd30ee 100644 --- a/tests/activity_driver.go +++ b/tests/activity_driver.go @@ -41,6 +41,7 @@ import ( // always send start-to-close, defaulted long enough not to fire. The other timeouts are simply // absent when unset. type activityConfig struct { + EagerStart bool MaxAttempts int32 // RetryPolicy MaximumAttempts; 0 = unlimited RetryInterval time.Duration // RetryPolicy InitialInterval; 0 => activityShortRetryInterval BackoffCoefficient float64 // RetryPolicy BackoffCoefficient; 0 => 1.0 (constant interval) @@ -157,12 +158,13 @@ type activityTerminalOutcome struct { // ways the options alone settle that no retry can follow. Deriving it means the two cannot disagree. func (c activityConfig) modelConfig() model.Config { return model.Config{ - MaxAttempts: c.MaxAttempts, - HasStartDelay: c.StartDelay > 0, - HasScheduleToClose: c.ScheduleToClose > 0, - HasScheduleToStart: c.ScheduleToStart > 0, - HasHeartbeat: c.HeartbeatTimeout > 0, - NonRetryableTimeouts: c.nonRetryableTimeouts(), + InitialAttemptStarted: c.EagerStart, + MaxAttempts: c.MaxAttempts, + HasStartDelay: c.StartDelay > 0, + HasScheduleToClose: c.ScheduleToClose > 0, + HasScheduleToStart: c.ScheduleToStart > 0, + HasHeartbeat: c.HeartbeatTimeout > 0, + NonRetryableTimeouts: c.nonRetryableTimeouts(), // In reality, the server decides this based on how long the attempt took, and any // NextRetryDelay sent by the worker. The model cannot express that. Instead it supports a // fixed retry delay and the calculation here pretends that the attempt duration is zero. diff --git a/tests/activity_parity_test.go b/tests/activity_parity_test.go index bae153f3ec8..3245e2a0504 100644 --- a/tests/activity_parity_test.go +++ b/tests/activity_parity_test.go @@ -4,6 +4,7 @@ package tests import ( "errors" + "slices" "strings" "testing" "time" @@ -17,6 +18,8 @@ import ( "go.temporal.io/server/chasm/lib/activity/model" "go.temporal.io/server/common" "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/metrics" + "go.temporal.io/server/common/metrics/metricstest" "go.temporal.io/server/common/retrypolicy" "go.temporal.io/server/common/testing/parallelsuite" "go.temporal.io/server/service/history/consts" @@ -41,10 +44,102 @@ func newActivityParityEnv(t *testing.T) *testcore.TestEnv { cluster := env.GetTestCluster() cluster.OverrideDynamicConfig(t, dynamicconfig.EnableChasm, nsValues(true)) cluster.OverrideDynamicConfig(t, activity.Enabled, nsValues(true)) + cluster.OverrideDynamicConfig(t, dynamicconfig.EnableActivityEagerExecution, nsValues(true)) + cluster.OverrideDynamicConfig(t, activity.EnableEagerStart, nsValues(true)) cluster.OverrideDynamicConfig(t, activity.EnableStandaloneActivityOperatorCommands, nsValues(true)) return env } +func (s *activityParityTestSuite) TestEagerActivityStartParity() { + env := newActivityParityEnv(s.T()) + cfg := activityConfig{ + EagerStart: true, + MaxAttempts: 2, + RetryInterval: activityShortDispatchDelay, + } + + // Both implementations return the first task already started. A retry of that task must then + // return to the ordinary Matching dispatch path before its second attempt starts. + trace := []model.Event{ + model.Heartbeat, + model.FailRetryably, + model.BackoffElapses, + model.Poll, + model.Complete, + } + + s.Run("WorkflowActivity", func(s *activityParityTestSuite) { + newWFADriver(s.T(), env, cfg).driveTrace(s.T(), trace) + }) + s.Run("StandaloneActivity", func(s *activityParityTestSuite) { + newSAADriver(s.T(), env, cfg).driveTrace(s.T(), trace) + }) +} + +func (s *activityParityTestSuite) TestEagerActivityMetricParity() { + type captureResult struct { + recordings []*metricstest.CapturedRecording + taskQueue string + } + + capture := func(t *testing.T, standalone, breakdownByTaskQueue bool) captureResult { + env := newActivityParityEnv(t) + env.GetTestCluster().OverrideDynamicConfig(t, dynamicconfig.MetricsBreakdownByTaskQueue, + []dynamicconfig.ConstrainedValue{{ + Constraints: dynamicconfig.Constraints{Namespace: env.Namespace().String()}, + Value: breakdownByTaskQueue, + }}) + metricCapture := env.StartNamespaceMetricCapture() + + cfg := activityConfig{EagerStart: true} + var taskQueue string + if standalone { + taskQueue = newSAADriver(t, env, cfg).start(t, cfg).taskQueue + } else { + taskQueue = newWFADriver(t, env, cfg).start(t, cfg).taskQueue + } + + return captureResult{ + recordings: metricCapture.Metric(metrics.ActivityEagerExecutionCounter.Name()), + taskQueue: taskQueue, + } + } + + tagKeys := func(recording *metricstest.CapturedRecording) []string { + keys := make([]string, 0, len(recording.Tags)) + for key := range recording.Tags { + keys = append(keys, key) + } + slices.Sort(keys) + return keys + } + + for _, breakdownByTaskQueue := range []bool{true, false} { + name := "TaskQueueBreakdownEnabled" + if !breakdownByTaskQueue { + name = "TaskQueueBreakdownDisabled" + } + s.Run(name, func(s *activityParityTestSuite) { + t := s.T() + wfa := capture(t, false, breakdownByTaskQueue) + saa := capture(t, true, breakdownByTaskQueue) + require.Len(t, wfa.recordings, 1) + require.Len(t, saa.recordings, 1) + require.Equal(t, tagKeys(wfa.recordings[0]), tagKeys(saa.recordings[0]), + "WFA and SAA eager activity metrics must use the same label keys") + + expectedTaskQueue := "__omitted__" + if breakdownByTaskQueue { + require.Equal(t, wfa.taskQueue, wfa.recordings[0].Tags["taskqueue"]) + require.Equal(t, saa.taskQueue, saa.recordings[0].Tags["taskqueue"]) + } else { + require.Equal(t, expectedTaskQueue, wfa.recordings[0].Tags["taskqueue"]) + require.Equal(t, expectedTaskQueue, saa.recordings[0].Tags["taskqueue"]) + } + }) + } +} + func assertActivityTaskNotCancelRequested(t *testing.T, err error) { var invalidArgumentErr *serviceerror.InvalidArgument require.ErrorAs(t, err, &invalidArgumentErr) diff --git a/tests/activity_standalone_driver.go b/tests/activity_standalone_driver.go index d9ee39457a7..6807b4ffa0f 100644 --- a/tests/activity_standalone_driver.go +++ b/tests/activity_standalone_driver.go @@ -118,8 +118,15 @@ func (d *saaDriver) start(t require.TestingT, cfg activityConfig) *saaHandle { id := fmt.Sprintf("%s-%d", d.activityIDPrefix, d.numStarted) resp, err := d.env.FrontendClient().StartActivityExecution(d.testContext(), d.startRequest(cfg, id, id)) require.NoError(t, err) + state := activityDriverState{cfg: cfg} + if cfg.EagerStart { + task := resp.GetEagerActivityTask() + require.NotNil(t, task, "eager standalone activity start did not return a task") + state.token = task.GetTaskToken() + state.startedAttempt = task.GetAttempt() + } return &saaHandle{ - activityDriverState: activityDriverState{cfg: cfg}, + activityDriverState: state, model: newActivityModel(cfg), d: d, activityID: id, @@ -137,6 +144,7 @@ func (d *saaDriver) startRequest(c activityConfig, activityID, taskQueue string) } return &workflowservice.StartActivityExecutionRequest{ Namespace: d.env.Namespace().String(), + RequestEagerExecution: c.EagerStart, ActivityId: activityID, ActivityType: d.env.Tv().ActivityType(), Identity: d.env.Tv().ClientIdentity(), diff --git a/tests/activity_standalone_test.go b/tests/activity_standalone_test.go index e4b14d28a50..1416042d68e 100644 --- a/tests/activity_standalone_test.go +++ b/tests/activity_standalone_test.go @@ -1220,6 +1220,264 @@ func (s *standaloneActivityTestSuite) TestStart() { }) } +func (s *standaloneActivityTestSuite) TestEagerStartResponseAndNoRedelivery() { + env := s.newTestEnv(testcore.WithDynamicConfig(dynamicconfig.EnableActivityEagerDispatchCheck, true)) + t := s.T() + ctx := metadata.NewOutgoingContext(s.Context(), metadata.Pairs( + headers.ClientNameHeaderName, headers.ClientNameGoSDK, + headers.ClientVersionHeaderName, temporal.SDKVersion, + )) + activityID := testcore.RandomizeStr(t.Name()) + taskQueue := testcore.RandomizeStr(t.Name()) + requestID := testcore.RandomizeStr(t.Name()) + request := &workflowservice.StartActivityExecutionRequest{ + Namespace: env.Namespace().String(), + RequestEagerExecution: true, + ActivityId: activityID, + ActivityType: env.Tv().ActivityType(), + Identity: defaultIdentity, + Input: defaultInput, + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue}, + StartToCloseTimeout: durationpb.New(defaultStartToCloseTimeout), + RequestId: requestID, + } + + first, err := env.FrontendClient().StartActivityExecution(ctx, request) + require.NoError(t, err) + require.True(t, first.GetStarted()) + require.NotNil(t, first.GetEagerActivityTask()) + require.Equal(t, activityID, first.GetEagerActivityTask().GetActivityId()) + require.Equal(t, first.GetRunId(), first.GetEagerActivityTask().GetActivityRunId()) + require.Nil(t, first.GetEagerActivityTask().GetWorkflowExecution()) + require.EqualValues(t, 1, first.GetEagerActivityTask().GetAttempt()) + + describe, err := env.FrontendClient().DescribeActivityExecution(ctx, &workflowservice.DescribeActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: first.GetRunId(), + }) + require.NoError(t, err) + require.Equal(t, headers.ClientNameGoSDK, describe.GetInfo().GetSdkName()) + require.Equal(t, temporal.SDKVersion, describe.GetInfo().GetSdkVersion()) + require.True(t, describe.GetInfo().GetEagerExecutionAccepted()) + + retry, err := env.FrontendClient().StartActivityExecution(ctx, request) + require.NoError(t, err) + require.False(t, retry.GetStarted()) + require.Nil(t, retry.GetEagerActivityTask()) + + useExisting := common.CloneProto(request) + useExisting.RequestId = testcore.RandomizeStr(t.Name()) + useExisting.IdConflictPolicy = enumspb.ACTIVITY_ID_CONFLICT_POLICY_USE_EXISTING + conflict, err := env.FrontendClient().StartActivityExecution(ctx, useExisting) + require.NoError(t, err) + require.False(t, conflict.GetStarted()) + require.Nil(t, conflict.GetEagerActivityTask()) + + _, err = env.FrontendClient().RespondActivityTaskCompleted(ctx, &workflowservice.RespondActivityTaskCompletedRequest{ + Namespace: env.Namespace().String(), + TaskToken: first.GetEagerActivityTask().GetTaskToken(), + Result: defaultResult, + Identity: defaultIdentity, + }) + require.NoError(t, err) + env.validateCompletion(ctx, t, activityID, first.GetRunId(), defaultIdentity) + +} + +func (s *standaloneActivityTestSuite) TestEagerStartFallsBackWhenMatchingDenies() { + env := s.newTestEnv( + testcore.WithDynamicConfig(dynamicconfig.EnableActivityEagerDispatchCheck, true), + testcore.WithDynamicConfig(dynamicconfig.MatchingBacklogNegligibleAge, time.Duration(0)), + testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueReadPartitions, 1), + testcore.WithDynamicConfig(dynamicconfig.MatchingNumTaskqueueWritePartitions, 1), + ) + t := s.T() + ctx := s.Context() + taskQueue := testcore.RandomizeStr(t.Name()) + backloggedActivityID := testcore.RandomizeStr(t.Name() + "-backlogged") + eagerActivityID := testcore.RandomizeStr(t.Name() + "-eager") + + start := func(activityID string, eager bool, priority int32) *workflowservice.StartActivityExecutionResponse { + response, err := env.FrontendClient().StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{ + Namespace: env.Namespace().String(), + RequestEagerExecution: eager, + ActivityId: activityID, + ActivityType: env.Tv().ActivityType(), + Identity: defaultIdentity, + Input: defaultInput, + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue}, + Priority: &commonpb.Priority{PriorityKey: priority}, + StartToCloseTimeout: durationpb.New(defaultStartToCloseTimeout), + RequestId: testcore.RandomizeStr(t.Name()), + }) + require.NoError(t, err) + require.True(t, response.GetStarted()) + return response + } + + backlogged := start(backloggedActivityID, false, 1) + await.RequireTruef(t, func() bool { + response, err := env.FrontendClient().DescribeTaskQueue(ctx, &workflowservice.DescribeTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue}, + TaskQueueType: enumspb.TASK_QUEUE_TYPE_ACTIVITY, + ReportStats: true, + }) + return err == nil && response.GetStats().GetApproximateBacklogCount() >= 1 + }, 10*time.Second, 100*time.Millisecond, "backlogged standalone activity never reached Matching") + + eager := start(eagerActivityID, true, 5) + require.Nil(t, eager.GetEagerActivityTask(), "Matching denial must fall back to normal dispatch") + + for _, startedActivity := range []struct { + id string + runID string + }{ + {backloggedActivityID, backlogged.GetRunId()}, + {eagerActivityID, eager.GetRunId()}, + } { + describe, err := env.FrontendClient().DescribeActivityExecution(ctx, &workflowservice.DescribeActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: startedActivity.id, + RunId: startedActivity.runID, + }) + require.NoError(t, err) + require.False(t, describe.GetInfo().GetEagerExecutionAccepted()) + } + + for _, startedActivity := range []struct { + id string + runID string + }{ + {backloggedActivityID, backlogged.GetRunId()}, + {eagerActivityID, eager.GetRunId()}, + } { + task := env.pollActivityTaskAndValidate(ctx, t, startedActivity.id, taskQueue, startedActivity.runID) + _, err := env.FrontendClient().RespondActivityTaskCompleted(ctx, &workflowservice.RespondActivityTaskCompletedRequest{ + Namespace: env.Namespace().String(), + TaskToken: task.GetTaskToken(), + Result: defaultResult, + Identity: defaultIdentity, + }) + require.NoError(t, err) + env.validateCompletion(ctx, t, startedActivity.id, startedActivity.runID, defaultIdentity) + } +} + +func (s *standaloneActivityTestSuite) TestLostEagerActivityTaskResponseRecovery() { + env := s.newTestEnv(testcore.WithDynamicConfig(dynamicconfig.EnableActivityEagerDispatchCheck, false)) + t := s.T() + + startEagerActivity := func( + ctx context.Context, + t *testing.T, + activityID string, + taskQueue string, + startToCloseTimeout time.Duration, + heartbeatTimeout time.Duration, + scheduleToCloseTimeout time.Duration, + ) *workflowservice.StartActivityExecutionResponse { + response, err := env.FrontendClient().StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{ + Namespace: env.Namespace().String(), + RequestEagerExecution: true, + ActivityId: activityID, + ActivityType: env.Tv().ActivityType(), + Identity: defaultIdentity, + Input: defaultInput, + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue}, + ScheduleToCloseTimeout: durationpb.New(scheduleToCloseTimeout), + StartToCloseTimeout: durationpb.New(startToCloseTimeout), + HeartbeatTimeout: durationpb.New(heartbeatTimeout), + RetryPolicy: &commonpb.RetryPolicy{ + InitialInterval: durationpb.New(time.Second), + BackoffCoefficient: 1, + MaximumAttempts: 2, + }, + RequestId: testcore.RandomizeStr(t.Name()), + }) + require.NoError(t, err) + require.True(t, response.GetStarted()) + require.NotNil(t, response.GetEagerActivityTask()) + return response + } + + completeRetriedActivity := func( + ctx context.Context, + t *testing.T, + activityID string, + taskQueue string, + runID string, + ) { + task, err := env.pollActivityTaskQueue(ctx, taskQueue) + require.NoError(t, err) + require.Equal(t, activityID, task.GetActivityId()) + require.EqualValues(t, 2, task.GetAttempt(), "retry must use the normal task-queue poll path") + + _, err = env.FrontendClient().RespondActivityTaskCompleted(ctx, &workflowservice.RespondActivityTaskCompletedRequest{ + Namespace: env.Namespace().String(), + TaskToken: task.GetTaskToken(), + Result: defaultResult, + Identity: defaultIdentity, + }) + require.NoError(t, err) + + describe, err := env.FrontendClient().DescribeActivityExecution(ctx, &workflowservice.DescribeActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: runID, + IncludeOutcome: true, + }) + require.NoError(t, err) + require.Equal(t, enumspb.ACTIVITY_EXECUTION_STATUS_COMPLETED, describe.GetInfo().GetStatus()) + require.EqualValues(t, 2, describe.GetInfo().GetAttempt()) + protorequire.ProtoEqual(t, defaultResult, describe.GetOutcome().GetResult()) + } + + t.Run("StartToCloseTimeout", func(t *testing.T) { + ctx, cancel := context.WithTimeout(s.Context(), 20*time.Second) + defer cancel() + activityID := testcore.RandomizeStr(t.Name()) + taskQueue := testcore.RandomizeStr(t.Name()) + + response := startEagerActivity(ctx, t, activityID, taskQueue, time.Second, 0, time.Minute) + // Simulate a client that lost the eager response: it never responds using this token. + require.NotEmpty(t, response.GetEagerActivityTask().GetTaskToken()) + completeRetriedActivity(ctx, t, activityID, taskQueue, response.GetRunId()) + }) + + t.Run("HeartbeatTimeout", func(t *testing.T) { + ctx, cancel := context.WithTimeout(s.Context(), 20*time.Second) + defer cancel() + activityID := testcore.RandomizeStr(t.Name()) + taskQueue := testcore.RandomizeStr(t.Name()) + + response := startEagerActivity(ctx, t, activityID, taskQueue, time.Minute, time.Second, time.Minute) + // Simulate a client that lost the eager response: it never heartbeats or completes. + require.NotEmpty(t, response.GetEagerActivityTask().GetTaskToken()) + completeRetriedActivity(ctx, t, activityID, taskQueue, response.GetRunId()) + }) + + t.Run("ScheduleToCloseTimeout", func(t *testing.T) { + ctx, cancel := context.WithTimeout(s.Context(), 20*time.Second) + defer cancel() + activityID := testcore.RandomizeStr(t.Name()) + taskQueue := testcore.RandomizeStr(t.Name()) + + response := startEagerActivity(ctx, t, activityID, taskQueue, time.Minute, 0, time.Second) + // Schedule-to-close is the activity's terminal deadline, so a lost eager task must not retry. + require.NotEmpty(t, response.GetEagerActivityTask().GetTaskToken()) + + outcome, err := env.FrontendClient().PollActivityExecution(ctx, &workflowservice.PollActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: response.GetRunId(), + }) + require.NoError(t, err) + require.Equal(t, enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE, outcome.GetOutcome().GetFailure().GetTimeoutFailureInfo().GetTimeoutType()) + }) +} + func (s *standaloneActivityTestSuite) TestComplete() { env := s.newTestEnv() t := s.T() diff --git a/tests/activity_workflow_driver.go b/tests/activity_workflow_driver.go index 73917e5a2cd..0e088b6f8e2 100644 --- a/tests/activity_workflow_driver.go +++ b/tests/activity_workflow_driver.go @@ -16,6 +16,8 @@ import ( "github.com/google/uuid" "github.com/stretchr/testify/require" activitypb "go.temporal.io/api/activity/v1" + commandpb "go.temporal.io/api/command/v1" + commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" taskqueuepb "go.temporal.io/api/taskqueue/v1" workflowpb "go.temporal.io/api/workflow/v1" @@ -122,6 +124,9 @@ func (a *wfaHandle) awaitDispatchDelay(t testing.TB, e model.Event) { } func (d *wfaDriver) start(t *testing.T, cfg activityConfig) *wfaHandle { + if cfg.EagerStart { + return d.startEager(t, cfg) + } cfg.StartDelay = 0 // WFA does not support start delay, but SAA/WFA tests often share config wfTQ := testcore.RandomizeStr("wfa-wf") actTQ := testcore.RandomizeStr("wfa-act") @@ -158,6 +163,88 @@ func (d *wfaDriver) start(t *testing.T, cfg activityConfig) *wfaHandle { return a } +func (d *wfaDriver) startEager(t *testing.T, cfg activityConfig) *wfaHandle { + wfTQ := testcore.RandomizeStr("wfa-eager-wf") + actTQ := testcore.RandomizeStr("wfa-eager-act") + const actID = "act" + wfID := testcore.RandomizeStr("wfa-eager-run") + fc := d.env.FrontendClient() + + started, err := fc.StartWorkflowExecution(d.testContext(), &workflowservice.StartWorkflowExecutionRequest{ + Namespace: d.env.Namespace().String(), + WorkflowId: wfID, + WorkflowType: &commonpb.WorkflowType{Name: "eagerActivityDriver"}, + TaskQueue: &taskqueuepb.TaskQueue{Name: wfTQ}, + RequestId: uuid.NewString(), + }) + require.NoError(t, err) + + ctx, cancel := context.WithTimeout(d.testContext(), activityDriverTimeout) + defer cancel() + workflowTask, err := fc.PollWorkflowTaskQueue(ctx, &workflowservice.PollWorkflowTaskQueueRequest{ + Namespace: d.env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: wfTQ}, + Identity: d.env.Tv().WorkerIdentity(), + }) + require.NoError(t, err) + require.NotEmpty(t, workflowTask.GetTaskToken(), "workflow start did not produce a workflow task") + + activityTask, err := fc.RespondWorkflowTaskCompleted(d.testContext(), &workflowservice.RespondWorkflowTaskCompletedRequest{ + Namespace: d.env.Namespace().String(), + Identity: d.env.Tv().WorkerIdentity(), + TaskToken: workflowTask.GetTaskToken(), + Commands: []*commandpb.Command{{ + CommandType: enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK, + Attributes: &commandpb.Command_ScheduleActivityTaskCommandAttributes{ + ScheduleActivityTaskCommandAttributes: &commandpb.ScheduleActivityTaskCommandAttributes{ + ActivityId: actID, + ActivityType: d.env.Tv().ActivityType(), + TaskQueue: &taskqueuepb.TaskQueue{Name: actTQ}, + Input: payloads.EncodeString(activityInput), + StartToCloseTimeout: durationpb.New(cfg.startToClose()), + ScheduleToCloseTimeout: optionalDuration(cfg.ScheduleToClose), + ScheduleToStartTimeout: optionalDuration(cfg.ScheduleToStart), + HeartbeatTimeout: optionalDuration(cfg.HeartbeatTimeout), + RetryPolicy: &commonpb.RetryPolicy{ + InitialInterval: durationpb.New(cfg.retryInterval()), + BackoffCoefficient: cmp.Or(cfg.BackoffCoefficient, 1.0), + MaximumInterval: durationpb.New(cmp.Or(cfg.MaxRetryInterval, cfg.retryInterval())), + MaximumAttempts: cfg.MaxAttempts, + NonRetryableErrorTypes: cfg.NonRetryableErrorTypes, + }, + RequestEagerExecution: true, + }, + }, + }}, + }) + require.NoError(t, err) + require.Len(t, activityTask.GetActivityTasks(), 1, "eager workflow activity start did not return a task") + task := activityTask.GetActivityTasks()[0] + + t.Cleanup(func() { + if err := d.env.SdkClient().TerminateWorkflow(context.Background(), wfID, started.GetRunId(), "eager activity driver cleanup"); err != nil { + t.Logf("eager activity driver cleanup: %v", err) + } + }) + return &wfaHandle{ + activityDriverState: activityDriverState{cfg: cfg, token: task.GetTaskToken(), startedAttempt: task.GetAttempt()}, + model: newActivityModel(cfg), + d: d, + run: d.env.SdkClient().GetWorkflow(d.testContext(), wfID, started.GetRunId()), + workflowID: wfID, + runID: started.GetRunId(), + activityID: actID, + taskQueue: actTQ, + } +} + +func optionalDuration(value time.Duration) *durationpb.Duration { + if value == 0 { + return nil + } + return durationpb.New(value) +} + // wfaActivityParams is what the helper workflow needs to schedule the activity: the activity the // test described, and where to put it. type wfaActivityParams struct { diff --git a/tests/mixedbrain/go.mod b/tests/mixedbrain/go.mod index 66148655770..dbc0b120a9c 100644 --- a/tests/mixedbrain/go.mod +++ b/tests/mixedbrain/go.mod @@ -7,7 +7,7 @@ require ( github.com/siderolabs/grpc-proxy v0.5.2 github.com/stretchr/testify v1.11.1 github.com/temporalio/omes v0.0.0-20260915193711-9667582a3051 - go.temporal.io/api v1.63.7-0.20261006170555-05ff170da40c + go.temporal.io/api v1.63.7-0.20261008212609-555b97e92c27 go.temporal.io/server v0.0.0-00010101000000-000000000000 google.golang.org/grpc v1.83.2 google.golang.org/protobuf v1.36.11 diff --git a/tests/mixedbrain/go.sum b/tests/mixedbrain/go.sum index 1c71db50a7c..1d32f68d4ab 100644 --- a/tests/mixedbrain/go.sum +++ b/tests/mixedbrain/go.sum @@ -55,8 +55,8 @@ go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRk go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= -go.temporal.io/api v1.63.7-0.20261006170555-05ff170da40c h1:obFBvsi8+5laUnGUFB5s3SpNh62CyW+C8A7pTJXgEV8= -go.temporal.io/api v1.63.7-0.20261006170555-05ff170da40c/go.mod h1:acM0I9WPuYg8W3Pd9jOZvEgi7mRUttUQ4+e7fowKVnM= +go.temporal.io/api v1.63.7-0.20261008212609-555b97e92c27 h1:QI4R6Tvtx0s+cCgMnFFsTUuKsw+xDpR/Z9mCx8gs6Os= +go.temporal.io/api v1.63.7-0.20261008212609-555b97e92c27/go.mod h1:acM0I9WPuYg8W3Pd9jOZvEgi7mRUttUQ4+e7fowKVnM= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=