Add support for standalone callbacks

chasm/lib/activity/activity.go

@@ -335,9 +335,9 @@ func (a *Activity) addCompletionCallbacks(
 			return serviceerror.NewInvalidArgumentf("unsupported callback variant: %T", variant)
 		}
 
-		// requestID (unique per API call) + idx (position within the request) ensures unique,idempotent callback IDs.
+		// requestID (unique per API call) + idx (position within the request) ensures unique, idempotent callback IDs.
 		id := fmt.Sprintf("%s-%d", requestID, idx)
-		callbackObj := callback.NewCallback(requestID, registrationTime, &callbackspb.CallbackState{}, chasmCB)
+		callbackObj := callback.NewEmbeddedCallback(requestID, registrationTime, chasmCB)
 		a.Callbacks[id] = chasm.NewComponentField(ctx, callbackObj)
 	}
 	return nil

chasm/lib/callback/component.go

@@ -4,22 +4,40 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/nexus-rpc/sdk-go/nexus"
+	callbackpb "go.temporal.io/api/callback/v1"
 	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/serviceerror"
 	"go.temporal.io/server/chasm"
 	callbackspb "go.temporal.io/server/chasm/lib/callback/gen/callbackpb/v1"
 	"go.temporal.io/server/common/backoff"
+	commonnexus "go.temporal.io/server/common/nexus"
 	"go.temporal.io/server/common/nexus/nexusrpc"
 	queueserrors "go.temporal.io/server/service/history/queues/errors"
+	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 type CompletionSource interface {
 	GetNexusCompletion(ctx chasm.Context, requestID string) (nexusrpc.CompleteOperationOptions, error)
 }
 
