Added callback support for standalone activities

Author: fretz12

chasm/lib/activity/activity.go

@@ -6,22 +6,28 @@ import (
 	"slices"
 	"time"
 
+	"github.com/nexus-rpc/sdk-go/nexus"
 	apiactivitypb "go.temporal.io/api/activity/v1" //nolint:importas
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	failurepb "go.temporal.io/api/failure/v1"
 	historypb "go.temporal.io/api/history/v1"
 	"go.temporal.io/api/serviceerror"
+	workflowpb "go.temporal.io/api/workflow/v1"
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	"go.temporal.io/server/api/matchingservice/v1"
 	tokenspb "go.temporal.io/server/api/token/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1"
+	"go.temporal.io/server/chasm/lib/callback"
+	callbackspb "go.temporal.io/server/chasm/lib/callback/gen/callbackpb/v1"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/backoff"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
+	commonnexus "go.temporal.io/server/common/nexus"
+	"go.temporal.io/server/common/nexus/nexusrpc"
 	"go.temporal.io/server/common/payload"
 	serviceerrors "go.temporal.io/server/common/serviceerror"
 	"go.temporal.io/server/common/tqid"
@@ -41,6 +47,7 @@ var (
 )
 
 var _ chasm.VisibilitySearchAttributesProvider = (*Activity)(nil)
+var _ callback.CompletionSource = (*Activity)(nil)
 
 type ActivityStore interface {
 	// RecordCompleted applies the provided function to record activity completion
@@ -65,6 +72,10 @@ type Activity struct {
 	// implements the ActivityStore interface).
 	// TODO(saa-preview): figure out better naming.
 	Store chasm.ParentPtr[ActivityStore]
+
+	// Callbacks holds completion callbacks to be invoked when this standalone activity reaches a terminal state. Nil
+	// for workflow-embedded activities as the workflow handles its own callbacks.
+	Callbacks chasm.Map[string, *callback.Callback]
 }
 
 // WithToken wraps a request with its deserialized task token.
@@ -256,8 +267,136 @@ func attemptScheduleTimeForRetry(attempt *activitypb.ActivityAttemptState) *time
 }
 
 // RecordCompleted applies the provided function to record activity completion.
+// For standalone activities, it also triggers any registered completion callbacks.
 func (a *Activity) RecordCompleted(ctx chasm.MutableContext, applyFn func(ctx chasm.MutableContext) error) error {
-	return applyFn(ctx)
+	if err := applyFn(ctx); err != nil {
+		return err
+	}
+	return a.processCloseCallbacks(ctx)
+}
+
+func (a *Activity) addCompletionCallbacks(
+	ctx chasm.MutableContext,
+	requestID string,
+	completionCallbacks []*commonpb.Callback,
+	maxCallbacks int,
+) error {
+	if len(completionCallbacks) == 0 {
+		return nil
+	}
+
+	currentCount := len(a.Callbacks)
+	if len(completionCallbacks)+currentCount > maxCallbacks {
+		return serviceerror.NewFailedPreconditionf(
+			"cannot attach more than %d callbacks to an activity (%d callbacks already attached)",
+			maxCallbacks,
+			currentCount,
+		)
+	}
+
+	if a.Callbacks == nil {
+		a.Callbacks = make(chasm.Map[string, *callback.Callback], len(completionCallbacks))
+	}
+
+	registrationTime := timestamppb.New(ctx.Now(a))
+
+	for idx, cb := range completionCallbacks {
+		chasmCB := &callbackspb.Callback{
+			Links: cb.GetLinks(),
+		}
+		switch variant := cb.Variant.(type) {
+		case *commonpb.Callback_Nexus_:
+			chasmCB.Variant = &callbackspb.Callback_Nexus_{
+				Nexus: &callbackspb.Callback_Nexus{
+					Url:    variant.Nexus.GetUrl(),
+					Header: variant.Nexus.GetHeader(),
+				},
+			}
+		default:
+			return serviceerror.NewInvalidArgumentf("unsupported callback variant: %T", variant)
+		}
+
+		id := fmt.Sprintf("%s-%d", requestID, idx)
+		callbackObj := callback.NewCallback(requestID, registrationTime, &callbackspb.CallbackState{}, chasmCB)
+		a.Callbacks[id] = chasm.NewComponentField(ctx, callbackObj)
+	}
+	return nil
+}
+
+// processCloseCallbacks triggers all STANDBY completion callbacks by transitioning them
+// to SCHEDULED state, which causes the callback library to invoke them.
+func (a *Activity) processCloseCallbacks(ctx chasm.MutableContext) error {
+	for _, field := range a.Callbacks {
+		cb := field.Get(ctx)
+		if cb.Status != callbackspb.CALLBACK_STATUS_STANDBY {
+			continue
+		}
+		if err := callback.TransitionScheduled.Apply(cb, ctx, callback.EventScheduled{}); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// GetNexusCompletion returns the activity's completion data in the format required by the Nexus callback invocation.
+// Implements callback.CompletionSource.
+func (a *Activity) GetNexusCompletion(ctx chasm.Context, _ string) (nexusrpc.CompleteOperationOptions, error) {
+	if !a.LifecycleState(ctx).IsClosed() {
+		return nexusrpc.CompleteOperationOptions{}, serviceerror.NewFailedPrecondition("activity has not completed yet")
+	}
+
+	attempt := a.LastAttempt.Get(ctx)
+	opts := nexusrpc.CompleteOperationOptions{
+		StartTime: attempt.GetStartedTime().AsTime(),
+		CloseTime: attempt.GetCompleteTime().AsTime(),
+	}
+
+	outcome := a.Outcome.Get(ctx)
+	if successful := outcome.GetSuccessful(); successful != nil {
+		// Successful completion: return the first output payload as the result as Nexus supports only a single payload
+		var p *commonpb.Payload
+		if payloads := successful.GetOutput().GetPayloads(); len(payloads) > 0 {
+			p = payloads[0]
+		}
+		opts.Result = p
+		return opts, nil
+	}
+
+	var failure *failurepb.Failure
+	if f := outcome.GetFailed(); f != nil {
+		failure = f.GetFailure()
+	}
+	if failure == nil {
+		if details := attempt.GetLastFailureDetails(); details != nil {
+			failure = details.GetFailure()
+		}
+	}
+
+	if failure != nil {
+		state := nexus.OperationStateFailed
+		message := "operation failed"
+		if a.Status == activitypb.ACTIVITY_EXECUTION_STATUS_CANCELED {
+			state = nexus.OperationStateCanceled
+			message = "operation canceled"
+		}
+
+		nf, err := commonnexus.TemporalFailureToNexusFailure(failure)
+		if err != nil {
+			return nexusrpc.CompleteOperationOptions{}, serviceerror.NewInternalf("failed to convert failure: %v", err)
+		}
+
+		opErr := &nexus.OperationError{
+			State:   state,
+			Message: message,
+			Cause:   &nexus.FailureError{Failure: nf},
+		}
+		if err := nexusrpc.MarkAsWrapperError(nexusrpc.DefaultFailureConverter(), opErr); err != nil {
+			return nexusrpc.CompleteOperationOptions{}, err
+		}
+		opts.Error = opErr
+	}
+
+	return opts, nil
 }
 
 // HandleCompleted updates the activity on activity completion.
@@ -716,11 +855,17 @@ func (a *Activity) buildDescribeActivityExecutionResponse(
 		input = a.RequestData.Get(ctx).GetInput()
 	}
 
+	callbackInfos, err := a.buildCallbackInfos(ctx)
+	if err != nil {
+		return nil, err
+	}
+
 	response := &workflowservice.DescribeActivityExecutionResponse{
 		Info:          info,
 		RunId:         ctx.ExecutionKey().RunID,
 		Input:         input,
 		LongPollToken: token,
+		Callbacks:     callbackInfos,
 	}
 
 	if request.GetIncludeOutcome() {
@@ -732,6 +877,52 @@ func (a *Activity) buildDescribeActivityExecutionResponse(
 	}, nil
 }
 
+func (a *Activity) buildCallbackInfos(ctx chasm.Context) ([]*workflowpb.CallbackInfo, error) {
+	if len(a.Callbacks) == 0 {
+		return nil, nil
+	}
+
+	cbInfos := make([]*workflowpb.CallbackInfo, 0, len(a.Callbacks))
+	for _, field := range a.Callbacks {
+		cb := field.Get(ctx)
+
+		cbSpec, err := cb.ToAPICallback()
+		if err != nil {
+			return nil, err
+		}
+
+		var state enumspb.CallbackState
+		switch cb.Status {
+		case callbackspb.CALLBACK_STATUS_UNSPECIFIED:
+			return nil, serviceerror.NewInternal("callback with UNSPECIFIED state")
+		case callbackspb.CALLBACK_STATUS_STANDBY:
+			state = enumspb.CALLBACK_STATE_STANDBY
+		case callbackspb.CALLBACK_STATUS_SCHEDULED:
+			state = enumspb.CALLBACK_STATE_SCHEDULED
+		case callbackspb.CALLBACK_STATUS_BACKING_OFF:
+			state = enumspb.CALLBACK_STATE_BACKING_OFF
+		case callbackspb.CALLBACK_STATUS_FAILED:
+			state = enumspb.CALLBACK_STATE_FAILED
+		case callbackspb.CALLBACK_STATUS_SUCCEEDED:
+			state = enumspb.CALLBACK_STATE_SUCCEEDED
+		default:
+			return nil, serviceerror.NewInternalf("unknown callback state: %v", cb.Status)
+		}
+
+		cbInfos = append(cbInfos, &workflowpb.CallbackInfo{
+			Callback:                cbSpec,
+			Trigger:                 &workflowpb.CallbackInfo_Trigger{Variant: &workflowpb.CallbackInfo_Trigger_WorkflowClosed{}},
+			RegistrationTime:        cb.RegistrationTime,
+			State:                   state,
+			Attempt:                 cb.Attempt,
+			LastAttemptCompleteTime: cb.LastAttemptCompleteTime,
+			LastAttemptFailure:      cb.LastAttemptFailure,
+			NextAttemptScheduleTime: cb.NextAttemptScheduleTime,
+		})
+	}
+	return cbInfos, nil
+}
+
 func (a *Activity) buildPollActivityExecutionResponse(
 	ctx chasm.Context,
 ) *activitypb.PollActivityExecutionResponse {

chasm/lib/activity/config.go

@@ -5,6 +5,7 @@ import (
 
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/retrypolicy"
+	"go.temporal.io/server/components/callbacks"
 )
 
 var (
@@ -33,10 +34,14 @@ type Config struct {
 	BlobSizeLimitError          dynamicconfig.IntPropertyFnWithNamespaceFilter
 	BlobSizeLimitWarn           dynamicconfig.IntPropertyFnWithNamespaceFilter
 	BreakdownMetricsByTaskQueue dynamicconfig.TypedPropertyFnWithTaskQueueFilter[bool]
+	CallbackEndpointConfigs     dynamicconfig.TypedPropertyFnWithNamespaceFilter[callbacks.AddressMatchRules]
+	CallbackHeaderMaxSize       dynamicconfig.IntPropertyFnWithNamespaceFilter
+	CallbackURLMaxLength        dynamicconfig.IntPropertyFnWithNamespaceFilter
 	Enabled                     dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	LongPollBuffer              dynamicconfig.DurationPropertyFnWithNamespaceFilter
 	LongPollTimeout             dynamicconfig.DurationPropertyFnWithNamespaceFilter
 	MaxIDLengthLimit            dynamicconfig.IntPropertyFn
+	MaxCallbacksPerExecution    dynamicconfig.IntPropertyFnWithNamespaceFilter
 	DefaultActivityRetryPolicy  dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings]
 	VisibilityMaxPageSize       dynamicconfig.IntPropertyFnWithNamespaceFilter
 }
@@ -46,11 +51,15 @@ func ConfigProvider(dc *dynamicconfig.Collection) *Config {
 		BlobSizeLimitError:          dynamicconfig.BlobSizeLimitError.Get(dc),
 		BlobSizeLimitWarn:           dynamicconfig.BlobSizeLimitWarn.Get(dc),
 		BreakdownMetricsByTaskQueue: dynamicconfig.MetricsBreakdownByTaskQueue.Get(dc),
+		CallbackEndpointConfigs:     callbacks.AllowedAddresses.Get(dc),
+		CallbackHeaderMaxSize:       dynamicconfig.FrontendCallbackHeaderMaxSize.Get(dc),
+		CallbackURLMaxLength:        dynamicconfig.FrontendCallbackURLMaxLength.Get(dc),
 		DefaultActivityRetryPolicy:  dynamicconfig.DefaultActivityRetryPolicy.Get(dc),
 		Enabled:                     Enabled.Get(dc),
 		LongPollBuffer:              LongPollBuffer.Get(dc),
 		LongPollTimeout:             LongPollTimeout.Get(dc),
 		MaxIDLengthLimit:            dynamicconfig.MaxIDLengthLimit.Get(dc),
+		MaxCallbacksPerExecution:    dynamicconfig.MaxCallbacksPerExecution.Get(dc),
 		VisibilityMaxPageSize:       dynamicconfig.FrontendVisibilityMaxPageSize.Get(dc),
 	}
 }

chasm/lib/activity/frontend.go

@@ -16,6 +16,7 @@ import (
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/searchattribute"
+	"go.temporal.io/server/components/callbacks"
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/emptypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -426,6 +427,19 @@ func (h *frontendHandler) validateAndNormalizeStartActivityExecutionRequest(
 		return serviceerror.NewInvalidArgument("input exceeds length limit")
 	}
 
+	if cbs := req.GetCompletionCallbacks(); len(cbs) > 0 {
+		if err := callbacks.ValidateCallbacks(
+			cbs,
+			h.config.MaxCallbacksPerExecution(req.GetNamespace()),
+			h.config.CallbackURLMaxLength(req.GetNamespace()),
+			h.config.CallbackHeaderMaxSize(req.GetNamespace()),
+			h.config.CallbackEndpointConfigs(req.GetNamespace()),
+			"an activity",
+		); err != nil {
+			return serviceerror.NewInvalidArgument(err.Error())
+		}
+	}
+
 	if req.GetSearchAttributes() != nil {
 		if err := validateAndNormalizeSearchAttributes(
 			req,

chasm/lib/activity/handler.go

@@ -73,6 +73,13 @@ func (h *handler) StartActivityExecution(ctx context.Context, req *activitypb.St
 				return nil, err
 			}
 
+			if cbs := request.GetCompletionCallbacks(); len(cbs) > 0 {
+				maxCallbacks := h.config.MaxCallbacksPerExecution(request.GetNamespace())
+				if err := newActivity.addCompletionCallbacks(mutableContext, request.GetRequestId(), cbs, maxCallbacks); err != nil {
+					return nil, err
+				}
+			}
+
 			err = TransitionScheduled.Apply(newActivity, mutableContext, nil)
 			if err != nil {
 				return nil, err

cmd/tools/getproto/files.go

@@ -1000,12 +1000,10 @@ so forwarding by endpoint ID will not work out of the box.`,
 		32,
 		`MaxCallbacksPerWorkflow is the maximum number of callbacks that can be attached to a workflow.`,
 	)
-	// NOTE (seankane): MaxCHASMCallbacksPerWorkflow is temporary, this will be removed and replaced with MaxCallbacksPerWorkflow
-	// once CHASM is fully enabled
-	MaxCHASMCallbacksPerWorkflow = NewNamespaceIntSetting(
-		"system.maxCHASMCallbacksPerWorkflow",
+	MaxCallbacksPerExecution = NewNamespaceIntSetting(
+		"system.maxCallbacksPerExecution",
 		2000,
-		`MaxCHASMCallbacksPerWorkflow is the maximum number of callbacks that can be attached to a workflow when using the CHASM implementation.`,
+		`MaxCallbacksPerExecution is the maximum number of callbacks that can be attached to a CHASM execution.`,
 	)
 	FrontendLinkMaxSize = NewNamespaceIntSetting(
 		"frontend.linkMaxSize",