Nexus Standalone: wire up

Author: stephanos

chasm/lib/nexusoperation/frontend.go

@@ -85,13 +85,14 @@ func (h *frontendHandler) StartNexusOperationExecution(
 		return nil, err
 	}
 
-	// Verify the endpoint exists before creating the operation.
-	if _, err := h.endpointRegistry.GetByName(ctx, namespaceID, req.GetEndpoint()); err != nil {
+	endpoint, err := h.endpointRegistry.GetByName(ctx, namespaceID, req.GetEndpoint())
+	if err != nil {
 		return nil, err
 	}
 
 	resp, err := h.client.StartNexusOperation(ctx, &nexusoperationpb.StartNexusOperationRequest{
 		NamespaceId:     namespaceID.String(),
+		EndpointId:      endpoint.Id,
 		FrontendRequest: req,
 	})
 	return resp.GetFrontendResponse(), err

chasm/lib/nexusoperation/gen/nexusoperationpb/v1/operation.pb.go

@@ -15,6 +15,7 @@ import (
 	"go.temporal.io/server/chasm"
 	nexusoperationpb "go.temporal.io/server/chasm/lib/nexusoperation/gen/nexusoperationpb/v1"
 	"go.temporal.io/server/common/backoff"
+	"go.temporal.io/server/common/softassert"
 	queueserrors "go.temporal.io/server/service/history/queues/errors"
 	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -74,6 +75,7 @@ func newStandaloneOperation(
 ) (*Operation, error) {
 	frontendReq := req.GetFrontendRequest()
 	op := NewOperation(&nexusoperationpb.OperationState{
+		EndpointId:             req.GetEndpointId(),
 		Endpoint:               frontendReq.GetEndpoint(),
 		Service:                frontendReq.GetService(),
 		Operation:              frontendReq.GetOperation(),
@@ -156,41 +158,29 @@ func (o *Operation) onStarted(ctx chasm.MutableContext, operationToken string, l
 	if store, ok := o.Store.TryGet(ctx); ok {
 		return store.OnNexusOperationStarted(ctx, o, operationToken, links)
 	}
+	o.Links = append(o.Links, links...)
 	return TransitionStarted.Apply(o, ctx, EventStarted{OperationToken: operationToken})
 }
 
 func (o *Operation) onCompleted(ctx chasm.MutableContext, result *commonpb.Payload, links []*commonpb.Link) error {
 	if store, ok := o.Store.TryGet(ctx); ok {
 		return store.OnNexusOperationCompleted(ctx, o, result, links)
 	}
-	outcome := o.outcome(ctx)
-	outcome.Variant = &nexusoperationpb.OperationOutcome_Successful_{
-		Successful: &nexusoperationpb.OperationOutcome_Successful{Result: result},
-	}
-	return TransitionSucceeded.Apply(o, ctx, EventSucceeded{})
+	o.Links = append(o.Links, links...)
+	return TransitionSucceeded.Apply(o, ctx, EventSucceeded{Result: result})
 }
 
 func (o *Operation) onFailed(ctx chasm.MutableContext, cause *failurepb.Failure) error {
 	if store, ok := o.Store.TryGet(ctx); ok {
 		return store.OnNexusOperationFailed(ctx, o, cause)
 	}
-	if cause != nil {
-		o.outcome(ctx).Variant = &nexusoperationpb.OperationOutcome_Failed_{
-			Failed: &nexusoperationpb.OperationOutcome_Failed{Failure: cause},
-		}
-	}
 	return TransitionFailed.Apply(o, ctx, EventFailed{Failure: cause})
 }
 
 func (o *Operation) onCanceled(ctx chasm.MutableContext, cause *failurepb.Failure) error {
 	if store, ok := o.Store.TryGet(ctx); ok {
 		return store.OnNexusOperationCanceled(ctx, o, cause)
 	}
-	if cause != nil {
-		o.outcome(ctx).Variant = &nexusoperationpb.OperationOutcome_Failed_{
-			Failed: &nexusoperationpb.OperationOutcome_Failed{Failure: cause},
-		}
-	}
 	return TransitionCanceled.Apply(o, ctx, EventCanceled{Failure: cause})
 }
 
@@ -389,14 +379,25 @@ func (o *Operation) buildPollResponse(
 }
 
 func (o *Operation) describeOutcome(ctx chasm.Context) (*commonpb.Payload, *failurepb.Failure) {
-	outcome := o.Outcome.Get(ctx)
-	if successful := outcome.GetSuccessful(); successful != nil {
-		return successful.GetResult(), nil
-	}
-	if failure := outcome.GetFailed().GetFailure(); failure != nil {
-		return nil, failure
+	outcome, hasOutcome := o.Outcome.TryGet(ctx)
+
+	switch {
+	case !hasOutcome:
+		return nil, o.LastAttemptFailure
+	case outcome.GetSuccessful() != nil:
+		return outcome.GetSuccessful().GetResult(), nil
+	case outcome.GetFailed() != nil:
+		// Timeouts often close after one or more retryable attempt failures. For describe/poll, prefer the
+		// last concrete attempt failure over the generic timeout wrapper because it is more actionable.
+		// Note that LastAttemptFailure may be nil if the operation timed out before any attempt.
+		if o.Status == nexusoperationpb.OPERATION_STATUS_TIMED_OUT && o.LastAttemptFailure != nil {
+			return nil, o.LastAttemptFailure
+		}
+		return nil, outcome.GetFailed().GetFailure()
+	default:
+		softassert.Fail(ctx.Logger(), "operation outcome has no variant set")
+		return nil, nil
 	}
-	return nil, o.LastAttemptFailure
 }
 
 func (o *Operation) isWaitStageReached(_ chasm.Context, waitStage enumspb.NexusOperationWaitStage) bool {
@@ -441,6 +442,7 @@ func (o *Operation) buildExecutionInfo(ctx chasm.Context) *nexuspb.NexusOperatio
 		},
 		NexusHeader:  requestData.GetNexusHeader(),
 		UserMetadata: requestData.GetUserMetadata(),
+		Links:        o.Links,
 		Identity:     requestData.GetIdentity(),
 	}
 

chasm/lib/nexusoperation/gen/nexusoperationpb/v1/request_response.pb.go

@@ -3,6 +3,7 @@ package nexusoperation
 import (
 	"time"
 
+	commonpb "go.temporal.io/api/common/v1"
 	failurepb "go.temporal.io/api/failure/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/nexusoperation/gen/nexusoperationpb/v1"
@@ -160,7 +161,7 @@ type EventSucceeded struct {
 	// If not nil, uses the provided time instead of the current component time.
 	// Used when a completion comes in before start is recorded (rare race).
 	CompleteTime *time.Time
-	// TODO(stephan): Store result
+	Result       *commonpb.Payload
 }
 
 var TransitionSucceeded = chasm.NewTransition(
@@ -179,6 +180,15 @@ var TransitionSucceeded = chasm.NewTransition(
 		}
 		o.NextAttemptScheduleTime = nil
 		o.ClosedTime = timestamppb.New(closeTime)
+
+		// Result is set by standalone operations. Workflow-attached operations store
+		// the result in the history event and remove the operation after this transition.
+		if event.Result != nil {
+			o.outcome(ctx).Variant = &nexusoperationpb.OperationOutcome_Successful_{
+				Successful: &nexusoperationpb.OperationOutcome_Successful{Result: event.Result},
+			}
+		}
+
 		// Terminal state - no tasks to emit.
 		return nil
 	},
@@ -252,8 +262,7 @@ var TransitionTerminated = chasm.NewTransition(
 		}
 		o.ClosedTime = timestamppb.New(ctx.Now(o))
 		o.NextAttemptScheduleTime = nil
-		outcome := o.Outcome.Get(ctx)
-		outcome.Variant = &nexusoperationpb.OperationOutcome_Failed_{
+		o.outcome(ctx).Variant = &nexusoperationpb.OperationOutcome_Failed_{
 			Failed: &nexusoperationpb.OperationOutcome_Failed{
 				Failure: &failurepb.Failure{
 					Message: event.Reason,

chasm/lib/nexusoperation/operation.go

@@ -22,7 +22,6 @@ var (
 )
 
 func newTestOperation() *Operation {
-	ctx := &chasm.MockMutableContext{}
 	op := &Operation{
 		OperationState: &nexusoperationpb.OperationState{
 			Status:                 nexusoperationpb.OPERATION_STATUS_UNSPECIFIED,
@@ -36,7 +35,6 @@ func newTestOperation() *Operation {
 			Attempt:                0,
 		},
 	}
-	op.Outcome = chasm.NewDataField(ctx, &nexusoperationpb.OperationOutcome{})
 	return op
 }
 

chasm/lib/nexusoperation/operation_statemachine.go

@@ -169,6 +169,11 @@ func (h *operationInvocationTaskHandler) Execute(
 		header.Set(nexusrpc.HeaderTemporalNexusFailureSupport, "true")
 	}
 
+	var links []nexus.Link
+	if args.nexusLink != (nexus.Link{}) {
+		links = []nexus.Link{args.nexusLink}
+	}
+
 	callCtx, cancel := context.WithTimeout(ctx, callTimeout)
 	defer cancel()
 	// Set this value on the parent context so that our custom HTTP caller can mutate it since we cannot
@@ -182,7 +187,7 @@ func (h *operationInvocationTaskHandler) Execute(
 		CallbackHeader: nexus.Header{
 			commonnexus.CallbackTokenHeader: token,
 		},
-		Links: []nexus.Link{args.nexusLink},
+		Links: links,
 	}
 
 	invocation, err := h.newInvocation(callCtx, ns, endpoint, opRef, args, task, callTimeout, timeoutType)