From 34ae4169abf7cc0cab6975da6aadcdccca81cbd4 Mon Sep 17 00:00:00 2001 From: tekkaya <86028633+tekkaya@users.noreply.github.com> Date: Fri, 26 Jun 2026 18:35:55 -0700 Subject: [PATCH 1/2] Support CHASM Nexus operations across workflow reset and async completion Nexus operations can be backed by either the HSM tree or the CHASM tree, and both coexist on the same mutable state. Before this change, reset/rebuild/reapply dispatched Nexus events only through HSM, and async completion committed to one framework by token type. So once nexusoperation.enableChasmWorkflowOperations is on, CHASM-backed ops were silently not rebuilt on reset, and an op that "moved" trees via reset could not be completed. This wires reset, state rebuild, the event reapplier, and async completion to support either tree using a probe-and-fallback approach keyed by the cross-tree-stable ScheduledEventId (HSM node ID; CHASM Workflow.Operations map key). No new event field/marker. - Reset reapply (ndc/workflow_resetter.go): reapplyEvents probes HSM, then falls back to the CHASM workflow registry, with an explicit cherryPickOutcome (applied/skipped/fallback) to distinguish "recognized but not cherry-pickable" (skip) from "not owned here" (fallback). - Flag-driven rebuild-create (workflow/mutable_state_rebuilder.go): the NexusOperationScheduled create event is routed by the per-namespace enableChasmWorkflowOperations flag (configs.Config EnableChasmNexusWorkflowOperations); on CHASM create error it falls back to HSM. Non-create events use HSM-first / CHASM-fallback. - Completion-token fallback, both directions, server-side in History (service/history/handler.go): - HSM token -> CHASM op: on HSM NotFound, recover ScheduledEventId from the StateMachineRef and complete the op on the current run's CHASM tree. - CHASM token -> after-reset re-resolution: a CHASM token pins the schedule-time run via its ChasmComponentRef; a reset closes that run and rebuilds the op on a new current run. On failure against the pinned run, re-resolve the op by ScheduledEventId on the current run and complete it in CHASM or fall back to the HSM handler depending on where the rebuild placed it. Namespace/business IDs are read from the ChasmComponentRef, not the legacy NamespaceId/WorkflowId completion fields. - Fallback gating keeps invalid/unrelated tokens rejected rather than silently re-targeting the current run: archetype must be the Workflow archetype; the pinned run must actually exist (a reset closes it but leaves it readable, whereas a bogus run id does not resolve) -- a positive existence check, not error-type based; and namespace EnableChasm must hold (avoids a noopChasmTree). The frontend completion handler stays a pure router by token type. - Event reapplier (ndc/events_reapplier.go): plumbs the real CHASM workflow registry through EventsReapplierImpl so the replication / current-run reapply path gets the same fallback. - Plumbing: new ShardContext.ChasmWorkflowRegistry() accessor (interface, ContextImpl, factory, regenerated mock, test util), fx-auto-wired. Tests: - Unit: handler_test.go (token/ref helpers, completion conversion); mutable_state_rebuilder_test.go (flag-on CHASM-create-error -> HSM). - Functional (tests/nexus_workflow_test.go): TestNexusOperationResetCrossTreeCompletion (op created in one tree, flag flipped, reset rebuilds into the other tree and the completion reapplies across trees, both directions); TestNexusOperationCompletionNotFoundInEitherTree (tampered token -> NotFound preserved); TestNexusOperationAsyncCompletionAfterReset now enabled for the CHASM variant (live completion after reset resolves onto the reset-created current run). --- service/history/configs/config.go | 3 + service/history/handler.go | 345 +++++++++++++++ service/history/handler_test.go | 158 +++++++ service/history/history_engine.go | 3 +- service/history/interfaces/shard_context.go | 3 + .../history/interfaces/shard_context_mock.go | 29 ++ service/history/ndc/events_reapplier.go | 18 +- service/history/ndc/workflow_resetter.go | 117 ++++- service/history/ndc/workflow_resetter_test.go | 261 +++++++++++- service/history/shard/context_factory.go | 3 + service/history/shard/context_impl.go | 12 +- service/history/shard/context_testutil.go | 5 + service/history/statemachine_environment.go | 10 +- .../workflow/mutable_state_rebuilder.go | 117 ++++- .../workflow/mutable_state_rebuilder_test.go | 267 ++++++++++++ tests/nexus_workflow_test.go | 399 +++++++++++++++++- 16 files changed, 1715 insertions(+), 35 deletions(-) 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..574aa1b0b5e 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,111 @@ 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) + } + + 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: request.GetCompletion().GetNamespaceId(), + BusinessID: request.GetCompletion().GetWorkflowId(), + }, + 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 +2328,255 @@ 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 { + opErr = &nexus.OperationError{ + State: nexus.OperationStateFailed, + 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..989972a922d 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,155 @@ 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 + }{ + { + name: "success yields no operation error", + request: &historyservice.CompleteNexusOperationChasmRequest{ + Completion: &tokenspb.NexusOperationCompletion{RequestId: "r"}, + Outcome: &historyservice.CompleteNexusOperationChasmRequest_Success{Success: &commonpb.Payload{}}, + }, + wantOpErr: false, + }, + { + name: "failure yields an operation error", + request: &historyservice.CompleteNexusOperationChasmRequest{ + Completion: &tokenspb.NexusOperationCompletion{RequestId: "r"}, + Outcome: &historyservice.CompleteNexusOperationChasmRequest_Failure{Failure: &failurepb.Failure{Message: "boom"}}, + }, + wantOpErr: true, + }, + } + + 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) + } 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..d0b59ad09f7 100644 --- a/tests/nexus_workflow_test.go +++ b/tests/nexus_workflow_test.go @@ -1038,6 +1038,402 @@ 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) + }) + } +} + +// 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 +2155,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()) From 2e7ae0fdd6895a1c2cf5dee3c214f2f226ae193e Mon Sep 17 00:00:00 2001 From: tekkaya <86028633+tekkaya@users.noreply.github.com> Date: Wed, 1 Jul 2026 12:32:27 -0700 Subject: [PATCH 2/2] Harden cross-tree Nexus completion fallback Two fixes to the server-side completion-token fallback: - Gate the HSM-token -> CHASM-op fallback (completeNexusOperationChasmFallback) on the token's schedule-time run actually existing, mirroring the CHASM-token path. Previously it re-targeted the current run using only ns/workflow/ scheduledEventId, so a bogus or tampered run ID with an otherwise valid event/request ID could complete the current-run op instead of preserving NotFound. - Preserve the canceled outcome in the CHASM-token -> HSM-op fallback. CompleteNexusOperationChasmRequest carries no explicit outcome state, so nexusOperationErrorFromChasmRequest defaulted non-OperationError failures to Failed. It now infers Canceled from CanceledFailureInfo, matching the native CHASM path (Operation.HandleNexusCompletion), so a completion records the same terminal state (canceled vs failed) regardless of which tree the reset placed the op in. --- service/history/handler.go | 29 ++++++++-- service/history/handler_test.go | 17 +++++- tests/nexus_workflow_test.go | 94 +++++++++++++++++++++++++++++++++ 3 files changed, 136 insertions(+), 4 deletions(-) diff --git a/service/history/handler.go b/service/history/handler.go index 574aa1b0b5e..79a7dc955b1 100644 --- a/service/history/handler.go +++ b/service/history/handler.go @@ -2213,6 +2213,20 @@ func (h *Handler) completeNexusOperationChasmFallback( 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 @@ -2223,8 +2237,8 @@ func (h *Handler) completeNexusOperationChasmFallback( return h.applyChasmNexusCompletionOnCurrentRun( ctx, chasm.ExecutionKey{ - NamespaceID: request.GetCompletion().GetNamespaceId(), - BusinessID: request.GetCompletion().GetWorkflowId(), + NamespaceID: namespaceID, + BusinessID: businessID, }, scheduledEventID, completion, @@ -2551,8 +2565,17 @@ func nexusOperationErrorFromChasmRequest( } 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: nexus.OperationStateFailed, + State: state, Message: "nexus operation completed unsuccessfully", Cause: recvdErr, } diff --git a/service/history/handler_test.go b/service/history/handler_test.go index 989972a922d..62898b92338 100644 --- a/service/history/handler_test.go +++ b/service/history/handler_test.go @@ -197,6 +197,7 @@ func TestNexusOperationErrorFromChasmRequest(t *testing.T) { name string request *historyservice.CompleteNexusOperationChasmRequest wantOpErr bool + wantState nexus.OperationState // checked only when wantOpErr is true }{ { name: "success yields no operation error", @@ -207,12 +208,25 @@ func TestNexusOperationErrorFromChasmRequest(t *testing.T) { wantOpErr: false, }, { - name: "failure yields an operation error", + 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, }, } @@ -222,6 +236,7 @@ func TestNexusOperationErrorFromChasmRequest(t *testing.T) { 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/tests/nexus_workflow_test.go b/tests/nexus_workflow_test.go index d0b59ad09f7..b74b2a288b2 100644 --- a/tests/nexus_workflow_test.go +++ b/tests/nexus_workflow_test.go @@ -1352,6 +1352,100 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationCrossTreeCompletionAfterReset } } +// 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.