-var _ chasm.Component = (*Callback)(nil)
-var _ chasm.StateMachine[callbackspb.CallbackStatus] = (*Callback)(nil)
+var (
+	_ chasm.RootComponent                            = (*Callback)(nil)
+	_ chasm.StateMachine[callbackspb.CallbackStatus] = (*Callback)(nil)
+	_ chasm.VisibilitySearchAttributesProvider       = (*Callback)(nil)
+
+	// The Callback itself is a completion source. Either delegating to its parent
+	// or returning the user-supplied completion.
+	_ CompletionSource = (*Callback)(nil)
+)
+
+var executionStatusSearchAttribute = chasm.NewSearchAttributeKeyword(
+	"ExecutionStatus",
+	chasm.SearchAttributeFieldLowCardinalityKeyword01,
+)
 
 // Callback represents a callback component in CHASM.
 type Callback struct {
@@ -28,14 +46,25 @@ type Callback struct {
 	// Persisted internal state
 	*callbackspb.CallbackState
 
-	// Interface to retrieve Nexus operation completion data
-	CompletionSource chasm.ParentPtr[CompletionSource]
+	// Failure from an external termination (timeout or terminate), stored separately because
+	// of its potential size, and to not overload CallbackState::LastAttemptFailure.
+	TerminalFailure chasm.Field[*failurepb.Failure]
+
+	// For embedded callbacks, the completion result is obtained from the parent component.
+	// e.g. the Workflow result to be delivered.
+	ParentCompletionSource chasm.ParentPtr[CompletionSource]
+	// The user-supplied completion for standalone callbacks.
+	SuppliedCompletion chasm.Field[*callbackpb.CallbackExecutionCompletion]
+
+	// Visibility sub-component for search attributes and memo indexing.
+	Visibility chasm.Field[*chasm.Visibility]
 }
 
-func NewCallback(
+// NewEmbeddedCallback returns a Callback component, which will deliver the completion from
+// its parent CHASM component. The parent must implement CompletionSource.
+func NewEmbeddedCallback(
 	requestID string,
 	registrationTime *timestamppb.Timestamp,
-	state *callbackspb.CallbackState,
 	cb *callbackspb.Callback,
 ) *Callback {
 	return &Callback{
@@ -48,11 +77,58 @@ func NewCallback(
 	}
 }
 
+// newStandaloneCallbackInput is the bundle of inputs to the CHASM execution.
+type newStandaloneCallbackInput struct {
+	RequestID              string
+	Callback               *callbackspb.Callback
+	ScheduleToCloseTimeout *durationpb.Duration
+	Completion             *callbackpb.CallbackExecutionCompletion
+	SearchAttributes       map[string]*commonpb.Payload
+}
+
+// newStandaloneCallback constructs a new Callback component in standalone mode.
+// The Callback is immediately transitioned to SCHEDULED state to begin invocation.
+func newStandaloneCallback(
+	ctx chasm.MutableContext,
+	input *newStandaloneCallbackInput,
+) (*Callback, error) {
+	now := timestamppb.Now()
+
+	// Create Callback component.
+	cb := NewEmbeddedCallback(input.RequestID, now, input.Callback)
+	cb.ScheduleToCloseTimeout = input.ScheduleToCloseTimeout
+	cb.SuppliedCompletion = chasm.NewDataField(ctx, input.Completion)
+
+	visibility := chasm.NewVisibilityWithData(ctx, input.SearchAttributes, nil)
+	cb.Visibility = chasm.NewComponentField(ctx, visibility)
+
+	// Immediately schedule the callback for invocation.
+	if err := TransitionScheduled.Apply(cb, ctx, EventScheduled{}); err != nil {
+		return nil, fmt.Errorf("failed to schedule callback: %w", err)
+	}
+
+	// Schedule the timeout as applicable.
+	if durationProto := input.ScheduleToCloseTimeout; durationProto != nil {
+		if duration := durationProto.AsDuration(); duration > 0 {
+			timeoutTime := now.AsTime().Add(duration)
+			ctx.AddTask(
+				cb,
+				chasm.TaskAttributes{ScheduledTime: timeoutTime},
+				&callbackspb.ScheduleToCloseTimeoutTask{},
+			)
+		}
+	}
+
+	return cb, nil
+}
+
 func (c *Callback) LifecycleState(_ chasm.Context) chasm.LifecycleState {
 	switch c.Status {
 	case callbackspb.CALLBACK_STATUS_SUCCEEDED:
 		return chasm.LifecycleStateCompleted
-	case callbackspb.CALLBACK_STATUS_FAILED:
+	case callbackspb.CALLBACK_STATUS_FAILED,
+		callbackspb.CALLBACK_STATUS_TERMINATED,
+		callbackspb.CALLBACK_STATUS_TIMED_OUT:
 		return chasm.LifecycleStateFailed
 	default:
 		return chasm.LifecycleStateRunning
@@ -67,6 +143,56 @@ func (c *Callback) SetStateMachineState(status callbackspb.CallbackStatus) {
 	c.Status = status
 }
 
+// ContextMetadata is used for root CHASM components, so this is only applicable
+// for the standalone callback case.
+func (c *Callback) ContextMetadata(ctx chasm.Context) map[string]string {
+	return map[string]string{
+		"request-id":  c.RequestId,
+		"callback-id": ctx.ExecutionKey().BusinessID,
+	}
+}
+
+// SearchAttributes implements chasm.VisibilitySearchAttributesProvider.
+func (c *Callback) SearchAttributes(ctx chasm.Context) []chasm.SearchAttributeKeyValue {
+	apiStatus := c.statusAsAPIExecutionStatus()
+	return []chasm.SearchAttributeKeyValue{
+		executionStatusSearchAttribute.Value(apiStatus.String()),
+	}
+}
+
+// Terminate forcefully terminates the callback execution.
+//
+// If already terminated with the same request ID, this is a no-op.
+// If already terminated with a different request ID, returns FailedPrecondition.
+func (c *Callback) Terminate(
+	ctx chasm.MutableContext,
+	req chasm.TerminateComponentRequest,
+) (chasm.TerminateComponentResponse, error) {
+	if c.LifecycleState(ctx).IsClosed() {
+		if c.TerminateRequestId == "" {
+			// Completed organically (succeeded/failed/timed out), not via Terminate.
+			err := serviceerror.NewFailedPreconditionf("callback execution already in terminal state %v", c.Status)
+			return chasm.TerminateComponentResponse{}, err
+		}
+		if c.TerminateRequestId != req.RequestID {
+			err := serviceerror.NewFailedPreconditionf("already terminated with request ID %s", c.TerminateRequestId)
+			return chasm.TerminateComponentResponse{}, err
+		}
+		return chasm.TerminateComponentResponse{}, nil
+	}
+	event := EventTerminated{
+		Identity: req.Identity,
+		Reason:   req.Reason,
+	}
+	if err := TransitionTerminated.Apply(c, ctx, event); err != nil {
+		return chasm.TerminateComponentResponse{}, fmt.Errorf("failed to terminate callback: %w", err)
+	}
+
+	c.TerminateRequestId = req.RequestID
+	// c.TerminalFailure is set in the transition handler.
+	return chasm.TerminateComponentResponse{}, nil
+}
+
 func (c *Callback) recordAttempt(ts time.Time) {
 	c.Attempt++
 	c.LastAttemptCompleteTime = timestamppb.New(ts)
@@ -77,9 +203,8 @@ func (c *Callback) loadInvocationArgs(
 	ctx chasm.Context,
 	_ chasm.NoValue,
 ) (invocable, error) {
-	target := c.CompletionSource.Get(ctx)
-
-	completion, err := target.GetNexusCompletion(ctx, c.RequestId)
+	// Get the completion result to be delivered.
+	completion, err := c.GetNexusCompletion(ctx, c.RequestId)
 	if err != nil {
 		return nil, err
 	}
@@ -179,3 +304,156 @@ func ScheduleStandbyCallbacks(ctx chasm.MutableContext, callbacks chasm.Map[stri
 	}
 	return nil
 }
+
+func callbackCompletionToNexusCompleteOperationOpts(
+	cb *Callback,
+	completion *callbackpb.CallbackExecutionCompletion) (nexusrpc.CompleteOperationOptions, error) {
+
+	nexusCompletion := nexusrpc.CompleteOperationOptions{
+		StartTime: cb.GetRegistrationTime().AsTime(),
+		CloseTime: cb.CloseTime.AsTime(),
+	}
+
+	switch completion.Result.(type) {
+	case *callbackpb.CallbackExecutionCompletion_Success:
+		nexusCompletion.Result = completion.GetSuccess()
+		return nexusCompletion, nil
+
+	case *callbackpb.CallbackExecutionCompletion_Failure:
+		f, err := commonnexus.TemporalFailureToNexusFailure(completion.GetFailure())
+		if err != nil {
+			wrappedErr := fmt.Errorf("failed to convert failure: %w", err)
+			return nexusrpc.CompleteOperationOptions{}, wrappedErr
+		}
+		opErr := &nexus.OperationError{
+			State:   nexus.OperationStateFailed,
+			Message: "operation failed",
+			Cause:   &nexus.FailureError{Failure: f},
+		}
+		if err := nexusrpc.MarkAsWrapperError(nexusrpc.DefaultFailureConverter(), opErr); err != nil {
+			wrappedErr := fmt.Errorf("failed to mark wrapper error: %w", err)
+			return nexusrpc.CompleteOperationOptions{}, wrappedErr
+		}
+		nexusCompletion.Error = opErr
+		return nexusCompletion, nil
+
+	default:
+		return nexusrpc.CompleteOperationOptions{}, serviceerror.NewInternal("no completion result provided")
+	}
+}
+
+// GetNexusCompletion returns the Nexus completion to be delivered for the callback.
+func (c *Callback) GetNexusCompletion(ctx chasm.Context, requestID string) (nexusrpc.CompleteOperationOptions, error) {
+	// Embedded callbacks use their parent component as a CompletionSource.
+	if source, ok := c.ParentCompletionSource.TryGet(ctx); ok {
+		return source.GetNexusCompletion(ctx, requestID)
+	}
+
+	// For standalone completions, get the user-supplied value and convert it into the Nexus API type.
+	suppliedCompletion, ok := c.SuppliedCompletion.TryGet(ctx)
+	if !ok {
+		return nexusrpc.CompleteOperationOptions{}, serviceerror.NewInvalidArgument("no completion result provided")
+	}
+	return callbackCompletionToNexusCompleteOperationOpts(c, suppliedCompletion)
+}
+
+// describe returns the CallbackExecutionInfo for the describe RPC. Only applies to standalone callbacks.
+func (c *Callback) describe(ctx chasm.Context) (*callbackpb.CallbackExecutionInfo, error) {
+	apiCb, err := c.ToAPICallback()
+	if err != nil {
+		return nil, err
+	}
+
+	exInfo := ctx.ExecutionInfo()
+	exKey := ctx.ExecutionKey()
+	info := &callbackpb.CallbackExecutionInfo{
+		CallbackId:              exKey.BusinessID,
+		RunId:                   exKey.RunID,
+		Callback:                apiCb,
+		Status:                  c.statusAsAPIExecutionStatus(),
+		State:                   c.statusAsAPIState(),
+		Attempt:                 c.Attempt,
+		CreateTime:              c.RegistrationTime,
+		LastAttemptCompleteTime: c.LastAttemptCompleteTime,
+		LastAttemptFailure:      c.LastAttemptFailure,
+		NextAttemptScheduleTime: c.NextAttemptScheduleTime,
+		CloseTime:               c.CloseTime,
+		ScheduleToCloseTimeout:  c.ScheduleToCloseTimeout,
+		StateTransitionCount:    exInfo.StateTransitionCount,
+	}
+	return info, nil
+}
+
+// outcome returns the callback execution outcome if the execution is in a terminal state. (Otherwise, nil.)
+//
+// IMPORTANT: This is specific to the callback delivery, and not the actual completion. The outcome will be
+// a success even if it was to deliver a failed completion result.
+func (c *Callback) outcome(ctx chasm.Context) *callbackpb.CallbackExecutionOutcome {
+	switch c.Status {
+	case callbackspb.CALLBACK_STATUS_SUCCEEDED:
+		val := &callbackpb.CallbackExecutionOutcome_Success{}
+		return &callbackpb.CallbackExecutionOutcome{
+			Value: val,
+		}
+
+	case callbackspb.CALLBACK_STATUS_FAILED:
+		// Organic failures leave TerminalFailure nil, and just set LastAttemptFailure.
+		val := &callbackpb.CallbackExecutionOutcome_Failure{
+			Failure: c.LastAttemptFailure,
+		}
+		return &callbackpb.CallbackExecutionOutcome{
+			Value: val,
+		}
+
+	case callbackspb.CALLBACK_STATUS_TERMINATED,
+		callbackspb.CALLBACK_STATUS_TIMED_OUT:
+		val := &callbackpb.CallbackExecutionOutcome_Failure{
+			Failure: c.TerminalFailure.Get(ctx),
+		}
+		return &callbackpb.CallbackExecutionOutcome{
+			Value: val,
+		}
+
+	default:
+		return nil
+	}
+}
+
+func (c *Callback) statusAsAPIExecutionStatus() enumspb.CallbackExecutionStatus {
+	switch c.Status {
+	case callbackspb.CALLBACK_STATUS_STANDBY,
+		callbackspb.CALLBACK_STATUS_SCHEDULED,
+		callbackspb.CALLBACK_STATUS_BACKING_OFF:
+		return enumspb.CALLBACK_EXECUTION_STATUS_RUNNING
+	case callbackspb.CALLBACK_STATUS_FAILED,
+		callbackspb.CALLBACK_STATUS_TIMED_OUT:
+		return enumspb.CALLBACK_EXECUTION_STATUS_FAILED
+	case callbackspb.CALLBACK_STATUS_SUCCEEDED:
+		return enumspb.CALLBACK_EXECUTION_STATUS_SUCCEEDED
+	case callbackspb.CALLBACK_STATUS_TERMINATED:
+		return enumspb.CALLBACK_EXECUTION_STATUS_TERMINATED
+	default:
+		return enumspb.CALLBACK_EXECUTION_STATUS_UNSPECIFIED
+	}
+}
+
+func (c *Callback) statusAsAPIState() enumspb.CallbackState {
+	switch c.Status {
+	case callbackspb.CALLBACK_STATUS_STANDBY:
+		return enumspb.CALLBACK_STATE_STANDBY
+	case callbackspb.CALLBACK_STATUS_SCHEDULED:
+		return enumspb.CALLBACK_STATE_SCHEDULED
+	case callbackspb.CALLBACK_STATUS_BACKING_OFF:
+		return enumspb.CALLBACK_STATE_BACKING_OFF
+	case callbackspb.CALLBACK_STATUS_FAILED:
+		return enumspb.CALLBACK_STATE_FAILED
+	case callbackspb.CALLBACK_STATUS_SUCCEEDED:
+		return enumspb.CALLBACK_STATE_SUCCEEDED
+	case callbackspb.CALLBACK_STATUS_TIMED_OUT:
+		return enumspb.CALLBACK_STATE_TIMED_OUT
+	case callbackspb.CALLBACK_STATUS_TERMINATED:
+		return enumspb.CALLBACK_STATE_TERMINATED
+	default:
+		return enumspb.CALLBACK_STATE_UNSPECIFIED
+	}
+}