diff --git a/service/history/configs/config.go b/service/history/configs/config.go index 44e4de8dc07..96685a1544a 100644 --- a/service/history/configs/config.go +++ b/service/history/configs/config.go @@ -2,6 +2,7 @@ package configs import ( "go.temporal.io/server/chasm/lib/callback" + "go.temporal.io/server/chasm/lib/nexusoperation" "go.temporal.io/server/common" "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/namespace" @@ -69,6 +70,7 @@ type Config struct { MaxCallbacksPerExecution dynamicconfig.IntPropertyFnWithNamespaceFilter MaxCallbacksPerUpdateID dynamicconfig.IntPropertyFnWithNamespaceFilter EnableChasm dynamicconfig.BoolPropertyFnWithNamespaceFilter + EnableChasmNexusWorkflowOperations dynamicconfig.BoolPropertyFnWithNamespaceFilter EnableCHASMCallbacks dynamicconfig.BoolPropertyFnWithNamespaceFilter EnableCHASMSignalBacklinks dynamicconfig.BoolPropertyFnWithNamespaceFilter EnableWorkflowUpdateCallbacks dynamicconfig.BoolPropertyFnWithNamespaceFilter @@ -494,6 +496,7 @@ func NewConfig( MaxCallbacksPerExecution: callback.MaxPerExecution.Get(dc), MaxCallbacksPerUpdateID: dynamicconfig.MaxCallbacksPerUpdateID.Get(dc), EnableChasm: dynamicconfig.EnableChasm.Get(dc), + EnableChasmNexusWorkflowOperations: nexusoperation.EnableChasmWorkflowOperations.Get(dc), ChasmMaxInMemoryPureTasks: dynamicconfig.ChasmMaxInMemoryPureTasks.Get(dc), EnableCHASMSchedulerCreation: dynamicconfig.EnableCHASMSchedulerCreation.Get(dc), diff --git a/service/history/handler.go b/service/history/handler.go index a1f24db4505..79a7dc955b1 100644 --- a/service/history/handler.go +++ b/service/history/handler.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "math" + "strconv" "sync" "sync/atomic" "time" @@ -26,6 +27,7 @@ import ( tokenspb "go.temporal.io/server/api/token/v1" "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/lib/activity" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/client/history" "go.temporal.io/server/common" "go.temporal.io/server/common/archiver" @@ -2165,11 +2167,125 @@ func (h *Handler) CompleteNexusOperation(ctx context.Context, request *historyse opErr, ) if err != nil { + // Cross-tree fallback (HSM token -> CHASM op). The completion token is an HSM StateMachineRef, + // but a workflow reset can rebuild the op into the CHASM tree (reset routes op creation by the + // current nexusoperation.enableChasmWorkflowOperations policy; see mutable_state_rebuilder.go). + // When that happens the HSM completion path returns NotFound. Both trees key ops by + // ScheduledEventId, which we recover from the HSM ref, then complete the op in the CHASM tree. + if errors.As(err, new(*serviceerror.NotFound)) { + if fallbackErr := h.completeNexusOperationChasmFallback(ctx, request); fallbackErr != nil { + // Preserve the original HSM NotFound when the op is in neither tree. + // Surface any other fallback error as-is. + if errors.As(fallbackErr, new(*serviceerror.NotFound)) { + return nil, h.convertError(err) + } + return nil, h.convertError(fallbackErr) + } + return &historyservice.CompleteNexusOperationResponse{}, nil + } return nil, h.convertError(err) } return &historyservice.CompleteNexusOperationResponse{}, nil } +// completeNexusOperationChasmFallback completes a Nexus operation that lives in the CHASM tree, given +// an HSM-format completion request. It recovers the ScheduledEventId from the HSM StateMachineRef and +// resolves the op via the root CHASM Workflow component (Operations[scheduledEventId]). +// +// It returns a *serviceerror.NotFound if the op does not exist in the CHASM tree either +func (h *Handler) completeNexusOperationChasmFallback( + ctx context.Context, + request *historyservice.CompleteNexusOperationRequest, +) error { + // Only attempt CHASM resolution when CHASM is enabled for the namespace + const msgOperationNotFound = "operation not found" + nsEntry, err := h.namespaceRegistry.GetNamespaceByID(namespace.ID(request.GetCompletion().GetNamespaceId())) + if err != nil { + return serviceerror.NewNotFound(msgOperationNotFound) + } + if !h.config.EnableChasm(nsEntry.Name().String()) { + return serviceerror.NewNotFound(msgOperationNotFound) + } + + scheduledEventID, err := scheduledEventIDFromStateMachineRef(request.GetCompletion().GetRef()) + if err != nil { + // Not recoverable from this token; nothing to fall back to. + return serviceerror.NewNotFound(msgOperationNotFound) + } + + // Require the token's schedule-time run to be a real run of this execution before re-targeting the + // current run, mirroring the CHASM-token path (completeNexusOperationAfterReset). A bogus/tampered + // run ID (with an otherwise valid event/request ID) must keep its NotFound rather than completing the + // current-run op. + namespaceID := request.GetCompletion().GetNamespaceId() + businessID := request.GetCompletion().GetWorkflowId() + if !h.scheduleTimeRunExists(ctx, chasm.ExecutionKey{ + NamespaceID: namespaceID, + BusinessID: businessID, + RunID: request.GetCompletion().GetRunId(), + }) { + return serviceerror.NewNotFound(msgOperationNotFound) + } + + completion, err := chasmNexusCompletionFromHSMRequest(request) + if err != nil { + return err + } + + // Complete the op on the CURRENT run (empty RunID). The token's RunID refers to the run at scheduling + // time, but a reset (the very scenario that moved the op to the CHASM tree) creates a new current run. + return h.applyChasmNexusCompletionOnCurrentRun( + ctx, + chasm.ExecutionKey{ + NamespaceID: namespaceID, + BusinessID: businessID, + }, + scheduledEventID, + completion, + ) +} + +// chasmNexusCompletionFromHSMRequest converts an HSM-format CompleteNexusOperationRequest into the +// CHASM ChasmNexusCompletion message consumed by Operation.HandleNexusCompletion. +func chasmNexusCompletionFromHSMRequest( + request *historyservice.CompleteNexusOperationRequest, +) (*persistencespb.ChasmNexusCompletion, error) { + completion := &persistencespb.ChasmNexusCompletion{ + StartTime: request.GetStartTime(), + RequestId: request.GetCompletion().GetRequestId(), + Links: request.GetLinks(), + OperationToken: request.GetOperationToken(), + } + if request.GetState() == string(nexus.OperationStateSucceeded) { + completion.Outcome = &persistencespb.ChasmNexusCompletion_Success{Success: request.GetSuccess()} + return completion, nil + } + temporalFailure, err := commonnexus.NexusFailureToTemporalFailure( + commonnexus.ProtoFailureToNexusFailure(request.GetFailure()), + ) + if err != nil { + return nil, serviceerror.NewInvalidArgument("unable to convert failure") + } + completion.Outcome = &persistencespb.ChasmNexusCompletion_Failure{Failure: temporalFailure} + return completion, nil +} + +// scheduledEventIDFromStateMachineRef extracts the ScheduledEventId from an HSM StateMachineRef. The +// op node is the last key in the path, whose Id is the ScheduledEventId encoded as a decimal string +// (see components/nexusoperations/events.go). +func scheduledEventIDFromStateMachineRef(ref *persistencespb.StateMachineRef) (int64, error) { + path := ref.GetPath() + if len(path) == 0 { + return 0, serviceerror.NewInvalidArgument("state machine ref has empty path") + } + last := path[len(path)-1] + id, err := strconv.ParseInt(last.GetId(), 10, 64) + if err != nil { + return 0, serviceerror.NewInvalidArgumentf("state machine key id %q is not a valid scheduled event ID", last.GetId()) + } + return id, nil +} + func (h *Handler) CompleteNexusOperationChasm( ctx context.Context, request *historyservice.CompleteNexusOperationChasmRequest, @@ -2226,12 +2342,264 @@ func (h *Handler) CompleteNexusOperationChasm( }, completion) if err != nil { + // The completion token's ComponentRef encodes the run as of schedule time. A workflow reset closes that run and + // rebuilds the op on a new current run; keeping the op in the CHASM tree (flag on) or moving it to the HSM tree + // (flag off, or a CHASM create error; see mutable_state_rebuilder.go). So a completion arriving after a reset + // fails against that now-closed schedule-time run. Resolve the op by ScheduledEventId (stable across reset and + // across trees) on the CURRENT run and complete it wherever the rebuild placed it. + if resolved, fallbackErr := h.completeNexusOperationAfterReset(ctx, ref, request, completion); resolved { + if fallbackErr != nil { + // The op was located on the current run but the completion was rejected (e.g. request-id + // mismatch -> NotFound), or it turned out to be in neither tree. Preserve the original + // CHASM error for a stable "operation not found"; surface any other fallback error as-is. + if errors.As(fallbackErr, new(*serviceerror.NotFound)) { + return nil, h.convertError(err) + } + return nil, h.convertError(fallbackErr) + } + return &historyservice.CompleteNexusOperationChasmResponse{}, nil + } return nil, h.convertError(err) } return &historyservice.CompleteNexusOperationChasmResponse{}, nil } +// completeNexusOperationAfterReset re-resolves a CHASM completion against the CURRENT run after the schedule-time +// (token) run was closed by a reset, and completes the op wherever the rebuild placed it. It recovers the +// ScheduledEventId from the ComponentRef and, depending on where the op now lives, completes it in the CHASM tree on +// the current run or falls back to the HSM tree. +// Returns resolved=false when there is nothing to fall back to, so the caller keeps the original error. +func (h *Handler) completeNexusOperationAfterReset( + ctx context.Context, + ref *persistencespb.ChasmComponentRef, + request *historyservice.CompleteNexusOperationChasmRequest, + completion *persistencespb.ChasmNexusCompletion, +) (resolved bool, err error) { + scheduledEventID, err := scheduledEventIDFromComponentPath(ref.GetComponentPath()) + if err != nil { + // Not recoverable from this token; nothing to fall back to. + return false, nil + } + // A legitimate Nexus completion token references the op under the root CHASM Workflow component, so + // its ref carries the Workflow archetype. The fallback below re-resolves the op against the Workflow + // root (lookupChasmOpOnCurrentRun / applyChasmNexusCompletionOnCurrentRun), which would silently + // ignore the token's claimed archetype. Reject any other archetype here so a wrong-archetype token + // keeps its original NotFound rather than being completed against the Workflow tree. + if ref.GetArchetypeId() != chasm.WorkflowArchetypeID { + return false, nil + } + // The namespace and business IDs come from the ChasmComponentRef: a CHASM completion token populates + // the marshaled ComponentRef rather than the legacy NamespaceId/WorkflowId fields on the completion. + namespaceID := ref.GetNamespaceId() + businessID := ref.GetBusinessId() + // Guard against driving CHASM reads/writes on an execution with no real CHASM tree (a noopChasmTree). + nsEntry, nsErr := h.namespaceRegistry.GetNamespaceByID(namespace.ID(namespaceID)) + if nsErr != nil || !h.config.EnableChasm(nsEntry.Name().String()) { + return false, nil + } + // A token whose schedule-time run does not exist (e.g. a bogus run ID) is not a reset case and must keep its + // original NotFound rather than being re-targeted at the current run. + if !h.scheduleTimeRunExists(ctx, chasm.ExecutionKey{NamespaceID: namespaceID, BusinessID: businessID, RunID: ref.GetRunId()}) { + return false, nil + } + // An empty RunID resolves the current (reset-created) run. + currentRun := chasm.ExecutionKey{NamespaceID: namespaceID, BusinessID: businessID} + present, lookupErr := h.lookupChasmOpOnCurrentRun(ctx, currentRun, scheduledEventID) + if lookupErr != nil { + // Current run unreadable (execution gone / not CHASM-enabled): nothing to fall back to, so the + // caller keeps the original completion error. + return false, nil + } + if present { + return true, h.applyChasmNexusCompletionOnCurrentRun(ctx, currentRun, scheduledEventID, completion) + } + // Current run is readable but the op is absent from its CHASM tree: it was rebuilt into the HSM tree + // (or it exists in neither, in which case the HSM fallback returns NotFound). + return true, h.completeNexusOperationHSMFallback(ctx, ref, request) +} + +// scheduleTimeRunExists reports whether the run identified by execKey (the run as of schedule time, taken from the +// completion token's ChasmComponentRef) is a real run of the execution. A reset closes that run but leaves it +// readable; a bogus/garbage run ID does not resolve at all. Reads do not trip the events-after-workflow-finish write +// validation, so a closed (reset) run reads back successfully while a nonexistent run returns NotFound. An empty RunID +// is treated as not-a-reset (there is no stale schedule-time run to recover from). +func (h *Handler) scheduleTimeRunExists(ctx context.Context, execKey chasm.ExecutionKey) bool { + if execKey.RunID == "" { + return false + } + _, err := chasm.ReadComponent( + ctx, + chasm.NewComponentRef[*chasmworkflow.Workflow](execKey), + func(*chasmworkflow.Workflow, chasm.Context, chasm.NoValue) (chasm.NoValue, error) { + return nil, nil + }, + nil, + ) + return err == nil +} + +// lookupChasmOpOnCurrentRun reports whether the Nexus op identified by scheduledEventID is present in the CHASM tree of +// the execution identified by execKey. Callers pass an execKey with an empty RunID so the lookup resolves the CURRENT +// run. A non-nil error means the run could not be read (execution gone / not CHASM-enabled); callers decline the +// fallback and keep the original error in that case. +func (h *Handler) lookupChasmOpOnCurrentRun( + ctx context.Context, + execKey chasm.ExecutionKey, + scheduledEventID int64, +) (bool, error) { + return chasm.ReadComponent( + ctx, + chasm.NewComponentRef[*chasmworkflow.Workflow](execKey), + func(wf *chasmworkflow.Workflow, _ chasm.Context, _ chasm.NoValue) (bool, error) { + _, ok := wf.Operations[scheduledEventID] + return ok, nil + }, + nil, + ) +} + +// applyChasmNexusCompletionOnCurrentRun completes the op at Operations[scheduledEventID] in the CHASM tree of the +// execution identified by execKey. Callers pass an execKey with an empty RunID so the completion targets the CURRENT +// run. It is shared by the HSM-token -> CHASM-op fallback (completeNexusOperationChasmFallback) and the after-reset +// CHASM-token -> CHASM-op path (completeNexusOperationAfterReset). +func (h *Handler) applyChasmNexusCompletionOnCurrentRun( + ctx context.Context, + execKey chasm.ExecutionKey, + scheduledEventID int64, + completion *persistencespb.ChasmNexusCompletion, +) error { + _, _, err := chasm.UpdateComponent( + ctx, + chasm.NewComponentRef[*chasmworkflow.Workflow](execKey), + func(wf *chasmworkflow.Workflow, mutableCtx chasm.MutableContext, completion *persistencespb.ChasmNexusCompletion) (chasm.NoValue, error) { + field, ok := wf.Operations[scheduledEventID] + if !ok { + // Op is not in the CHASM tree on the current run. + return nil, serviceerror.NewNotFound("operation not found") + } + op := field.Get(mutableCtx) + return nil, op.HandleNexusCompletion(mutableCtx, completion) + }, + completion, + ) + return err +} + +// completeNexusOperationHSMFallback completes a Nexus operation that has been rebuilt into the HSM +// tree, given a CHASM-format completion request. It recovers the ScheduledEventId from the +// ComponentRef and drives the existing HSM completion handler with a synthesized StateMachineRef. +func (h *Handler) completeNexusOperationHSMFallback( + ctx context.Context, + ref *persistencespb.ChasmComponentRef, + request *historyservice.CompleteNexusOperationChasmRequest, +) error { + scheduledEventID, err := scheduledEventIDFromComponentPath(ref.GetComponentPath()) + if err != nil { + return serviceerror.NewNotFound("operation not found") + } + + namespaceID := ref.GetNamespaceId() + shardContext, err := h.controller.GetShardByNamespaceWorkflow( + namespace.ID(namespaceID), + ref.GetBusinessId(), + ) + if err != nil { + return err + } + engine, err := shardContext.GetEngine(ctx) + if err != nil { + return err + } + + opErr, err := nexusOperationErrorFromChasmRequest(request) + if err != nil { + return err + } + + // RunID is intentionally empty: the op moved to HSM via a reset that created a new current run. + // The HSM completion handler resolves the current run and tolerates the run having changed + // since the token was minted. + hsmRef := hsm.Ref{ + WorkflowKey: definition.NewWorkflowKey(namespaceID, ref.GetBusinessId(), ""), + StateMachineRef: &persistencespb.StateMachineRef{ + Path: []*persistencespb.StateMachineKey{ + { + Type: nexusoperations.OperationMachineType, + Id: strconv.FormatInt(scheduledEventID, 10), + }, + }, + }, + } + return h.nexusCompletionHandler.Handle( + ctx, + engine.StateMachineEnvironment(metrics.OperationTag(metrics.HistoryCompleteNexusOperationScope)), + hsmRef, + request.GetCompletion().GetRequestId(), + request.GetOperationToken(), + request.GetStartTime(), + request.GetLinks(), + request.GetSuccess(), + opErr, + ) +} + +// nexusOperationErrorFromChasmRequest converts a CHASM completion request's outcome into the +// *nexus.OperationError the HSM completion handler expects (nil for a successful completion). The +// CHASM request carries a Temporal failure; the HSM path expects a Nexus OperationError. +func nexusOperationErrorFromChasmRequest( + request *historyservice.CompleteNexusOperationChasmRequest, +) (*nexus.OperationError, error) { + failure, ok := request.GetOutcome().(*historyservice.CompleteNexusOperationChasmRequest_Failure) + if !ok { + // Success (or no failure set): no operation error. + return nil, nil + } + nexusFailure, err := commonnexus.TemporalFailureToNexusFailure(failure.Failure) + if err != nil { + return nil, serviceerror.NewInvalidArgument("unable to convert failure") + } + recvdErr, err := nexusrpc.DefaultFailureConverter().FailureToError(nexusFailure) + if err != nil { + return nil, serviceerror.NewInvalidArgument("unable to convert failure to error") + } + opErr, ok := recvdErr.(*nexus.OperationError) + if !ok { + // The CHASM request has no explicit outcome state, so a failure that doesn't already carry a + // Nexus OperationError must be classified here. Mirror the native CHASM path + // (Operation.HandleNexusCompletion) and treat a Temporal cancellation as Canceled rather than + // defaulting everything to Failed, so a cross-tree completion records the same terminal state + // (canceled vs failed) as an in-CHASM completion. + state := nexus.OperationStateFailed + if failure.Failure.GetCanceledFailureInfo() != nil { + state = nexus.OperationStateCanceled + } + opErr = &nexus.OperationError{ + State: state, + Message: "nexus operation completed unsuccessfully", + Cause: recvdErr, + } + if err := nexusrpc.MarkAsWrapperError(nexusrpc.DefaultFailureConverter(), opErr); err != nil { + return nil, serviceerror.NewInvalidArgument("unable to convert operation error to failure") + } + } + return opErr, nil +} + +// scheduledEventIDFromComponentPath extracts the ScheduledEventId (the chasm.Map[int64] key) from a +// CHASM ComponentRef component path of the form ["Operations", ""]. +func scheduledEventIDFromComponentPath(componentPath []string) (int64, error) { + if len(componentPath) == 0 { + return 0, serviceerror.NewInvalidArgument("component ref has empty component path") + } + last := componentPath[len(componentPath)-1] + id, err := strconv.ParseInt(last, 10, 64) + if err != nil { + return 0, serviceerror.NewInvalidArgumentf("component path segment %q is not a valid scheduled event ID", last) + } + return id, nil +} + // convertError is a helper method to convert ShardOwnershipLostError from persistence layer returned by various // HistoryEngine API calls to ShardOwnershipLost error return by HistoryService for client to be redirected to the // correct shard. diff --git a/service/history/handler_test.go b/service/history/handler_test.go index 9b8852d9b0d..62898b92338 100644 --- a/service/history/handler_test.go +++ b/service/history/handler_test.go @@ -5,9 +5,15 @@ import ( "errors" "testing" + "github.com/nexus-rpc/sdk-go/nexus" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + commonpb "go.temporal.io/api/common/v1" + failurepb "go.temporal.io/api/failure/v1" + nexuspb "go.temporal.io/api/nexus/v1" "go.temporal.io/server/api/historyservice/v1" persistencespb "go.temporal.io/server/api/persistence/v1" + tokenspb "go.temporal.io/server/api/token/v1" "go.temporal.io/server/common/log" "go.temporal.io/server/common/membership" "go.temporal.io/server/common/metrics" @@ -70,3 +76,170 @@ func TestDescribeHistoryHost(t *testing.T) { }) assert.NoError(t, err) } + +func TestScheduledEventIDFromStateMachineRef(t *testing.T) { + testCases := []struct { + name string + ref *persistencespb.StateMachineRef + wantID int64 + wantErr bool + }{ + { + name: "recovers id from last key", + ref: &persistencespb.StateMachineRef{Path: []*persistencespb.StateMachineKey{{Type: "nexusoperations.Operation", Id: "42"}}}, + wantID: 42, + }, + { + name: "uses the last (deepest) key", + ref: &persistencespb.StateMachineRef{Path: []*persistencespb.StateMachineKey{ + {Type: "nexusoperations.Operation", Id: "7"}, + {Type: "nexusoperations.Cancelation", Id: "99"}, + }}, + wantID: 99, + }, + { + name: "empty path is an error", + ref: &persistencespb.StateMachineRef{}, + wantErr: true, + }, + { + name: "non-numeric id is an error", + ref: &persistencespb.StateMachineRef{Path: []*persistencespb.StateMachineKey{{Type: "x", Id: "not-a-number"}}}, + wantErr: true, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + id, err := scheduledEventIDFromStateMachineRef(tc.ref) + if tc.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + require.Equal(t, tc.wantID, id) + }) + } +} + +func TestChasmNexusCompletionFromHSMRequest(t *testing.T) { + testCases := []struct { + name string + request *historyservice.CompleteNexusOperationRequest + verify func(t *testing.T, completion *persistencespb.ChasmNexusCompletion) + }{ + { + name: "success carries the payload", + request: &historyservice.CompleteNexusOperationRequest{ + State: string(nexus.OperationStateSucceeded), + OperationToken: "op-token", + Completion: &tokenspb.NexusOperationCompletion{RequestId: "req-1"}, + Outcome: &historyservice.CompleteNexusOperationRequest_Success{Success: &commonpb.Payload{Data: []byte("ok")}}, + }, + verify: func(t *testing.T, completion *persistencespb.ChasmNexusCompletion) { + require.Equal(t, "req-1", completion.GetRequestId()) + require.Equal(t, "op-token", completion.GetOperationToken()) + require.NotNil(t, completion.GetSuccess()) + require.Equal(t, []byte("ok"), completion.GetSuccess().GetData()) + }, + }, + { + name: "failure is converted to a temporal failure", + request: &historyservice.CompleteNexusOperationRequest{ + State: string(nexus.OperationStateFailed), + Completion: &tokenspb.NexusOperationCompletion{RequestId: "req-2"}, + Outcome: &historyservice.CompleteNexusOperationRequest_Failure{Failure: &nexuspb.Failure{Message: "boom"}}, + }, + verify: func(t *testing.T, completion *persistencespb.ChasmNexusCompletion) { + require.NotNil(t, completion.GetFailure()) + require.Equal(t, "boom", completion.GetFailure().GetMessage()) + }, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + completion, err := chasmNexusCompletionFromHSMRequest(tc.request) + require.NoError(t, err) + tc.verify(t, completion) + }) + } +} + +func TestScheduledEventIDFromComponentPath(t *testing.T) { + testCases := []struct { + name string + path []string + wantID int64 + wantErr bool + }{ + {name: "recovers id from Operations path", path: []string{"Operations", "42"}, wantID: 42}, + {name: "uses the last segment", path: []string{"Operations", "7", "13"}, wantID: 13}, + {name: "empty path is an error", path: nil, wantErr: true}, + {name: "non-numeric segment is an error", path: []string{"Operations", "nope"}, wantErr: true}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + id, err := scheduledEventIDFromComponentPath(tc.path) + if tc.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + require.Equal(t, tc.wantID, id) + }) + } +} + +func TestNexusOperationErrorFromChasmRequest(t *testing.T) { + testCases := []struct { + name string + request *historyservice.CompleteNexusOperationChasmRequest + wantOpErr bool + wantState nexus.OperationState // checked only when wantOpErr is true + }{ + { + name: "success yields no operation error", + request: &historyservice.CompleteNexusOperationChasmRequest{ + Completion: &tokenspb.NexusOperationCompletion{RequestId: "r"}, + Outcome: &historyservice.CompleteNexusOperationChasmRequest_Success{Success: &commonpb.Payload{}}, + }, + wantOpErr: false, + }, + { + name: "generic failure yields a failed operation error", + request: &historyservice.CompleteNexusOperationChasmRequest{ + Completion: &tokenspb.NexusOperationCompletion{RequestId: "r"}, + Outcome: &historyservice.CompleteNexusOperationChasmRequest_Failure{Failure: &failurepb.Failure{Message: "boom"}}, + }, + wantOpErr: true, + wantState: nexus.OperationStateFailed, + }, + { + name: "canceled failure yields a canceled operation error", + request: &historyservice.CompleteNexusOperationChasmRequest{ + Completion: &tokenspb.NexusOperationCompletion{RequestId: "r"}, + Outcome: &historyservice.CompleteNexusOperationChasmRequest_Failure{Failure: &failurepb.Failure{ + Message: "canceled", + FailureInfo: &failurepb.Failure_CanceledFailureInfo{CanceledFailureInfo: &failurepb.CanceledFailureInfo{}}, + }}, + }, + wantOpErr: true, + wantState: nexus.OperationStateCanceled, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + opErr, err := nexusOperationErrorFromChasmRequest(tc.request) + require.NoError(t, err) + if tc.wantOpErr { + require.NotNil(t, opErr) + require.Equal(t, tc.wantState, opErr.State) + } else { + require.Nil(t, opErr) + } + }) + } +} diff --git a/service/history/history_engine.go b/service/history/history_engine.go index e8b8789bec1..e0c8d1dfa94 100644 --- a/service/history/history_engine.go +++ b/service/history/history_engine.go @@ -244,7 +244,8 @@ func NewEngineWithShardContext( historyEngImpl.queueProcessors[processor.Category()] = processor } - historyEngImpl.eventsReapplier = ndc.NewEventsReapplier(shard.StateMachineRegistry(), shard.GetMetricsHandler(), logger) + historyEngImpl.eventsReapplier = ndc.NewEventsReapplier(shard.StateMachineRegistry(), shard.GetMetricsHandler(), logger). + WithChasmWorkflowRegistry(shard.ChasmWorkflowRegistry()) if shard.GetClusterMetadata().IsGlobalNamespaceEnabled() { historyEngImpl.replicationAckMgr = replication.NewAckManager( diff --git a/service/history/interfaces/shard_context.go b/service/history/interfaces/shard_context.go index 1bebd47c066..68a84fbf289 100644 --- a/service/history/interfaces/shard_context.go +++ b/service/history/interfaces/shard_context.go @@ -10,6 +10,7 @@ import ( "go.temporal.io/server/api/historyservice/v1" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/chasm" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/common/archiver" "go.temporal.io/server/common/clock" "go.temporal.io/server/common/cluster" @@ -119,6 +120,8 @@ type ( GetFinalizer() *finalizer.Finalizer ChasmRegistry() *chasm.Registry + // ChasmWorkflowRegistry returns the CHASM workflow library's event/command registry. + ChasmWorkflowRegistry() *chasmworkflow.Registry EndpointRegistry() chasm.EndpointRegistry BusinessIDReuseRateLimiter(namespaceID namespace.ID, businessID string, archetypeID chasm.ArchetypeID) quotas.RateLimiter diff --git a/service/history/interfaces/shard_context_mock.go b/service/history/interfaces/shard_context_mock.go index 3e00c0663a2..4d767ac5a63 100644 --- a/service/history/interfaces/shard_context_mock.go +++ b/service/history/interfaces/shard_context_mock.go @@ -20,6 +20,7 @@ import ( historyservice "go.temporal.io/server/api/historyservice/v1" persistence "go.temporal.io/server/api/persistence/v1" chasm "go.temporal.io/server/chasm" + workflow "go.temporal.io/server/chasm/lib/workflow" archiver "go.temporal.io/server/common/archiver" clock0 "go.temporal.io/server/common/clock" cluster "go.temporal.io/server/common/cluster" @@ -150,6 +151,20 @@ func (mr *MockShardContextMockRecorder) ChasmRegistry() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChasmRegistry", reflect.TypeOf((*MockShardContext)(nil).ChasmRegistry)) } +// ChasmWorkflowRegistry mocks base method. +func (m *MockShardContext) ChasmWorkflowRegistry() *workflow.Registry { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ChasmWorkflowRegistry") + ret0, _ := ret[0].(*workflow.Registry) + return ret0 +} + +// ChasmWorkflowRegistry indicates an expected call of ChasmWorkflowRegistry. +func (mr *MockShardContextMockRecorder) ChasmWorkflowRegistry() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChasmWorkflowRegistry", reflect.TypeOf((*MockShardContext)(nil).ChasmWorkflowRegistry)) +} + // ConflictResolveWorkflowExecution mocks base method. func (m *MockShardContext) ConflictResolveWorkflowExecution(ctx context.Context, request *persistence0.ConflictResolveWorkflowExecutionRequest) (*persistence0.ConflictResolveWorkflowExecutionResponse, error) { m.ctrl.T.Helper() @@ -969,6 +984,20 @@ func (mr *MockControllableContextMockRecorder) ChasmRegistry() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChasmRegistry", reflect.TypeOf((*MockControllableContext)(nil).ChasmRegistry)) } +// ChasmWorkflowRegistry mocks base method. +func (m *MockControllableContext) ChasmWorkflowRegistry() *workflow.Registry { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ChasmWorkflowRegistry") + ret0, _ := ret[0].(*workflow.Registry) + return ret0 +} + +// ChasmWorkflowRegistry indicates an expected call of ChasmWorkflowRegistry. +func (mr *MockControllableContextMockRecorder) ChasmWorkflowRegistry() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ChasmWorkflowRegistry", reflect.TypeOf((*MockControllableContext)(nil).ChasmWorkflowRegistry)) +} + // ConflictResolveWorkflowExecution mocks base method. func (m *MockControllableContext) ConflictResolveWorkflowExecution(ctx context.Context, request *persistence0.ConflictResolveWorkflowExecutionRequest) (*persistence0.ConflictResolveWorkflowExecutionResponse, error) { m.ctrl.T.Helper() diff --git a/service/history/ndc/events_reapplier.go b/service/history/ndc/events_reapplier.go index e01ad8d56fb..1a3dc99ba84 100644 --- a/service/history/ndc/events_reapplier.go +++ b/service/history/ndc/events_reapplier.go @@ -8,6 +8,7 @@ import ( historypb "go.temporal.io/api/history/v1" "go.temporal.io/api/serviceerror" enumsspb "go.temporal.io/server/api/enums/v1" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" "go.temporal.io/server/service/history/hsm" @@ -27,9 +28,10 @@ type ( } EventsReapplierImpl struct { - stateMachineRegistry *hsm.Registry - metricsHandler metrics.Handler - logger log.Logger + stateMachineRegistry *hsm.Registry + chasmWorkflowRegistry *chasmworkflow.Registry + metricsHandler metrics.Handler + logger log.Logger } ) @@ -46,6 +48,14 @@ func NewEventsReapplier( } } +// WithChasmWorkflowRegistry sets the CHASM workflow registry used to resolve CHASM-backed events on the +// reapply path. Kept separate from NewEventsReapplier so the constructor signature stays backward compatible +// for external callers. +func (r *EventsReapplierImpl) WithChasmWorkflowRegistry(chasmWorkflowRegistry *chasmworkflow.Registry) *EventsReapplierImpl { + r.chasmWorkflowRegistry = chasmWorkflowRegistry + return r +} + func (r *EventsReapplierImpl) ReapplyEvents( ctx context.Context, ms historyi.MutableState, @@ -57,7 +67,7 @@ func (r *EventsReapplierImpl) ReapplyEvents( if !ms.IsWorkflowExecutionRunning() { return nil, serviceerror.NewInternal("unable to reapply events to closed workflow.") } - reappliedEvents, err := reapplyEvents(ctx, ms, updateRegistry, r.stateMachineRegistry, historyEvents, nil, runID, false) + reappliedEvents, err := reapplyEvents(ctx, ms, updateRegistry, r.stateMachineRegistry, r.chasmWorkflowRegistry, historyEvents, nil, runID, false) if err != nil { return nil, err } diff --git a/service/history/ndc/workflow_resetter.go b/service/history/ndc/workflow_resetter.go index 8b3ede020b0..58555eace48 100644 --- a/service/history/ndc/workflow_resetter.go +++ b/service/history/ndc/workflow_resetter.go @@ -16,6 +16,7 @@ import ( "go.temporal.io/server/api/historyservice/v1" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/chasm" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/common" "go.temporal.io/server/common/cluster" "go.temporal.io/server/common/collection" @@ -835,7 +836,17 @@ func (r *workflowResetterImpl) reapplyEvents( // When reapplying events during WorkflowReset, we do not check for conflicting update IDs (they are not possible, // since the workflow was in a consistent state before reset), and we do not perform deduplication (because we never // did, before the refactoring that unified two code paths; see comment below.) - return reapplyEvents(ctx, mutableState, nil, r.shardContext.StateMachineRegistry(), events, resetReapplyExcludeTypes, "", true) + return reapplyEvents( + ctx, + mutableState, + nil, + r.shardContext.StateMachineRegistry(), + r.shardContext.ChasmWorkflowRegistry(), + events, + resetReapplyExcludeTypes, + "", + true, + ) } func reapplyEvents( @@ -843,6 +854,7 @@ func reapplyEvents( mutableState historyi.MutableState, targetBranchUpdateRegistry update.Registry, stateMachineRegistry *hsm.Registry, + chasmWorkflowRegistry *chasmworkflow.Registry, events []*historypb.HistoryEvent, resetReapplyExcludeTypes map[enumspb.ResetReapplyExcludeType]struct{}, runIdForDeduplication string, @@ -1012,17 +1024,27 @@ func reapplyEvents( return nil, err } default: - root := mutableState.HSM() - def, ok := stateMachineRegistry.EventDefinition(event.GetEventType()) - if !ok { - // Only reapply hardcoded events above or ones registered and are cherry-pickable in the HSM framework. - continue + // Nexus operations (and other state-machine-backed components) can be backed by either the + // HSM tree or the CHASM tree, and both coexist on the same mutable state. Reset must rebuild + // the op regardless of which framework owns it. + // TODO(follow-up): the completion-token resolution path (HSM StateMachineRef vs CHASM + // ComponentRef) also needs the same fallback; see the completion handler in nexusoperation. + outcome, err := cherryPickHSMEvent(mutableState, stateMachineRegistry, event, resetReapplyExcludeTypes) + if err != nil { + return reappliedEvents, err } - if err := def.CherryPick(root, event, resetReapplyExcludeTypes); err != nil { - if errors.Is(err, hsm.ErrNotCherryPickable) || errors.Is(err, hsm.ErrStateMachineNotFound) || errors.Is(err, hsm.ErrInvalidTransition) { - continue + if outcome == cherryPickFallback { + // HSM doesn't own this op (unknown type, or component not in the HSM tree): try the + // CHASM workflow tree. + outcome, err = cherryPickChasmEvent(ctx, mutableState, chasmWorkflowRegistry, event, resetReapplyExcludeTypes) + if err != nil { + return reappliedEvents, err } - return reappliedEvents, err + } + if outcome != cherryPickApplied { + // Either skipped (recognized but not cherry-pickable) or unhandled by both frameworks. + // Only reapply hardcoded events above or ones cherry-picked in HSM or CHASM. + continue } mutableState.AddHistoryEvent(event.EventType, func(he *historypb.HistoryEvent) { he.Attributes = event.Attributes @@ -1037,6 +1059,81 @@ func reapplyEvents( return reappliedEvents, nil } +// cherryPickOutcome is the result of attempting to cherry-pick an event against a single framework +// (HSM or CHASM) during reset reapply. +type cherryPickOutcome int + +const ( + // cherryPickApplied: the event was cherry-picked and should be reappended to history. + cherryPickApplied cherryPickOutcome = iota + // cherryPickSkipped: the framework recognizes the event but intentionally won't cherry-pick it + // (e.g. excluded by reset-reapply-exclude-types, or not a cherry-pickable transition). The event + // must be skipped, NOT routed to the other framework, to avoid double-applying. + cherryPickSkipped + // cherryPickFallback: this framework doesn't own the op (unknown event type, or the component is + // not in this tree). The caller should try the other framework. + cherryPickFallback +) + +// cherryPickHSMEvent attempts to cherry-pick an event against the HSM tree. +func cherryPickHSMEvent( + mutableState historyi.MutableState, + stateMachineRegistry *hsm.Registry, + event *historypb.HistoryEvent, + resetReapplyExcludeTypes map[enumspb.ResetReapplyExcludeType]struct{}, +) (cherryPickOutcome, error) { + def, ok := stateMachineRegistry.EventDefinition(event.GetEventType()) + if !ok { + // Event type isn't an HSM event at all; let the caller try CHASM. + return cherryPickFallback, nil + } + if err := def.CherryPick(mutableState.HSM(), event, resetReapplyExcludeTypes); err != nil { + switch { + case errors.Is(err, hsm.ErrStateMachineNotFound): + // The op isn't in the HSM tree. It may live in the CHASM tree instead, so fall back. + return cherryPickFallback, nil + case errors.Is(err, hsm.ErrNotCherryPickable), errors.Is(err, hsm.ErrInvalidTransition): + // Recognized by HSM but intentionally not cherry-pickable here; skip without falling back. + return cherryPickSkipped, nil + default: + return cherryPickSkipped, err + } + } + return cherryPickApplied, nil +} + +// cherryPickChasmEvent attempts to cherry-pick an event against the CHASM workflow tree. It mirrors +// cherryPickHSMEvent. When CHASM is disabled, the registry is unavailable, or the event type is +// unknown to CHASM, it returns cherryPickFallback (the caller has already exhausted HSM, so this +// degrades to "unhandled" and the event is skipped). +func cherryPickChasmEvent( + ctx context.Context, + mutableState historyi.MutableState, + chasmWorkflowRegistry *chasmworkflow.Registry, + event *historypb.HistoryEvent, + resetReapplyExcludeTypes map[enumspb.ResetReapplyExcludeType]struct{}, +) (cherryPickOutcome, error) { + // Only consult the CHASM tree when CHASM is enabled and the registry has been wired through. + if chasmWorkflowRegistry == nil || !mutableState.ChasmEnabled() { + return cherryPickFallback, nil + } + def, ok := chasmWorkflowRegistry.EventDefinitionByEventType(event.GetEventType()) + if !ok { + return cherryPickFallback, nil + } + wf, chasmCtx, err := mutableState.ChasmWorkflowComponent(ctx) + if err != nil { + return cherryPickSkipped, err + } + if err := def.CherryPick(chasmCtx, wf, event, resetReapplyExcludeTypes); err != nil { + if errors.Is(err, chasmworkflow.ErrEventNotCherryPickable) { + return cherryPickSkipped, nil + } + return cherryPickSkipped, err + } + return cherryPickApplied, nil +} + // reapplyChildEvents reapplies all child events except EVENT_TYPE_START_CHILD_WORKFLOW_EXECUTION_INITIATED. // This function is intended to pick up all the events for a child that was already initialized before the reset point. // Re-applying these events is needed to support reconnecting of the child with parent. diff --git a/service/history/ndc/workflow_resetter_test.go b/service/history/ndc/workflow_resetter_test.go index 11c0296e1dd..359798d9481 100644 --- a/service/history/ndc/workflow_resetter_test.go +++ b/service/history/ndc/workflow_resetter_test.go @@ -2,6 +2,7 @@ package ndc import ( "context" + "errors" "slices" "testing" "time" @@ -22,6 +23,7 @@ import ( "go.temporal.io/server/api/historyservice/v1" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/chasm" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/common" "go.temporal.io/server/common/collection" "go.temporal.io/server/common/definition" @@ -959,7 +961,7 @@ func (s *workflowResetterSuite) TestReapplyEvents_WithPendingChildren() { for _, tc := range testcases { s.Run(tc.name+" "+tcReset.name, func() { - _, err := reapplyEvents(context.Background(), mutableState, nil, nil, tc.events, nil, "", tcReset.isReset) + _, err := reapplyEvents(context.Background(), mutableState, nil, nil, nil, tc.events, nil, "", tcReset.isReset) s.NoError(err) }) } @@ -1026,7 +1028,7 @@ func (s *workflowResetterSuite) TestReapplyEvents_WithNoPendingChildren() { for _, tc := range testCases { s.Run(tc.name+" "+tcReset.name, func() { - _, err := reapplyEvents(context.Background(), mutableState, nil, nil, tc.events, nil, "", tcReset.isReset) + _, err := reapplyEvents(context.Background(), mutableState, nil, nil, nil, tc.events, nil, "", tcReset.isReset) s.NoError(err) }) } @@ -1261,7 +1263,7 @@ func (s *workflowResetterSuite) TestReapplyEvents() { } } - appliedEvents, err := reapplyEvents(context.Background(), ms, nil, smReg, events, nil, "", tc.isReset) + appliedEvents, err := reapplyEvents(context.Background(), ms, nil, smReg, nil, events, nil, "", tc.isReset) s.NoError(err) s.Equal(tc.expected, appliedEvents) @@ -1332,7 +1334,7 @@ func (s *workflowResetterSuite) TestReapplyEvents_Excludes() { enumspb.RESET_REAPPLY_EXCLUDE_TYPE_UPDATE: {}, enumspb.RESET_REAPPLY_EXCLUDE_TYPE_NEXUS: {}, } - reappliedEvents, err := reapplyEvents(context.Background(), ms, nil, smReg, events, excludes, "", false) + reappliedEvents, err := reapplyEvents(context.Background(), ms, nil, smReg, nil, events, excludes, "", false) s.Empty(reappliedEvents) s.NoError(err) @@ -1358,7 +1360,7 @@ func (s *workflowResetterSuite) TestReapplyEvents_Excludes() { }, } events = append(events, event7, event8) - reappliedEvents, err = reapplyEvents(context.Background(), ms, nil, smReg, events, excludes, "", true) + reappliedEvents, err = reapplyEvents(context.Background(), ms, nil, smReg, nil, events, excludes, "", true) s.Empty(reappliedEvents) s.NoError(err) } @@ -1653,7 +1655,7 @@ func (s *workflowResetterSuite) TestReapplyEvents_WorkflowOptionsUpdated_Complet events := []*historypb.HistoryEvent{event} // Call reapplyEvents and expect an error - appliedEvents, err := reapplyEvents(context.Background(), ms, nil, smReg, events, nil, "", true) + appliedEvents, err := reapplyEvents(context.Background(), ms, nil, smReg, nil, events, nil, "", true) s.Error(err) s.Contains(err.Error(), tc.expectedErrorContains) s.Empty(appliedEvents) @@ -1701,7 +1703,7 @@ func (s *workflowResetterSuite) TestReapplyEvents_WorkflowOptionsUpdated_Complet events := []*historypb.HistoryEvent{event} // Call reapplyEvents - should skip the event (no error, no applied events) - appliedEvents, err := reapplyEvents(context.Background(), ms, nil, smReg, events, nil, "", true) + appliedEvents, err := reapplyEvents(context.Background(), ms, nil, smReg, nil, events, nil, "", true) s.NoError(err) s.Empty(appliedEvents) // Event should be skipped } @@ -1739,7 +1741,250 @@ func (s *workflowResetterSuite) TestReapplyEvents_WorkflowOptionsUpdated_WithTim attr.GetWorkflowUpdateOptions(), ).Return(&historypb.HistoryEvent{}, nil) - appliedEvents, err := reapplyEvents(context.Background(), ms, nil, smReg, []*historypb.HistoryEvent{event}, nil, "", true) + appliedEvents, err := reapplyEvents(context.Background(), ms, nil, smReg, nil, []*historypb.HistoryEvent{event}, nil, "", true) s.NoError(err) s.Len(appliedEvents, 1) } + +// fakeChasmEventDefinition is a chasmworkflow.EventDefinition whose CherryPick returns a configurable +// error, letting tests drive each cherryPickChasmEvent branch. +type fakeChasmEventDefinition struct { + eventType enumspb.EventType + cherryPickErr error +} + +func (d *fakeChasmEventDefinition) Type() enumspb.EventType { return d.eventType } +func (d *fakeChasmEventDefinition) IsWorkflowTaskTrigger() bool { return false } +func (d *fakeChasmEventDefinition) Apply(chasm.MutableContext, *chasmworkflow.Workflow, *historypb.HistoryEvent) error { + return nil +} +func (d *fakeChasmEventDefinition) CherryPick(chasm.MutableContext, *chasmworkflow.Workflow, *historypb.HistoryEvent, map[enumspb.ResetReapplyExcludeType]struct{}) error { + return d.cherryPickErr +} + +// fakeChasmLibrary registers a set of fake event definitions into a chasmworkflow.Registry. +type fakeChasmLibrary struct { + defs []chasmworkflow.EventDefinition +} + +func (l fakeChasmLibrary) CommandHandlers() map[enumspb.CommandType]chasmworkflow.CommandHandler { + return nil +} + +func (l fakeChasmLibrary) EventDefinitions() []chasmworkflow.EventDefinition { return l.defs } + +func newChasmRegistryWithEvent(eventType enumspb.EventType, cherryPickErr error) *chasmworkflow.Registry { + reg := chasmworkflow.NewRegistry() + _ = reg.Register(fakeChasmLibrary{defs: []chasmworkflow.EventDefinition{ + &fakeChasmEventDefinition{eventType: eventType, cherryPickErr: cherryPickErr}, + }}) + return reg +} + +func (s *workflowResetterSuite) TestCherryPickChasmEvent() { + const eventType = enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED + event := &historypb.HistoryEvent{EventType: eventType} + cherryPickErr := errors.New("cherry-pick failed") + + testCases := []struct { + name string + registry *chasmworkflow.Registry + setupMock func(ms *historyi.MockMutableState) + wantOutcome cherryPickOutcome + wantErr error + }{ + { + name: "nil registry falls back", + registry: nil, + setupMock: func(*historyi.MockMutableState) {}, + wantOutcome: cherryPickFallback, + }, + { + name: "chasm disabled falls back", + registry: newChasmRegistryWithEvent(eventType, nil), + setupMock: func(ms *historyi.MockMutableState) { ms.EXPECT().ChasmEnabled().Return(false) }, + wantOutcome: cherryPickFallback, + }, + { + name: "event type unknown to chasm falls back", + registry: chasmworkflow.NewRegistry(), + setupMock: func(ms *historyi.MockMutableState) { ms.EXPECT().ChasmEnabled().Return(true) }, + wantOutcome: cherryPickFallback, + }, + { + name: "component lookup error is skipped", + registry: newChasmRegistryWithEvent(eventType, nil), + setupMock: func(ms *historyi.MockMutableState) { + ms.EXPECT().ChasmEnabled().Return(true) + ms.EXPECT().ChasmWorkflowComponent(gomock.Any()).Return(nil, nil, cherryPickErr) + }, + wantOutcome: cherryPickSkipped, + wantErr: cherryPickErr, + }, + { + name: "not-cherry-pickable is skipped without error", + registry: newChasmRegistryWithEvent(eventType, chasmworkflow.ErrEventNotCherryPickable), + setupMock: func(ms *historyi.MockMutableState) { + ms.EXPECT().ChasmEnabled().Return(true) + ms.EXPECT().ChasmWorkflowComponent(gomock.Any()).Return(nil, nil, nil) + }, + wantOutcome: cherryPickSkipped, + }, + { + name: "cherry-pick error is skipped and surfaced", + registry: newChasmRegistryWithEvent(eventType, cherryPickErr), + setupMock: func(ms *historyi.MockMutableState) { + ms.EXPECT().ChasmEnabled().Return(true) + ms.EXPECT().ChasmWorkflowComponent(gomock.Any()).Return(nil, nil, nil) + }, + wantOutcome: cherryPickSkipped, + wantErr: cherryPickErr, + }, + { + name: "owned by chasm is applied", + registry: newChasmRegistryWithEvent(eventType, nil), + setupMock: func(ms *historyi.MockMutableState) { + ms.EXPECT().ChasmEnabled().Return(true) + ms.EXPECT().ChasmWorkflowComponent(gomock.Any()).Return(nil, nil, nil) + }, + wantOutcome: cherryPickApplied, + }, + } + + for _, tc := range testCases { + s.Run(tc.name, func() { + ms := historyi.NewMockMutableState(s.controller) + tc.setupMock(ms) + + outcome, err := cherryPickChasmEvent(context.Background(), ms, tc.registry, event, nil) + + s.Equal(tc.wantOutcome, outcome) + if tc.wantErr != nil { + s.ErrorIs(err, tc.wantErr) + } else { + s.NoError(err) + } + }) + } +} + +func (s *workflowResetterSuite) TestReapplyEventsHSMToChasmFallback() { + const eventType = enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED + event := &historypb.HistoryEvent{EventId: 5, EventType: eventType} + smReg := hsm.NewRegistry() + + s.Run("falls back to chasm and reapplies when chasm owns the op", func() { + ms := historyi.NewMockMutableState(s.controller) + ms.EXPECT().ChasmEnabled().Return(true) + ms.EXPECT().ChasmWorkflowComponent(gomock.Any()).Return(nil, nil, nil) + ms.EXPECT().AddHistoryEvent(eventType, gomock.Any()).Return(&historypb.HistoryEvent{}) + + applied, err := reapplyEvents( + context.Background(), ms, nil, smReg, newChasmRegistryWithEvent(eventType, nil), + []*historypb.HistoryEvent{event}, nil, "", true, + ) + s.NoError(err) + s.Equal([]*historypb.HistoryEvent{event}, applied) + }) + + s.Run("skips the event when neither tree owns the op", func() { + ms := historyi.NewMockMutableState(s.controller) + ms.EXPECT().ChasmEnabled().Return(true) + + applied, err := reapplyEvents( + context.Background(), ms, nil, smReg, chasmworkflow.NewRegistry(), + []*historypb.HistoryEvent{event}, nil, "", true, + ) + s.NoError(err) + s.Empty(applied) + }) +} + +// fakeHSMEventDefinition is an hsm.EventDefinition whose CherryPick returns a configurable error, +// letting tests drive each cherryPickHSMEvent branch. +type fakeHSMEventDefinition struct { + eventType enumspb.EventType + cherryPickErr error +} + +func (d *fakeHSMEventDefinition) Type() enumspb.EventType { return d.eventType } +func (d *fakeHSMEventDefinition) IsWorkflowTaskTrigger() bool { return false } +func (d *fakeHSMEventDefinition) Apply(*hsm.Node, *historypb.HistoryEvent) error { return nil } +func (d *fakeHSMEventDefinition) CherryPick(*hsm.Node, *historypb.HistoryEvent, map[enumspb.ResetReapplyExcludeType]struct{}) error { + return d.cherryPickErr +} + +func newHSMRegistryWithEvent(eventType enumspb.EventType, cherryPickErr error) *hsm.Registry { + reg := hsm.NewRegistry() + _ = reg.RegisterEventDefinition(&fakeHSMEventDefinition{eventType: eventType, cherryPickErr: cherryPickErr}) + return reg +} + +func (s *workflowResetterSuite) TestCherryPickHSMEvent() { + const eventType = enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED + event := &historypb.HistoryEvent{EventType: eventType} + cherryPickErr := errors.New("cherry-pick failed") + + testCases := []struct { + name string + registry *hsm.Registry + expectHSM bool + wantOutcome cherryPickOutcome + wantErr error + }{ + { + name: "event type unknown to hsm falls back", + registry: hsm.NewRegistry(), + wantOutcome: cherryPickFallback, + }, + { + name: "state machine not found falls back", + registry: newHSMRegistryWithEvent(eventType, hsm.ErrStateMachineNotFound), + expectHSM: true, + wantOutcome: cherryPickFallback, + }, + { + name: "not-cherry-pickable is skipped without error", + registry: newHSMRegistryWithEvent(eventType, hsm.ErrNotCherryPickable), + expectHSM: true, + wantOutcome: cherryPickSkipped, + }, + { + name: "invalid transition is skipped without error", + registry: newHSMRegistryWithEvent(eventType, hsm.ErrInvalidTransition), + expectHSM: true, + wantOutcome: cherryPickSkipped, + }, + { + name: "cherry-pick error is skipped and surfaced", + registry: newHSMRegistryWithEvent(eventType, cherryPickErr), + expectHSM: true, + wantOutcome: cherryPickSkipped, + wantErr: cherryPickErr, + }, + { + name: "owned by hsm is applied", + registry: newHSMRegistryWithEvent(eventType, nil), + expectHSM: true, + wantOutcome: cherryPickApplied, + }, + } + + for _, tc := range testCases { + s.Run(tc.name, func() { + ms := historyi.NewMockMutableState(s.controller) + if tc.expectHSM { + ms.EXPECT().HSM().Return(nil) + } + + outcome, err := cherryPickHSMEvent(ms, tc.registry, event, nil) + + s.Equal(tc.wantOutcome, outcome) + if tc.wantErr != nil { + s.ErrorIs(err, tc.wantErr) + } else { + s.NoError(err) + } + }) + } +} diff --git a/service/history/shard/context_factory.go b/service/history/shard/context_factory.go index 9f32a385816..f7ef9343cf2 100644 --- a/service/history/shard/context_factory.go +++ b/service/history/shard/context_factory.go @@ -2,6 +2,7 @@ package shard import ( "go.temporal.io/server/chasm" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/client" "go.temporal.io/server/common/archiver" "go.temporal.io/server/common/clock" @@ -60,6 +61,7 @@ type ( StateMachineRegistry *hsm.Registry ChasmRegistry *chasm.Registry + ChasmWorkflowRegistry *chasmworkflow.Registry EndpointRegistry commonnexus.EndpointRegistry HandoverTrackerFactory HandoverTrackerFactory } @@ -104,6 +106,7 @@ func (c *contextFactoryImpl) CreateContext( c.EventsCache, c.StateMachineRegistry, c.ChasmRegistry, + c.ChasmWorkflowRegistry, c.EndpointRegistry, c.HandoverTrackerFactory, ) diff --git a/service/history/shard/context_impl.go b/service/history/shard/context_impl.go index 8094c55f228..29b0cc9814b 100644 --- a/service/history/shard/context_impl.go +++ b/service/history/shard/context_impl.go @@ -19,6 +19,7 @@ import ( "go.temporal.io/server/api/historyservice/v1" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/chasm" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/client" "go.temporal.io/server/common" "go.temporal.io/server/common/archiver" @@ -147,8 +148,9 @@ type ( stateMachineRegistry *hsm.Registry - chasmRegistry *chasm.Registry - endpointRegistry chasm.EndpointRegistry + chasmRegistry *chasm.Registry + chasmWorkflowRegistry *chasmworkflow.Registry + endpointRegistry chasm.EndpointRegistry businessIDRateLimiters cache.Cache } @@ -2062,6 +2064,7 @@ func newContext( eventsCache events.Cache, stateMachineRegistry *hsm.Registry, chasmRegistry *chasm.Registry, + chasmWorkflowRegistry *chasmworkflow.Registry, endpointRegistry chasm.EndpointRegistry, handoverTrackerFactory HandoverTrackerFactory, ) (*ContextImpl, error) { @@ -2112,6 +2115,7 @@ func newContext( ioSemaphore: locks.NewPrioritySemaphore(ioConcurrency), stateMachineRegistry: stateMachineRegistry, chasmRegistry: chasmRegistry, + chasmWorkflowRegistry: chasmWorkflowRegistry, endpointRegistry: endpointRegistry, businessIDRateLimiters: cache.New( historyConfig.BusinessIDReuseLimiterCacheSize(), @@ -2231,6 +2235,10 @@ func (s *ContextImpl) ChasmRegistry() *chasm.Registry { return s.chasmRegistry } +func (s *ContextImpl) ChasmWorkflowRegistry() *chasmworkflow.Registry { + return s.chasmWorkflowRegistry +} + func (s *ContextImpl) EndpointRegistry() chasm.EndpointRegistry { return s.endpointRegistry } diff --git a/service/history/shard/context_testutil.go b/service/history/shard/context_testutil.go index 9b03a310a69..4bdc9081e90 100644 --- a/service/history/shard/context_testutil.go +++ b/service/history/shard/context_testutil.go @@ -8,6 +8,7 @@ import ( "go.temporal.io/server/api/historyservice/v1" persistencespb "go.temporal.io/server/api/persistence/v1" "go.temporal.io/server/chasm" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/common/cache" "go.temporal.io/server/common/clock" "go.temporal.io/server/common/cluster" @@ -222,6 +223,10 @@ func (s *ContextTest) SetChasmRegistry(reg *chasm.Registry) { s.chasmRegistry = reg } +func (s *ContextTest) SetChasmWorkflowRegistry(reg *chasmworkflow.Registry) { + s.chasmWorkflowRegistry = reg +} + func (s *ContextTest) SetClusterMetadata(metadata cluster.Metadata) { s.clusterMetadata = metadata } diff --git a/service/history/statemachine_environment.go b/service/history/statemachine_environment.go index 0b516a6aeec..d2e7ca719b8 100644 --- a/service/history/statemachine_environment.go +++ b/service/history/statemachine_environment.go @@ -309,8 +309,14 @@ func (e *stateMachineEnvironment) validateStateMachineRefWithoutTransitionHistor return fmt.Errorf("%w: %w", serviceerror.NewInternal("node lookup failed"), err) } - if node.InternalRepr().InitialVersionedTransition.NamespaceFailoverVersion != - ref.StateMachineRef.MachineInitialVersionedTransition.NamespaceFailoverVersion { + // A ref reconstructed from a cross-tree Nexus completion (see completeNexusOperationHSMFallback) has no + // MachineInitialVersionedTransition: it is built from a CHASM completion token that carries no HSM + // versioned-transition info. There is nothing to compare for staleness in that case, and the op node + // was already located by ScheduledEventId above, so skip the failover-version check rather than + // dereferencing a nil transition. + if ref.StateMachineRef.MachineInitialVersionedTransition != nil && + node.InternalRepr().InitialVersionedTransition.NamespaceFailoverVersion != + ref.StateMachineRef.MachineInitialVersionedTransition.NamespaceFailoverVersion { if potentialStaleState { return fmt.Errorf("%w: state machine ref initial failover version mismatch", consts.ErrStaleState) } diff --git a/service/history/workflow/mutable_state_rebuilder.go b/service/history/workflow/mutable_state_rebuilder.go index af6716a7b14..4099d447d38 100644 --- a/service/history/workflow/mutable_state_rebuilder.go +++ b/service/history/workflow/mutable_state_rebuilder.go @@ -6,6 +6,7 @@ package workflow import ( "context" + "errors" "github.com/google/uuid" commonpb "go.temporal.io/api/common/v1" @@ -15,10 +16,12 @@ import ( enumsspb "go.temporal.io/server/api/enums/v1" "go.temporal.io/server/common/cluster" "go.temporal.io/server/common/log" + "go.temporal.io/server/common/log/tag" "go.temporal.io/server/common/namespace" "go.temporal.io/server/common/persistence/versionhistory" "go.temporal.io/server/common/primitives/timestamp" "go.temporal.io/server/service/history/historybuilder" + "go.temporal.io/server/service/history/hsm" historyi "go.temporal.io/server/service/history/interfaces" ) @@ -682,11 +685,7 @@ func (b *MutableStateRebuilderImpl) applyEvents( } default: - def, ok := b.shard.StateMachineRegistry().EventDefinition(event.GetEventType()) - if !ok { - return nil, serviceerror.NewInvalidArgumentf("Unknown event type: %v", event.GetEventType()) - } - if err := def.Apply(b.mutableState.HSM(), event); err != nil { + if err := b.applyStateMachineEvent(ctx, event); err != nil { return nil, err } } @@ -712,6 +711,114 @@ func (b *MutableStateRebuilderImpl) applyEvents( ) } +// applyStateMachineEvent applies a state-machine-backed history event (e.g. Nexus operation events) +// during state rebuild (replication / reset). +// - The CREATE event (NexusOperationScheduled) is routed by the per-namespace feature flag +// nexusoperation.enableChasmWorkflowOperations: flag ON -> create in the CHASM tree; flag OFF -> +// HSM tree (legacy behavior). +// - All OTHER (non-create) events try HSM first; if the event type is unknown to HSM, or the HSM +// component for the op can't be found (ErrStateMachineNotFound), we fall back to the CHASM +// workflow tree. +func (b *MutableStateRebuilderImpl) applyStateMachineEvent( + ctx context.Context, + event *historypb.HistoryEvent, +) error { + if event.GetEventType() == enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED { + return b.applyNexusOperationScheduledEvent(ctx, event) + } + + // Non-create events: HSM-first, CHASM-fallback. + err := b.applyHSMEvent(event) + if err == nil { + return nil + } + // Only fall back to CHASM when HSM doesn't own this op (component not found). Any other HSM error + // (including a genuinely unknown event type) is returned as-is. + if !errors.Is(err, hsm.ErrStateMachineNotFound) { + return err + } + applied, chasmErr := b.applyChasmEvent(ctx, event) + if chasmErr != nil { + return chasmErr + } + if !applied { + // Neither tree could apply it; surface the original HSM error to preserve existing semantics. + return err + } + return nil +} + +// applyNexusOperationScheduledEvent (re)creates a Nexus operation during state rebuild, choosing the +// backing tree by the per-namespace nexusoperation.enableChasmWorkflowOperations flag. Unlike non-create events, +// the create is not "probed" against an existing component (none exists yet) — instead it is routed by current +// creation policy. This means a reset realigns an op to whichever framework new ops would be created in today. +func (b *MutableStateRebuilderImpl) applyNexusOperationScheduledEvent( + ctx context.Context, + event *historypb.HistoryEvent, +) error { + if !b.mutableState.ChasmEnabled() || b.shard.ChasmWorkflowRegistry() == nil { + return b.applyHSMEvent(event) + } + nsName := b.mutableState.GetNamespaceEntry().Name().String() + if !b.shard.GetConfig().EnableChasmNexusWorkflowOperations(nsName) { + // Flag off: legacy path, create the op in the HSM tree. + return b.applyHSMEvent(event) + } + + applied, err := b.applyChasmEvent(ctx, event) + if err != nil { + // CHASM create failed; fall back to HSM. + b.logger.Warn( + "Nexus operation CHASM create failed during state rebuild; falling back to HSM tree", + tag.WorkflowNamespace(nsName), + tag.WorkflowScheduledEventID(event.GetEventId()), + tag.Error(err), + ) + return b.applyHSMEvent(event) + } + if !applied { + // CHASM didn't claim the event (registry/component unavailable); fall back to HSM. + return b.applyHSMEvent(event) + } + return nil +} + +// applyHSMEvent applies an event to the HSM tree +func (b *MutableStateRebuilderImpl) applyHSMEvent(event *historypb.HistoryEvent) error { + def, ok := b.shard.StateMachineRegistry().EventDefinition(event.GetEventType()) + if !ok { + return serviceerror.NewInvalidArgumentf("Unknown event type: %v", event.GetEventType()) + } + return def.Apply(b.mutableState.HSM(), event) +} + +// applyChasmEvent applies an event to the CHASM workflow tree. Returns (true, nil) if applied, +// (false, nil) if CHASM is disabled / the registry is unavailable / the event type is unknown to +// CHASM, and (false, err) for a fatal error. +func (b *MutableStateRebuilderImpl) applyChasmEvent( + ctx context.Context, + event *historypb.HistoryEvent, +) (bool, error) { + chasmWorkflowRegistry := b.shard.ChasmWorkflowRegistry() + if chasmWorkflowRegistry == nil || !b.mutableState.ChasmEnabled() { + return false, nil + } + def, ok := chasmWorkflowRegistry.EventDefinitionByEventType(event.GetEventType()) + if !ok { + return false, nil + } + // Ensure the root CHASM workflow component exists before applying. + b.mutableState.EnsureChasmWorkflowComponent(ctx) + wf, chasmCtx, err := b.mutableState.ChasmWorkflowComponent(ctx) + if err != nil { + return false, err + } + if err := def.Apply(chasmCtx, wf, event); err != nil { + return false, err + } + return true, nil +} + func (b *MutableStateRebuilderImpl) applyNewRunHistory( ctx context.Context, namespaceID namespace.ID, diff --git a/service/history/workflow/mutable_state_rebuilder_test.go b/service/history/workflow/mutable_state_rebuilder_test.go index 02fd09ffc44..10f7f544ed3 100644 --- a/service/history/workflow/mutable_state_rebuilder_test.go +++ b/service/history/workflow/mutable_state_rebuilder_test.go @@ -12,14 +12,18 @@ import ( commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" historypb "go.temporal.io/api/history/v1" + "go.temporal.io/api/serviceerror" taskqueuepb "go.temporal.io/api/taskqueue/v1" enumsspb "go.temporal.io/server/api/enums/v1" historyspb "go.temporal.io/server/api/history/v1" persistencespb "go.temporal.io/server/api/persistence/v1" + "go.temporal.io/server/chasm" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/common" "go.temporal.io/server/common/backoff" "go.temporal.io/server/common/cluster" "go.temporal.io/server/common/definition" + "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/log" "go.temporal.io/server/common/namespace" "go.temporal.io/server/common/payload" @@ -2184,6 +2188,8 @@ func (s *stateBuilderSuite) TestApplyEvents_HSMRegistry() { } s.mockMutableState.EXPECT().ClearStickyTaskQueue() s.mockUpdateVersion(event) + // CHASM disabled -> the create event is routed to the HSM tree (legacy behavior). + s.mockMutableState.EXPECT().ChasmEnabled().Return(false).AnyTimes() _, err := s.stateRebuilder.ApplyEvents(context.Background(), tests.NamespaceID, requestID, execution, s.toHistory(event), nil, "") s.NoError(err) @@ -2193,6 +2199,55 @@ func (s *stateBuilderSuite) TestApplyEvents_HSMRegistry() { s.Equal(enumsspb.NEXUS_OPERATION_STATE_SCHEDULED, sm.State()) } +// TestApplyEvents_NexusScheduled_ChasmCreateFallsBackToHSM verifies that when the creation-policy +// flag is ON but creating the op in the CHASM tree fails, rebuild falls back to creating the op in +// the HSM tree rather than failing the whole rebuild (rollout resilience). +func (s *stateBuilderSuite) TestApplyEvents_NexusScheduled_ChasmCreateFallsBackToHSM() { + version := int64(1) + requestID := uuid.NewString() + execution := &commonpb.WorkflowExecution{ + WorkflowId: "wf-id", + RunId: tests.RunID, + } + event := &historypb.HistoryEvent{ + TaskId: rand.Int63(), + Version: version, + EventId: 5, + EventTime: timestamppb.New(time.Now().UTC()), + EventType: enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED, + Attributes: &historypb.HistoryEvent_NexusOperationScheduledEventAttributes{ + NexusOperationScheduledEventAttributes: &historypb.NexusOperationScheduledEventAttributes{ + EndpointId: "endpoint-id", + Endpoint: "endpoint", + Service: "service", + Operation: "operation", + WorkflowTaskCompletedEventId: 4, + RequestId: "request-id", + }, + }, + } + + // Flag ON + a non-nil CHASM workflow registry so the create attempts the CHASM tree first. + s.mockShard.GetConfig().EnableChasmNexusWorkflowOperations = dynamicconfig.GetBoolPropertyFnFilteredByNamespace(true) + s.mockShard.SetChasmWorkflowRegistry(chasmworkflow.NewRegistry()) + + s.mockMutableState.EXPECT().ClearStickyTaskQueue() + s.mockUpdateVersion(event) + s.mockMutableState.EXPECT().ChasmEnabled().Return(true).AnyTimes() + s.mockMutableState.EXPECT().GetNamespaceEntry().Return(tests.GlobalNamespaceEntry).AnyTimes() + s.mockMutableState.EXPECT().EnsureChasmWorkflowComponent(gomock.Any()).AnyTimes() + // Force the CHASM create to fail so the HSM fallback path is taken. + s.mockMutableState.EXPECT().ChasmWorkflowComponent(gomock.Any()). + Return(nil, nil, serviceerror.NewInternal("chasm unavailable")).AnyTimes() + + _, err := s.stateRebuilder.ApplyEvents(context.Background(), tests.NamespaceID, requestID, execution, s.toHistory(event), nil, "") + s.NoError(err) + // The op must have been created in the HSM tree by the fallback. + sm, err := nexusoperations.MachineCollection(s.mockMutableState.HSM()).Data("5") + s.NoError(err) + s.Equal(enumsspb.NEXUS_OPERATION_STATE_SCHEDULED, sm.State()) +} + func (p *testTaskGeneratorProvider) NewTaskGenerator( shardContext historyi.ShardContext, mutableState historyi.MutableState, @@ -2209,3 +2264,215 @@ func (p *testTaskGeneratorProvider) NewTaskGenerator( shardContext.GetLogger(), ) } + +// --- applyStateMachineEvent (HSM<->CHASM dispatch) coverage ------------------------------------- +// +// These exercise the probe-and-fallback dispatch added for rebuilding Nexus operations that may live +// in either the HSM or CHASM tree. + +type fakeChasmEventDefinition struct { + eventType enumspb.EventType + applyErr error + applied *bool +} + +func (d *fakeChasmEventDefinition) Type() enumspb.EventType { return d.eventType } +func (d *fakeChasmEventDefinition) IsWorkflowTaskTrigger() bool { return false } +func (d *fakeChasmEventDefinition) Apply(chasm.MutableContext, *chasmworkflow.Workflow, *historypb.HistoryEvent) error { + if d.applied != nil { + *d.applied = true + } + return d.applyErr +} + +func (d *fakeChasmEventDefinition) CherryPick(chasm.MutableContext, *chasmworkflow.Workflow, *historypb.HistoryEvent, map[enumspb.ResetReapplyExcludeType]struct{}) error { + return nil +} + +type fakeChasmLibrary struct { + defs []chasmworkflow.EventDefinition +} + +func (l fakeChasmLibrary) CommandHandlers() map[enumspb.CommandType]chasmworkflow.CommandHandler { + return nil +} + +func (l fakeChasmLibrary) EventDefinitions() []chasmworkflow.EventDefinition { return l.defs } + +func newFakeChasmRegistry(def *fakeChasmEventDefinition) *chasmworkflow.Registry { + reg := chasmworkflow.NewRegistry() + var defs []chasmworkflow.EventDefinition + if def != nil { + defs = append(defs, def) + } + _ = reg.Register(fakeChasmLibrary{defs: defs}) + return reg +} + +type fakeHSMEventDefinition struct { + eventType enumspb.EventType + applyErr error + applied *bool +} + +func (d *fakeHSMEventDefinition) Type() enumspb.EventType { return d.eventType } +func (d *fakeHSMEventDefinition) IsWorkflowTaskTrigger() bool { return false } +func (d *fakeHSMEventDefinition) Apply(*hsm.Node, *historypb.HistoryEvent) error { + if d.applied != nil { + *d.applied = true + } + return d.applyErr +} + +func (d *fakeHSMEventDefinition) CherryPick(*hsm.Node, *historypb.HistoryEvent, map[enumspb.ResetReapplyExcludeType]struct{}) error { + return nil +} + +func newFakeHSMRegistry(def *fakeHSMEventDefinition) *hsm.Registry { + reg := hsm.NewRegistry() + if def != nil { + _ = reg.RegisterEventDefinition(def) + } + return reg +} + +type nexusRebuildCase struct { + chasmEnabled bool + registryNil bool + flagOn bool + chasmHasDef bool + hsmApplyErr error + chasmApplyErr error + chasmComponentErr error +} + +// runApplyStateMachineEvent wires fake HSM/CHASM registries onto the shard, drives applyStateMachineEvent for the given +// event, and reports which tree's Apply ran plus the returned error. +func (s *stateBuilderSuite) runApplyStateMachineEvent( + event *historypb.HistoryEvent, + tc nexusRebuildCase, +) (chasmApplied bool, hsmApplied bool, err error) { + s.mockShard.SetStateMachineRegistry(newFakeHSMRegistry(&fakeHSMEventDefinition{ + eventType: event.GetEventType(), + applyErr: tc.hsmApplyErr, + applied: &hsmApplied, + })) + if tc.registryNil { + s.mockShard.SetChasmWorkflowRegistry(nil) + } else { + var def *fakeChasmEventDefinition + if tc.chasmHasDef { + def = &fakeChasmEventDefinition{ + eventType: event.GetEventType(), + applyErr: tc.chasmApplyErr, + applied: &chasmApplied, + } + } + s.mockShard.SetChasmWorkflowRegistry(newFakeChasmRegistry(def)) + } + s.mockShard.GetConfig().EnableChasmNexusWorkflowOperations = dynamicconfig.GetBoolPropertyFnFilteredByNamespace(tc.flagOn) + + ms := historyi.NewMockMutableState(s.controller) + ms.EXPECT().HSM().Return(nil).AnyTimes() + ms.EXPECT().ChasmEnabled().Return(tc.chasmEnabled).AnyTimes() + ms.EXPECT().GetNamespaceEntry().Return(tests.GlobalNamespaceEntry).AnyTimes() + ms.EXPECT().EnsureChasmWorkflowComponent(gomock.Any()).AnyTimes() + ms.EXPECT().ChasmWorkflowComponent(gomock.Any()). + Return(&chasmworkflow.Workflow{}, nil, tc.chasmComponentErr).AnyTimes() + + rebuilder := NewMutableStateRebuilder(s.mockShard, s.logger, ms) + err = rebuilder.applyStateMachineEvent(context.Background(), event) + return chasmApplied, hsmApplied, err +} + +func nexusScheduledEvent() *historypb.HistoryEvent { + return &historypb.HistoryEvent{EventId: 5, EventType: enumspb.EVENT_TYPE_NEXUS_OPERATION_SCHEDULED} +} + +func nexusCompletedEvent() *historypb.HistoryEvent { + return &historypb.HistoryEvent{EventId: 6, EventType: enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED} +} + +func (s *stateBuilderSuite) TestApplyStateMachineEvent() { + hsmErr := serviceerror.NewInternal("hsm apply failed") + + testCases := []struct { + name string + event *historypb.HistoryEvent + tc nexusRebuildCase + wantChasmApplied bool + wantHSMApplied bool + wantErr error + }{ + { + name: "nexus scheduled, flag on, creates in chasm", + event: nexusScheduledEvent(), + tc: nexusRebuildCase{chasmEnabled: true, flagOn: true, chasmHasDef: true}, + wantChasmApplied: true, + }, + { + name: "nexus scheduled, flag off, routes to hsm", + event: nexusScheduledEvent(), + tc: nexusRebuildCase{chasmEnabled: true, flagOn: false, chasmHasDef: true}, + wantHSMApplied: true, + }, + { + name: "nexus scheduled, chasm disabled, routes to hsm", + event: nexusScheduledEvent(), + tc: nexusRebuildCase{chasmEnabled: false, flagOn: true, chasmHasDef: true}, + wantHSMApplied: true, + }, + { + name: "nexus scheduled, nil chasm registry, routes to hsm", + event: nexusScheduledEvent(), + tc: nexusRebuildCase{chasmEnabled: true, registryNil: true, flagOn: true}, + wantHSMApplied: true, + }, + { + name: "nexus scheduled, chasm does not claim, falls back to hsm", + event: nexusScheduledEvent(), + tc: nexusRebuildCase{chasmEnabled: true, flagOn: true, chasmHasDef: false}, + wantHSMApplied: true, + }, + { + name: "non-create, hsm owns and applies", + event: nexusCompletedEvent(), + tc: nexusRebuildCase{chasmEnabled: true, chasmHasDef: true}, + wantHSMApplied: true, + }, + { + name: "non-create, hsm not found, falls back to chasm", + event: nexusCompletedEvent(), + tc: nexusRebuildCase{chasmEnabled: true, chasmHasDef: true, hsmApplyErr: hsm.ErrStateMachineNotFound}, + wantChasmApplied: true, + wantHSMApplied: true, + }, + { + name: "non-create, neither tree owns, returns original hsm error", + event: nexusCompletedEvent(), + tc: nexusRebuildCase{chasmEnabled: true, chasmHasDef: false, hsmApplyErr: hsm.ErrStateMachineNotFound}, + wantHSMApplied: true, + wantErr: hsm.ErrStateMachineNotFound, + }, + { + name: "non-create, non-not-found hsm error returned as-is", + event: nexusCompletedEvent(), + tc: nexusRebuildCase{chasmEnabled: true, chasmHasDef: true, hsmApplyErr: hsmErr}, + wantHSMApplied: true, + wantErr: hsmErr, + }, + } + + for _, tc := range testCases { + s.Run(tc.name, func() { + chasmApplied, hsmApplied, err := s.runApplyStateMachineEvent(tc.event, tc.tc) + if tc.wantErr != nil { + s.ErrorIs(err, tc.wantErr) + } else { + s.NoError(err) + } + s.Equal(tc.wantChasmApplied, chasmApplied) + s.Equal(tc.wantHSMApplied, hsmApplied) + }) + } +} diff --git a/tests/nexus_workflow_test.go b/tests/nexus_workflow_test.go index afc330bc548..b74b2a288b2 100644 --- a/tests/nexus_workflow_test.go +++ b/tests/nexus_workflow_test.go @@ -1038,6 +1038,496 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletion(chasmEnabled s.RequireNoHistoryEvent(resetHist2, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED) } +// TestNexusOperationResetCrossTreeCompletion exercises the HSM<->CHASM probe-and-fallback across the +// full lifecycle: an op is scheduled in one framework, the per-namespace creation-policy flag is +// flipped, the workflow is reset (so rebuild re-creates the op in the OTHER framework's tree), and an +// async completion arrives carrying the ORIGINAL framework's token. The completion must still resolve +// the op via the cross-tree fallback. Both directions are covered: +// - nexusOpInitiallyInHSM=true: op scheduled in HSM, flag flipped ON, reset rebuilds op into CHASM, +// completion uses the HSM token -> server-side HSM-token->CHASM-op fallback (History handler). +// - nexusOpInitiallyInHSM=false: op scheduled in CHASM, flag flipped OFF, reset rebuilds op into HSM, +// completion uses the CHASM token -> frontend CHASM-token->HSM-op fallback. +func (s *NexusWorkflowTestSuite) TestNexusOperationResetCrossTreeCompletion(chasmEnabled bool) { + // This test drives the creation-policy flag itself, so run it once (not per-suite-mode). + if chasmEnabled { + s.T().Skip("cross-tree test controls the flag explicitly; run once via the HSM suite") + } + + testCases := []struct { + name string + nexusOpInitiallyInHSM bool + }{ + {name: "HSMOpRebuiltIntoChasm", nexusOpInitiallyInHSM: true}, + {name: "ChasmOpRebuiltIntoHSM", nexusOpInitiallyInHSM: false}, + } + + for _, tc := range testCases { + s.Run(tc.name, func(s *NexusWorkflowTestSuite) { + // CHASM tree + callbacks are enabled in both directions; only the per-op creation policy + // (EnableChasmWorkflowOperations) is flipped during the test so reset rebuilds the op into + // the OTHER tree. + env := s.newTestEnv(true) + ctx := env.Context() + taskQueue := testcore.RandomizeStr(s.T().Name()) + nsName := env.Namespace().String() + // Use the env-level (namespace-scoped) override: on a shared cluster a global + // cluster.OverrideDynamicConfig is less specific than the namespace-scoped baseline that + // newTestEnv installs, so it would not actually flip the flag for this namespace. + setFlag := func(on bool) { + // Mergeable, latest-wins override, so a later setFlag supersedes an earlier one. + env.OverrideDynamicConfig(chasmnexus.EnableChasmWorkflowOperations, on) + } + setFlag(!tc.nexusOpInitiallyInHSM) + + var callbackToken, publicCallbackURL string + var run client.WorkflowRun + + h := nexustest.Handler{ + OnStartOperation: func( + ctx context.Context, + service, operation string, + input *nexus.LazyValue, + options nexus.StartOperationOptions, + ) (nexus.HandlerStartOperationResult[any], error) { + callbackToken = options.CallbackHeader.Get(commonnexus.CallbackTokenHeader) + publicCallbackURL = options.CallbackURL + return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil + }, + } + endpointName := env.createRandomExternalNexusServer(ctx, s.T(), h) + + callerWF := func(ctx workflow.Context) (string, error) { + c := workflow.NewNexusClient(endpointName, "service") + fut := c.ExecuteOperation(ctx, "operation", "input", workflow.NexusOperationOptions{}) + var result string + err := fut.Get(ctx, &result) + return result, err + } + + run, err := env.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{ + TaskQueue: taskQueue, + }, callerWF) + s.NoError(err) + + w := worker.New(env.SdkClient(), taskQueue, worker.Options{}) + w.RegisterWorkflow(callerWF) + s.NoError(w.Start()) + defer w.Stop() + + // Wait for the op to start, then complete it on its ORIGINAL tree and let the workflow finish. + s.EventuallyWithT(func(t *assert.CollectT) { + hist := env.GetHistory(nsName, &commonpb.WorkflowExecution{WorkflowId: run.GetID()}) + historyrequire.New(t).RequireHistoryEvent(hist, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED) + }, time.Second*10, time.Millisecond*200) + s.NotEmpty(callbackToken) + + // While the op is still live (pending), verify the creation-policy flag actually routed it to + // the expected initial tree. This confirms the (namespace-scoped) override took effect; if it + // hadn't, the op would land in the wrong tree here. + desc, err := env.AdminClient().DescribeMutableState(ctx, &adminservice.DescribeMutableStateRequest{ + Namespace: nsName, + Execution: &commonpb.WorkflowExecution{WorkflowId: run.GetID()}, + Archetype: chasm.WorkflowArchetype, + }) + s.NoError(err) + _, opInHSMTree := desc.GetDatabaseMutableState().GetExecutionInfo().GetSubStateMachinesByType()[nexusoperations.OperationMachineType] + opInChasmTree := false + for nodePath := range desc.GetDatabaseMutableState().GetChasmNodes() { + if strings.Contains(nodePath, "Operations") { + opInChasmTree = true + break + } + } + if tc.nexusOpInitiallyInHSM { + // Flag off at creation -> op created in the HSM tree. + s.True(opInHSMTree, "op should have been created in the HSM tree") + s.False(opInChasmTree, "op should not be in the CHASM tree") + } else { + // Flag on at creation -> op created in the CHASM tree. + s.False(opInHSMTree, "op should not be in the HSM tree") + s.True(opInChasmTree, "op should have been created in the CHASM tree") + } + + completion := nexusrpc.CompleteOperationOptions{ + Result: testcore.MustToPayload(s.T(), "result"), + Header: nexus.Header{commonnexus.CallbackTokenHeader: callbackToken}, + } + s.NoError(s.sendNexusCompletionRequest(ctx, publicCallbackURL, completion)) + + var result string + s.NoError(run.Get(ctx, &result)) + s.Equal("result", result) + + wfExec := &commonpb.WorkflowExecution{WorkflowId: run.GetID(), RunId: run.GetRunID()} + hist := env.GetHistory(nsName, wfExec) + opStartedEvent := s.RequireHistoryEvent(hist, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED) + wftCompletedIdx := slices.IndexFunc(hist, func(e *historypb.HistoryEvent) bool { + return e.EventType == enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED && e.EventId > opStartedEvent.EventId + }) + s.Positive(wftCompletedIdx, "expected WorkflowTaskCompleted after NexusOperationStarted") + wftCompletedEventID := hist[wftCompletedIdx].EventId + + // Flip the creation-policy flag, then reset to just after the op started. Rebuild re-creates the + // op under the NEW policy (i.e. into the OTHER tree), and the original NexusOperationCompleted + // event must be reapplied via the reset HSM<->CHASM cross-tree fallback. + setFlag(tc.nexusOpInitiallyInHSM) + + resp, err := env.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{ + Namespace: nsName, + WorkflowExecution: wfExec, + Reason: "test cross-tree reapply", + RequestId: uuid.NewString(), + WorkflowTaskFinishEventId: wftCompletedEventID, + }) + s.NoError(err) + resetRunID := resp.RunId + s.NotEqual(run.GetRunID(), resetRunID, "reset should produce a new run") + + resetHist := env.GetHistory(nsName, &commonpb.WorkflowExecution{WorkflowId: run.GetID(), RunId: resetRunID}) + s.RequireHistoryEvent(resetHist, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED) + }) + } +} + +// TestNexusOperationCrossTreeCompletionAfterReset covers the server-side completion-token cross-tree +// fallback for a LIVE completion that arrives after a reset moved the op to the other tree. Unlike +// TestNexusOperationResetCrossTreeCompletion (which completes before the reset and checks the reapplied +// event), here the completion is sent AFTER the reset using the ORIGINAL token, whose format matches the +// op's original tree: +// - HSM token -> CHASM op: CompleteNexusOperation returns NotFound, then completeNexusOperationChasmFallback +// completes the rebuilt op in the CHASM tree. +// - CHASM token -> HSM op: CompleteNexusOperationChasm fails against the closed schedule-time run, then +// completeNexusOperationAfterReset routes to completeNexusOperationHSMFallback. +// +// We reset to just after NexusOperationStarted so the rebuilt op is still PENDING on the reset run; the +// run only finishes once the cross-tree completion lands, which avoids the reset-run-finishes-first race. +func (s *NexusWorkflowTestSuite) TestNexusOperationCrossTreeCompletionAfterReset(chasmEnabled bool) { + // This test drives the creation-policy flag itself, so run it once (not per-suite-mode). + if chasmEnabled { + s.T().Skip("cross-tree test controls the flag explicitly; run once via the HSM suite") + } + + testCases := []struct { + name string + nexusOpInitiallyInHSM bool + }{ + {name: "HSMTokenCompletesRebuiltChasmOp", nexusOpInitiallyInHSM: true}, + {name: "ChasmTokenCompletesRebuiltHSMOp", nexusOpInitiallyInHSM: false}, + } + + for _, tc := range testCases { + s.Run(tc.name, func(s *NexusWorkflowTestSuite) { + // CHASM tree + callbacks are enabled in both directions; only the per-op creation policy is + // flipped during the test so the reset rebuilds the op into the OTHER tree. + env := s.newTestEnv(true) + ctx := env.Context() + taskQueue := testcore.RandomizeStr(s.T().Name()) + nsName := env.Namespace().String() + // Use the env-level (namespace-scoped) override: on a shared cluster a global + // cluster.OverrideDynamicConfig is less specific than the namespace-scoped baseline that + // newTestEnv installs, so it would not actually flip the flag for this namespace. + setFlag := func(on bool) { + // Mergeable, latest-wins override, so a later setFlag supersedes an earlier one. + env.OverrideDynamicConfig(chasmnexus.EnableChasmWorkflowOperations, on) + } + // The op's original tree fixes the completion token's format: HSM token vs CHASM token. + setFlag(!tc.nexusOpInitiallyInHSM) + + var callbackToken, publicCallbackURL string + h := nexustest.Handler{ + OnStartOperation: func( + ctx context.Context, + service, operation string, + input *nexus.LazyValue, + options nexus.StartOperationOptions, + ) (nexus.HandlerStartOperationResult[any], error) { + callbackToken = options.CallbackHeader.Get(commonnexus.CallbackTokenHeader) + publicCallbackURL = options.CallbackURL + return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil + }, + } + endpointName := env.createRandomExternalNexusServer(ctx, s.T(), h) + + callerWF := func(ctx workflow.Context) (string, error) { + c := workflow.NewNexusClient(endpointName, "service") + fut := c.ExecuteOperation(ctx, "operation", "input", workflow.NexusOperationOptions{}) + var result string + err := fut.Get(ctx, &result) + return result, err + } + + w := worker.New(env.SdkClient(), taskQueue, worker.Options{}) + w.RegisterWorkflow(callerWF) + s.NoError(w.Start()) + defer w.Stop() + + run, err := env.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{ + TaskQueue: taskQueue, + }, callerWF) + s.NoError(err) + wfExec := &commonpb.WorkflowExecution{WorkflowId: run.GetID(), RunId: run.GetRunID()} + + // Wait for NexusOperationStarted and find the WFT completed after it (the reset point). + var wftCompletedEventID int64 + s.EventuallyWithT(func(t *assert.CollectT) { + hr := historyrequire.New(t) + hist := env.GetHistory(nsName, wfExec) + opStartedEvent := hr.RequireHistoryEvent(hist, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED) + if opStartedEvent == nil { + return + } + wftCompletedIdx := slices.IndexFunc(hist, func(e *historypb.HistoryEvent) bool { + return e.EventType == enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED && e.EventId > opStartedEvent.EventId + }) + require.Positive(t, wftCompletedIdx, "expected WorkflowTaskCompleted after NexusOperationStarted") + wftCompletedEventID = hist[wftCompletedIdx].EventId + }, time.Second*10, time.Millisecond*200) + // Snapshot the original token now; the reset reapplies the started op (no re-invocation), so this + // stays the token minted for the op's ORIGINAL tree. + s.NotEmpty(callbackToken) + originalCallbackToken := callbackToken + originalCallbackURL := publicCallbackURL + + // Flip the creation policy so the reset rebuilds the op into the OTHER tree, then reset to just + // after the op started (the rebuilt op stays pending until our completion lands). + setFlag(tc.nexusOpInitiallyInHSM) + + resetResp, err := env.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{ + Namespace: nsName, + WorkflowExecution: wfExec, + Reason: "test cross-tree live completion after reset", + RequestId: uuid.NewString(), + WorkflowTaskFinishEventId: wftCompletedEventID, + }) + s.NoError(err) + s.NotEqual(run.GetRunID(), resetResp.RunId, "reset should produce a new run") + + resetHist := env.GetHistory(nsName, &commonpb.WorkflowExecution{WorkflowId: run.GetID(), RunId: resetResp.RunId}) + s.RequireHistoryEvent(resetHist, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED) + + // Verify the reset actually rebuilt the op into the OTHER tree (not merely that completion + // later succeeded). This is what makes the subsequent completion genuinely cross-tree: the + // original-tree token will route to a tree that no longer holds the op and must fall back. + desc, err := env.AdminClient().DescribeMutableState(ctx, &adminservice.DescribeMutableStateRequest{ + Namespace: nsName, + Execution: &commonpb.WorkflowExecution{WorkflowId: run.GetID(), RunId: resetResp.RunId}, + Archetype: chasm.WorkflowArchetype, + }) + s.NoError(err) + _, opInHSMTree := desc.GetDatabaseMutableState().GetExecutionInfo().GetSubStateMachinesByType()[nexusoperations.OperationMachineType] + opInChasmTree := false + for nodePath := range desc.GetDatabaseMutableState().GetChasmNodes() { + if strings.Contains(nodePath, "Operations") { + opInChasmTree = true + break + } + } + if tc.nexusOpInitiallyInHSM { + // Op created in HSM, flag flipped on -> reset must rebuild it into the CHASM tree. + s.False(opInHSMTree, "op should have been rebuilt out of the HSM tree") + s.True(opInChasmTree, "op should have been rebuilt into the CHASM tree") + } else { + // Op created in CHASM, flag flipped off -> reset must rebuild it into the HSM tree. + s.True(opInHSMTree, "op should have been rebuilt into the HSM tree") + s.False(opInChasmTree, "op should have been rebuilt out of the CHASM tree") + } + + // Complete using the ORIGINAL token (old tree's format) against the op now rebuilt into the other + // tree: this is the live cross-tree completion fallback under test. + completion := nexusrpc.CompleteOperationOptions{ + Result: testcore.MustToPayload(s.T(), "result"), + Header: nexus.Header{commonnexus.CallbackTokenHeader: originalCallbackToken}, + } + s.NoError(s.sendNexusCompletionRequest(ctx, originalCallbackURL, completion)) + + // The reset run completes only after the cross-tree completion is applied to the rebuilt op. + var result string + resetRun := env.SdkClient().GetWorkflow(ctx, run.GetID(), resetResp.RunId) + s.NoError(resetRun.Get(ctx, &result)) + s.Equal("result", result) + + resetHist = env.GetHistory(nsName, &commonpb.WorkflowExecution{WorkflowId: run.GetID(), RunId: resetResp.RunId}) + s.RequireHistoryEvent(resetHist, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED) + }) + } +} + +// TestNexusOperationCompletionRejectsBogusScheduleTimeRun verifies the schedule-time-run gate on the +// HSM-token -> CHASM-op fallback: a completion whose token has valid workflow/event/request IDs but a +// bogus (never-existed) run ID must return NotFound rather than being re-targeted at the current CHASM op. +func (s *NexusWorkflowTestSuite) TestNexusOperationCompletionRejectsBogusScheduleTimeRun(chasmEnabled bool) { + // Controls the creation-policy flag itself, so run it once (not per-suite-mode). + if chasmEnabled { + s.T().Skip("controls the flag explicitly; run once via the HSM suite") + } + + env := s.newTestEnv(true) + ctx := env.Context() + taskQueue := testcore.RandomizeStr(s.T().Name()) + nsName := env.Namespace().String() + setFlag := func(on bool) { + env.OverrideDynamicConfig(chasmnexus.EnableChasmWorkflowOperations, on) + } + // Op created in HSM -> HSM-format callback token. + setFlag(false) + + var callbackToken, publicCallbackURL string + h := nexustest.Handler{ + OnStartOperation: func(ctx context.Context, service, operation string, input *nexus.LazyValue, options nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[any], error) { + callbackToken = options.CallbackHeader.Get(commonnexus.CallbackTokenHeader) + publicCallbackURL = options.CallbackURL + return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil + }, + } + endpointName := env.createRandomExternalNexusServer(ctx, s.T(), h) + + callerWF := func(ctx workflow.Context) (string, error) { + c := workflow.NewNexusClient(endpointName, "service") + fut := c.ExecuteOperation(ctx, "operation", "input", workflow.NexusOperationOptions{}) + var result string + err := fut.Get(ctx, &result) + return result, err + } + + w := worker.New(env.SdkClient(), taskQueue, worker.Options{}) + w.RegisterWorkflow(callerWF) + s.NoError(w.Start()) + defer w.Stop() + + run, err := env.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{TaskQueue: taskQueue}, callerWF) + s.NoError(err) + wfExec := &commonpb.WorkflowExecution{WorkflowId: run.GetID(), RunId: run.GetRunID()} + + var wftCompletedEventID int64 + s.EventuallyWithT(func(t *assert.CollectT) { + hr := historyrequire.New(t) + hist := env.GetHistory(nsName, wfExec) + opStartedEvent := hr.RequireHistoryEvent(hist, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED) + if opStartedEvent == nil { + return + } + wftCompletedIdx := slices.IndexFunc(hist, func(e *historypb.HistoryEvent) bool { + return e.EventType == enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED && e.EventId > opStartedEvent.EventId + }) + require.Positive(t, wftCompletedIdx, "expected WorkflowTaskCompleted after NexusOperationStarted") + wftCompletedEventID = hist[wftCompletedIdx].EventId + }, time.Second*10, time.Millisecond*200) + s.NotEmpty(callbackToken) + + // Flip the flag so the reset rebuilds the op into CHASM; the HSM token's op no longer lives in HSM, + // which routes the completion into completeNexusOperationChasmFallback. + setFlag(true) + _, err = env.FrontendClient().ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{ + Namespace: nsName, + WorkflowExecution: wfExec, + Reason: "test bogus schedule-time run", + RequestId: uuid.NewString(), + WorkflowTaskFinishEventId: wftCompletedEventID, + }) + s.NoError(err) + + // Tamper ONLY the run ID (keep valid workflow/event/request IDs), then re-sign the token. + gen := &commonnexus.CallbackTokenGenerator{} + decoded, err := commonnexus.DecodeCallbackToken(callbackToken) + s.NoError(err) + completionToken, err := gen.DecodeCompletion(decoded) + s.NoError(err) + completionToken.RunId = uuid.NewString() // never-existed run + tamperedCallbackToken, err := gen.Tokenize(completionToken) + s.NoError(err) + + completion := nexusrpc.CompleteOperationOptions{ + Result: testcore.MustToPayload(s.T(), "result"), + Header: nexus.Header{commonnexus.CallbackTokenHeader: tamperedCallbackToken}, + } + err = s.sendNexusCompletionRequest(ctx, publicCallbackURL, completion) + var handlerErr *nexus.HandlerError + s.ErrorAs(err, &handlerErr) + s.Equal(nexus.HandlerErrorTypeNotFound, handlerErr.Type, "bogus schedule-time run must yield NotFound") +} + +// TestNexusOperationCompletionNotFoundInEitherTree verifies the cross-tree fallback preserves a +// NotFound when the op exists in neither tree: a completion whose ScheduledEventId does not match any +// operation must still be rejected with NotFound rather than masked by the fallback. +func (s *NexusWorkflowTestSuite) TestNexusOperationCompletionNotFoundInEitherTree(chasmEnabled bool) { + env := s.newTestEnv(chasmEnabled) + ctx := env.Context() + taskQueue := testcore.RandomizeStr(s.T().Name()) + + var callbackToken, publicCallbackURL string + var run client.WorkflowRun + h := nexustest.Handler{ + OnStartOperation: func( + ctx context.Context, + service, operation string, + input *nexus.LazyValue, + options nexus.StartOperationOptions, + ) (nexus.HandlerStartOperationResult[any], error) { + callbackToken = options.CallbackHeader.Get(commonnexus.CallbackTokenHeader) + publicCallbackURL = options.CallbackURL + return &nexus.HandlerStartOperationResultAsync{OperationToken: "test"}, nil + }, + } + endpointName := env.createRandomExternalNexusServer(ctx, s.T(), h) + + callerWF := func(ctx workflow.Context) (string, error) { + c := workflow.NewNexusClient(endpointName, "service") + fut := c.ExecuteOperation(ctx, "operation", "input", workflow.NexusOperationOptions{}) + var result string + err := fut.Get(ctx, &result) + return result, err + } + + run, err := env.SdkClient().ExecuteWorkflow(ctx, client.StartWorkflowOptions{ + TaskQueue: taskQueue, + }, callerWF) + s.NoError(err) + + w := worker.New(env.SdkClient(), taskQueue, worker.Options{}) + w.RegisterWorkflow(callerWF) + s.NoError(w.Start()) + defer w.Stop() + + s.EventuallyWithT(func(t *assert.CollectT) { + hist := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: run.GetID()}) + historyrequire.New(t).RequireHistoryEvent(hist, enumspb.EVENT_TYPE_NEXUS_OPERATION_STARTED) + }, time.Second*10, time.Millisecond*200) + s.NotEmpty(callbackToken) + + // Tamper the token so its ScheduledEventId points at a non-existent operation in BOTH trees. + gen := &commonnexus.CallbackTokenGenerator{} + decodedToken, err := commonnexus.DecodeCallbackToken(callbackToken) + s.NoError(err) + completionToken, err := gen.DecodeCompletion(decodedToken) + s.NoError(err) + mutated := common.CloneProto(completionToken) + if len(mutated.GetComponentRef()) > 0 { + // CHASM token: rewrite the component path's last segment (the scheduled event ID). + ref := &persistencespb.ChasmComponentRef{} + s.NoError(ref.Unmarshal(mutated.GetComponentRef())) + s.NotEmpty(ref.ComponentPath) + ref.ComponentPath[len(ref.ComponentPath)-1] = "99999" + mutatedRef, err := ref.Marshal() + s.NoError(err) + mutated.ComponentRef = mutatedRef + } else { + // HSM token: rewrite the last state machine key's ID (the scheduled event ID). + s.NotEmpty(mutated.GetRef().GetPath()) + mutated.Ref.Path[len(mutated.Ref.Path)-1].Id = "99999" + } + mutatedCallbackToken, err := gen.Tokenize(mutated) + s.NoError(err) + + completion := nexusrpc.CompleteOperationOptions{ + Result: testcore.MustToPayload(s.T(), "result"), + Header: nexus.Header{commonnexus.CallbackTokenHeader: mutatedCallbackToken}, + } + err = s.sendNexusCompletionRequest(ctx, publicCallbackURL, completion) + var handlerErr *nexus.HandlerError + s.ErrorAs(err, &handlerErr) + s.Equal(nexus.HandlerErrorTypeNotFound, handlerErr.Type, "op missing in both trees must surface NotFound") +} + func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletionBeforeStart(chasmEnabled bool) { env := s.newTestEnv(chasmEnabled) ctx := env.Context() @@ -1759,9 +2249,6 @@ NexusOperationStarted`, hist) } func (s *NexusWorkflowTestSuite) TestNexusOperationAsyncCompletionAfterReset(chasmEnabled bool) { - if chasmEnabled { - s.T().Skip("Blocked on CHASM Nexus async completion after reset support") - } env := s.newTestEnv(chasmEnabled) ctx := env.Context() taskQueue := testcore.RandomizeStr(s.T().Name())