[Nexus] Migrate cancellation executors to chasm

Author: gow

chasm/lib/nexusoperation/cancellation.go

@@ -1,8 +1,16 @@
 package nexusoperation
 
 import (
+	"errors"
+	"time"
+
+	"github.com/nexus-rpc/sdk-go/nexus"
+	commonpb "go.temporal.io/api/common/v1"
+	failurepb "go.temporal.io/api/failure/v1"
+	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/chasm"
-	"go.temporal.io/server/chasm/lib/nexusoperation/gen/nexusoperationpb/v1"
+	nexusoperationpb "go.temporal.io/server/chasm/lib/nexusoperation/gen/nexusoperationpb/v1"
+	"go.temporal.io/server/common/backoff"
 )
 
 var _ chasm.Component = (*Cancellation)(nil)
@@ -14,6 +22,9 @@ type Cancellation struct {
 
 	// Persisted internal state
 	*nexusoperationpb.CancellationState
+
+	// Operation is a pointer to the parent Operation component.
+	Operation chasm.ParentPtr[*Operation]
 }
 
 func newCancellation(state *nexusoperationpb.CancellationState) *Cancellation {
@@ -42,3 +53,95 @@ func (o *Cancellation) StateMachineState() nexusoperationpb.CancellationStatus {
 func (o *Cancellation) SetStateMachineState(status nexusoperationpb.CancellationStatus) {
 	o.Status = status
 }
+
+// cancelArgs holds the arguments needed to cancel a Nexus operation.
+type cancelArgs struct {
+	service                string
+	operation              string
+	token                  string
+	requestID              string
+	endpointName           string
+	endpointID             string
+	currentTime            time.Time
+	scheduledTime          time.Time
+	scheduleToCloseTimeout time.Duration
+	startToCloseTimeout    time.Duration
+	headers                map[string]string
+	payload                *commonpb.Payload
+}
+
+// loadCancelArgs is a ReadComponent callback that loads the cancel arguments from the cancellation
+// and its parent operation.
+func (c *Cancellation) loadCancelArgs(
+	ctx chasm.Context,
+	_ chasm.NoValue,
+) (cancelArgs, error) {
+	op := c.Operation.Get(ctx)
+
+	store, ok := op.Store.TryGet(ctx)
+	if !ok {
+		// TODO: For standalone operations, load invocation data from the operation state.
+		return cancelArgs{}, serviceerror.NewInternal("no store available to load invocation data")
+	}
+	invocationData, err := store.NexusOperationInvocationData(ctx, op)
+	if err != nil {
+		return cancelArgs{}, err
+	}
+
+	return cancelArgs{
+		service:                op.GetService(),
+		operation:              op.GetOperation(),
+		token:                  op.GetOperationToken(),
+		requestID:              op.GetRequestId(),
+		endpointName:           op.GetEndpoint(),
+		endpointID:             op.GetEndpointId(),
+		currentTime:            ctx.Now(c),
+		scheduledTime:          op.GetScheduledTime().AsTime(),
+		scheduleToCloseTimeout: op.GetScheduleToCloseTimeout().AsDuration(),
+		startToCloseTimeout:    op.GetStartToCloseTimeout().AsDuration(),
+		headers:                invocationData.Header,
+		payload:                invocationData.Input,
+	}, nil
+}
+
+// saveCancellationResultInput is the input to the Cancellation.applyCancellationResult method.
+type saveCancellationResultInput struct {
+	callErr     error
+	retryPolicy func() backoff.RetryPolicy
+}
+
+// applyCancellationResult applies the outcome of a cancel operation call to the cancellation state machine.
+func (c *Cancellation) applyCancellationResult(
+	ctx chasm.MutableContext,
+	input saveCancellationResultInput,
+) (chasm.NoValue, error) {
+	if input.callErr != nil {
+		var handlerErr *nexus.HandlerError
+		var opTimeoutBelowMinErr *operationTimeoutBelowMinError
+		isRetryable := !errors.As(input.callErr, &opTimeoutBelowMinErr) &&
+			(!errors.As(input.callErr, &handlerErr) || handlerErr.Retryable())
+
+		failure, err := callErrToFailure(input.callErr, isRetryable)
+		if err != nil {
+			failure = &failurepb.Failure{Message: input.callErr.Error()}
+		}
+
+		if !isRetryable {
+			// TODO: Emit EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_FAILED history event
+			// when AddHistoryEvent is available on MutableContext.
+			return nil, TransitionCancellationFailed.Apply(c, ctx, EventCancellationFailed{})
+		}
+
+		return nil, transitionCancellationAttemptFailed.Apply(c, ctx, EventCancellationAttemptFailed{
+			Failure:     failure,
+			RetryPolicy: input.retryPolicy(),
+		})
+	}
+
+	// Cancellation request transmitted successfully.
+	// The operation is not yet canceled and may ignore our request, the outcome will be known via the
+	// completion callback.
+	// TODO: Emit EVENT_TYPE_NEXUS_OPERATION_CANCEL_REQUEST_COMPLETED history event
+	// when AddHistoryEvent is available on MutableContext.
+	return nil, TransitionCancellationSucceeded.Apply(c, ctx, EventCancellationSucceeded{})
+}