Add eager start support for standalone activities

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

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(

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

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