Add dynamic config to toggle standalone activity functionality

chasm/lib/activity/config.go

@@ -4,17 +4,24 @@ import (
 	"time"
 
 	"go.temporal.io/server/common/dynamicconfig"
+	"go.temporal.io/server/common/retrypolicy"
 )
 
 var (
+	Enabled = dynamicconfig.NewNamespaceBoolSetting(
+		"activity.enableStandalone",
+		false,
+		`Toggles standalone activity functionality on the server.`,
+	)
+
 	LongPollTimeout = dynamicconfig.NewNamespaceDurationSetting(
-		"chasm.activity.longPollTimeout",
+		"activity.longPollTimeout",
 		20*time.Second,
 		`Timeout for activity long-poll requests.`,
 	)
 
 	LongPollBuffer = dynamicconfig.NewNamespaceDurationSetting(
-		"chasm.activity.longPollBuffer",
+		"activity.longPollBuffer",
 		time.Second,
 		`A buffer used to adjust the activity long-poll timeouts.
  Specifically, activity long-poll requests are timed out at a time which leaves at least the buffer's duration
@@ -23,15 +30,25 @@ var (
 )
 
 type Config struct {
+	BlobSizeLimitError          dynamicconfig.IntPropertyFnWithNamespaceFilter
+	BlobSizeLimitWarn           dynamicconfig.IntPropertyFnWithNamespaceFilter
 	BreakdownMetricsByTaskQueue dynamicconfig.TypedPropertyFnWithTaskQueueFilter[bool]
-	LongPollTimeout             dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	Enabled                     dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	LongPollBuffer              dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	LongPollTimeout             dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	MaxIDLengthLimit            dynamicconfig.IntPropertyFn
+	DefaultActivityRetryPolicy  dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings]
 }
 
 func ConfigProvider(dc *dynamicconfig.Collection) *Config {
 	return &Config{
-		LongPollTimeout:             LongPollTimeout.Get(dc),
-		LongPollBuffer:              LongPollBuffer.Get(dc),
+		BlobSizeLimitError:          dynamicconfig.BlobSizeLimitError.Get(dc),
+		BlobSizeLimitWarn:           dynamicconfig.BlobSizeLimitWarn.Get(dc),
 		BreakdownMetricsByTaskQueue: dynamicconfig.MetricsBreakdownByTaskQueue.Get(dc),
+		DefaultActivityRetryPolicy:  dynamicconfig.DefaultActivityRetryPolicy.Get(dc),
+		Enabled:                     Enabled.Get(dc),
+		LongPollBuffer:              LongPollBuffer.Get(dc),
+		LongPollTimeout:             LongPollTimeout.Get(dc),
+		MaxIDLengthLimit:            dynamicconfig.MaxIDLengthLimit.Get(dc),
 	}
 }

chasm/lib/activity/frontend.go

@@ -10,14 +10,15 @@ import (
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1"
 	"go.temporal.io/server/common"
-	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/searchattribute"
 	"google.golang.org/protobuf/types/known/durationpb"
 )
 
+const StandaloneActivityDisabledError = "Standalone activity is disabled"
+
 type FrontendHandler interface {
 	StartActivityExecution(ctx context.Context, req *workflowservice.StartActivityExecutionRequest) (*workflowservice.StartActivityExecutionResponse, error)
 	DescribeActivityExecution(ctx context.Context, req *workflowservice.DescribeActivityExecutionRequest) (*workflowservice.DescribeActivityExecutionResponse, error)
@@ -27,12 +28,13 @@ type FrontendHandler interface {
 	ListActivityExecutions(context.Context, *workflowservice.ListActivityExecutionsRequest) (*workflowservice.ListActivityExecutionsResponse, error)
 	RequestCancelActivityExecution(context.Context, *workflowservice.RequestCancelActivityExecutionRequest) (*workflowservice.RequestCancelActivityExecutionResponse, error)
 	TerminateActivityExecution(context.Context, *workflowservice.TerminateActivityExecutionRequest) (*workflowservice.TerminateActivityExecutionResponse, error)
+	IsStandaloneActivityEnabled(namespaceName string) bool
 }
 
 type frontendHandler struct {
 	FrontendHandler
 	client            activitypb.ActivityServiceClient
-	dc                *dynamicconfig.Collection
+	config            *Config
 	logger            log.Logger
 	metricsHandler    metrics.Handler
 	namespaceRegistry namespace.Registry
@@ -43,7 +45,7 @@ type frontendHandler struct {
 // NewFrontendHandler creates a new FrontendHandler instance for processing activity frontend requests.
 func NewFrontendHandler(
 	client activitypb.ActivityServiceClient,
-	dc *dynamicconfig.Collection,
+	config *Config,
 	logger log.Logger,
 	metricsHandler metrics.Handler,
 	namespaceRegistry namespace.Registry,
@@ -52,7 +54,7 @@ func NewFrontendHandler(
 ) FrontendHandler {
 	return &frontendHandler{
 		client:            client,
-		dc:                dc,
+		config:            config,
 		logger:            logger,
 		metricsHandler:    metricsHandler,
 		namespaceRegistry: namespaceRegistry,
@@ -61,6 +63,11 @@ func NewFrontendHandler(
 	}
 }
 
+// IsStandaloneActivityEnabled checks if standalone activities are enabled for the given namespace
+func (h *frontendHandler) IsStandaloneActivityEnabled(namespaceName string) bool {
+	return h.config.Enabled(namespaceName)
+}
+
 // StartActivityExecution initiates a standalone activity execution in the specified namespace.
 // It validates the request, resolves the namespace ID, applies default configurations,
 // and forwards the request to the activity service handler.
@@ -71,6 +78,10 @@ func NewFrontendHandler(
 // before mutation to preserve the original for retries.
 // 3. Sends the request to the history activity service.
 func (h *frontendHandler) StartActivityExecution(ctx context.Context, req *workflowservice.StartActivityExecutionRequest) (*workflowservice.StartActivityExecutionResponse, error) {
+	if !h.config.Enabled(req.GetNamespace()) {
+		return nil, serviceerror.NewUnavailable(StandaloneActivityDisabledError)
+	}
+
 	namespaceID, err := h.namespaceRegistry.GetNamespaceID(namespace.Name(req.GetNamespace()))
 	if err != nil {
 		return nil, err
@@ -95,9 +106,13 @@ func (h *frontendHandler) DescribeActivityExecution(
 	ctx context.Context,
 	req *workflowservice.DescribeActivityExecutionRequest,
 ) (*workflowservice.DescribeActivityExecutionResponse, error) {
+	if !h.config.Enabled(req.GetNamespace()) {
+		return nil, serviceerror.NewUnavailable(StandaloneActivityDisabledError)
+	}
+
 	err := ValidateDescribeActivityExecutionRequest(
 		req,
-		dynamicconfig.MaxIDLengthLimit.Get(h.dc)(),
+		h.config.MaxIDLengthLimit(),
 	)
 	if err != nil {
 		return nil, err
@@ -120,9 +135,13 @@ func (h *frontendHandler) GetActivityExecutionOutcome(
 	ctx context.Context,
 	req *workflowservice.GetActivityExecutionOutcomeRequest,
 ) (*workflowservice.GetActivityExecutionOutcomeResponse, error) {
+	if !h.config.Enabled(req.GetNamespace()) {
+		return nil, serviceerror.NewUnavailable(StandaloneActivityDisabledError)
+	}
+
 	err := ValidateGetActivityExecutionOutcomeRequest(
 		req,
-		dynamicconfig.MaxIDLengthLimit.Get(h.dc)(),
+		h.config.MaxIDLengthLimit(),
 	)
 	if err != nil {
 		return nil, err
@@ -143,6 +162,10 @@ func (h *frontendHandler) TerminateActivityExecution(
 	ctx context.Context,
 	req *workflowservice.TerminateActivityExecutionRequest,
 ) (*workflowservice.TerminateActivityExecutionResponse, error) {
+	if !h.config.Enabled(req.GetNamespace()) {
+		return nil, serviceerror.NewUnavailable(StandaloneActivityDisabledError)
+	}
+
 	namespaceName := req.GetNamespace()
 	namespaceID, err := h.namespaceRegistry.GetNamespaceID(namespace.Name(namespaceName))
 	if err != nil {
@@ -152,8 +175,8 @@ func (h *frontendHandler) TerminateActivityExecution(
 	if err := validateInputSize(
 		req.GetActivityId(),
 		"activity-termination",
-		dynamicconfig.BlobSizeLimitError.Get(h.dc),
-		dynamicconfig.BlobSizeLimitWarn.Get(h.dc),
+		h.config.BlobSizeLimitError,
+		h.config.BlobSizeLimitWarn,
 		len(req.GetReason()),
 		h.logger,
 		namespaceName); err != nil {
@@ -177,6 +200,10 @@ func (h *frontendHandler) RequestCancelActivityExecution(
 	ctx context.Context,
 	req *workflowservice.RequestCancelActivityExecutionRequest,
 ) (*workflowservice.RequestCancelActivityExecutionResponse, error) {
+	if !h.config.Enabled(req.GetNamespace()) {
+		return nil, serviceerror.NewUnavailable(StandaloneActivityDisabledError)
+	}
+
 	namespaceID, err := h.namespaceRegistry.GetNamespaceID(namespace.Name(req.GetNamespace()))
 	if err != nil {
 		return nil, err
@@ -185,7 +212,7 @@ func (h *frontendHandler) RequestCancelActivityExecution(
 	// Since validation potentially mutates the request, we clone it first so that any retries use the original request.
 	req = common.CloneProto(req)
 
-	maxIDLen := dynamicconfig.MaxIDLengthLimit.Get(h.dc)()
+	maxIDLen := h.config.MaxIDLengthLimit()
 
 	if len(req.GetRequestId()) > maxIDLen {
 		return nil, serviceerror.NewInvalidArgument("RequestID length exceeds limit.")
@@ -226,8 +253,8 @@ func (h *frontendHandler) validateAndPopulateStartRequest(
 	err := ValidateAndNormalizeActivityAttributes(
 		req.ActivityId,
 		activityType,
-		dynamicconfig.DefaultActivityRetryPolicy.Get(h.dc),
-		dynamicconfig.MaxIDLengthLimit.Get(h.dc)(),
+		h.config.DefaultActivityRetryPolicy,
+		h.config.MaxIDLengthLimit(),
 		namespaceID,
 		opts,
 		req.Priority,
@@ -240,10 +267,10 @@ func (h *frontendHandler) validateAndPopulateStartRequest(
 
 	err = validateAndNormalizeStartActivityExecutionRequest(
 		req,
-		dynamicconfig.BlobSizeLimitError.Get(h.dc),
-		dynamicconfig.BlobSizeLimitWarn.Get(h.dc),
+		h.config.BlobSizeLimitError,
+		h.config.BlobSizeLimitWarn,
 		h.logger,
-		dynamicconfig.MaxIDLengthLimit.Get(h.dc)(),
+		h.config.MaxIDLengthLimit(),
 		h.saValidator)
 	if err != nil {
 		return nil, err

chasm/lib/activity/fx.go

@@ -26,6 +26,7 @@ var HistoryModule = fx.Module(
 
 var FrontendModule = fx.Module(
 	"activity-frontend",
+	fx.Provide(ConfigProvider),
 	fx.Provide(activitypb.NewActivityServiceLayeredClient),
 	fx.Provide(NewFrontendHandler),
 	fx.Provide(resource.SearchAttributeValidatorProvider),

config/dynamicconfig/development-cass.yaml

@@ -57,3 +57,5 @@ history.enableTransitionHistory:
   - value: true
 history.enableChasm:
   - value: true
+activity.enableStandalone:
+  - value: true

config/dynamicconfig/development-sql.yaml

@@ -73,3 +73,5 @@ history.enableTransitionHistory:
   - value: true
 history.enableChasm:
   - value: true
+activity.enableStandalone:
+  - value: true

config/dynamicconfig/development-xdc.yaml

@@ -49,3 +49,5 @@ history.EnableReplicationTaskTieredProcessing:
   - value: true
 history.enableChasm:
   - value: true
+activity.enableStandalone:
+  - value: true

service/frontend/workflow_handler.go

@@ -1233,6 +1233,11 @@ func (wh *WorkflowHandler) RecordActivityTaskHeartbeat(ctx context.Context, requ
 	if err != nil {
 		return nil, err
 	}
+	namespaceName := namespaceEntry.Name().String()
+
+	if len(taskToken.GetComponentRef()) > 0 && !wh.IsStandaloneActivityEnabled(namespaceName) {
+		return nil, serviceerror.NewUnavailable(activity.StandaloneActivityDisabledError)
+	}
 
 	sizeLimitError := wh.config.BlobSizeLimitError(namespaceEntry.Name().String())
 	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceEntry.Name().String())
@@ -1404,13 +1409,18 @@ func (wh *WorkflowHandler) RespondActivityTaskCompleted(
 	if err != nil {
 		return nil, err
 	}
+	namespaceName := namespaceEntry.Name().String()
+
+	if len(taskToken.GetComponentRef()) > 0 && !wh.IsStandaloneActivityEnabled(namespaceName) {
+		return nil, serviceerror.NewUnavailable(activity.StandaloneActivityDisabledError)
+	}
 
 	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
 		return nil, errIdentityTooLong
 	}
 
-	sizeLimitError := wh.config.BlobSizeLimitError(namespaceEntry.Name().String())
-	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceEntry.Name().String())
+	sizeLimitError := wh.config.BlobSizeLimitError(namespaceName)
+	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceName)
 
 	if err := common.CheckEventBlobSizeLimit(
 		request.GetResult().Size(),
@@ -1477,10 +1487,14 @@ func (wh *WorkflowHandler) RespondActivityTaskCompletedById(ctx context.Context,
 		return nil, errIdentityTooLong
 	}
 
-	// If workflowID is empty, it means the activity is a standalone activity and we need to set the component ref
-	// TODO Need to add a dynamic config to enable standalone configs, and incorporate that into the check below
+	// If workflowID is empty, it means the activity is a standalone activity and we need to set the component ref.
+	// Else this should be a validation error.
 	var componentRef []byte
 	if workflowID == "" {
+		if !wh.IsStandaloneActivityEnabled(request.GetNamespace()) {
+			return nil, errWorkflowIDNotSet
+		}
+
 		ref := chasm.NewComponentRef[*activity.Activity](chasm.ExecutionKey{
 			NamespaceID: namespaceID.String(),
 			BusinessID:  activityID,
@@ -1587,6 +1601,11 @@ func (wh *WorkflowHandler) RespondActivityTaskFailed(
 	if err != nil {
 		return nil, err
 	}
+	namespaceName := namespaceEntry.Name().String()
+
+	if len(taskToken.GetComponentRef()) > 0 && !wh.IsStandaloneActivityEnabled(namespaceName) {
+		return nil, serviceerror.NewUnavailable(activity.StandaloneActivityDisabledError)
+	}
 
 	if request.GetFailure() != nil && request.GetFailure().GetApplicationFailureInfo() == nil {
 		return nil, errFailureMustHaveApplicationFailureInfo
@@ -1596,8 +1615,8 @@ func (wh *WorkflowHandler) RespondActivityTaskFailed(
 		return nil, errIdentityTooLong
 	}
 
-	sizeLimitError := wh.config.BlobSizeLimitError(namespaceEntry.Name().String())
-	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceEntry.Name().String())
+	sizeLimitError := wh.config.BlobSizeLimitError(namespaceName)
+	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceName)
 
 	response := workflowservice.RespondActivityTaskFailedResponse{}
 
@@ -1676,10 +1695,14 @@ func (wh *WorkflowHandler) RespondActivityTaskFailedById(ctx context.Context, re
 		return nil, errIdentityTooLong
 	}
 
-	// If workflowID is empty, it means the activity is a standalone activity and we need to set the component ref
-	// TODO Need to add a dynamic config to enable standalone configs, and incorporate that into the check below
+	// If workflowID is empty, it means the activity is a standalone activity and we need to set the component ref.
+	// Else this should be a validation error.
 	var componentRef []byte
 	if workflowID == "" {
+		if !wh.IsStandaloneActivityEnabled(request.GetNamespace()) {
+			return nil, errWorkflowIDNotSet
+		}
+
 		ref := chasm.NewComponentRef[*activity.Activity](chasm.ExecutionKey{
 			NamespaceID: namespaceID.String(),
 			BusinessID:  activityID,
@@ -1795,13 +1818,18 @@ func (wh *WorkflowHandler) RespondActivityTaskCanceled(ctx context.Context, requ
 	if err != nil {
 		return nil, err
 	}
+	namespaceName := namespaceEntry.Name().String()
+
+	if len(taskToken.GetComponentRef()) > 0 && !wh.IsStandaloneActivityEnabled(namespaceName) {
+		return nil, serviceerror.NewUnavailable(activity.StandaloneActivityDisabledError)
+	}
 
 	if len(request.GetIdentity()) > wh.config.MaxIDLengthLimit() {
 		return nil, errIdentityTooLong
 	}
 
-	sizeLimitError := wh.config.BlobSizeLimitError(namespaceEntry.Name().String())
-	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceEntry.Name().String())
+	sizeLimitError := wh.config.BlobSizeLimitError(namespaceName)
+	sizeLimitWarn := wh.config.BlobSizeLimitWarn(namespaceName)
 
 	if err := common.CheckEventBlobSizeLimit(
 		request.GetDetails().Size(),
@@ -1867,10 +1895,14 @@ func (wh *WorkflowHandler) RespondActivityTaskCanceledById(ctx context.Context,
 		return nil, errIdentityTooLong
 	}
 
-	// If workflowID is empty, it means the activity is a standalone activity and we need to set the component ref
-	// TODO Need to add a dynamic config to enable standalone configs, and incorporate that into the check below
+	// If workflowID is empty, it means the activity is a standalone activity and we need to set the component ref.
+	// Else this should be a validation error.
 	var componentRef []byte
 	if workflowID == "" {
+		if !wh.IsStandaloneActivityEnabled(request.GetNamespace()) {
+			return nil, errWorkflowIDNotSet
+		}
+
 		ref := chasm.NewComponentRef[*activity.Activity](chasm.ExecutionKey{
 			NamespaceID: namespaceID.String(),
 			BusinessID:  activityID,

tests/standalone_activity_test.go

@@ -56,6 +56,10 @@ func (s *standaloneActivityTestSuite) SetupSuite() {
 		dynamicconfig.EnableChasm,
 		true,
 	)
+	s.OverrideDynamicConfig(
+		activity.Enabled,
+		true,
+	)
 }
 
 func (s *standaloneActivityTestSuite) SetupTest() {