diff --git a/service/history/configs/config.go b/service/history/configs/config.go index a6a3804a3b..ce98025647 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" @@ -72,6 +73,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 @@ -499,6 +501,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/history_engine.go b/service/history/history_engine.go index 23ccb7f532..e31b591bd3 100644 --- a/service/history/history_engine.go +++ b/service/history/history_engine.go @@ -244,7 +244,7 @@ func NewEngineWithShardContext( historyEngImpl.queueProcessors[processor.Category()] = processor } - historyEngImpl.eventsReapplier = ndc.NewEventsReapplier(shard.StateMachineRegistry(), shard.GetMetricsHandler(), logger) + historyEngImpl.eventsReapplier = ndc.NewEventsReapplier(shard.StateMachineRegistry(), shard.ChasmWorkflowRegistry(), shard.GetMetricsHandler(), logger) 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 525ad7c1ec..d0a773435c 100644 --- a/service/history/interfaces/shard_context.go +++ b/service/history/interfaces/shard_context.go @@ -11,6 +11,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" @@ -121,6 +122,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 6d16cceccf..0fefd04a30 100644 --- a/service/history/interfaces/shard_context_mock.go +++ b/service/history/interfaces/shard_context_mock.go @@ -21,6 +21,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" @@ -151,6 +152,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() @@ -984,6 +999,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 e01ad8d56f..5ff94f2fb4 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,22 +28,25 @@ type ( } EventsReapplierImpl struct { - stateMachineRegistry *hsm.Registry - metricsHandler metrics.Handler - logger log.Logger + stateMachineRegistry *hsm.Registry + chasmWorkflowRegistry *chasmworkflow.Registry + metricsHandler metrics.Handler + logger log.Logger } ) func NewEventsReapplier( stateMachineRegistry *hsm.Registry, + chasmWorkflowRegistry *chasmworkflow.Registry, metricsHandler metrics.Handler, logger log.Logger, ) *EventsReapplierImpl { return &EventsReapplierImpl{ - stateMachineRegistry: stateMachineRegistry, - metricsHandler: metricsHandler, - logger: logger, + stateMachineRegistry: stateMachineRegistry, + chasmWorkflowRegistry: chasmWorkflowRegistry, + metricsHandler: metricsHandler, + logger: logger, } } @@ -57,7 +61,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/events_reapplier_test.go b/service/history/ndc/events_reapplier_test.go index b78f343b7a..4a7433b050 100644 --- a/service/history/ndc/events_reapplier_test.go +++ b/service/history/ndc/events_reapplier_test.go @@ -15,6 +15,7 @@ import ( updatepb "go.temporal.io/api/update/v1" enumsspb "go.temporal.io/server/api/enums/v1" persistencespb "go.temporal.io/server/api/persistence/v1" + chasmworkflow "go.temporal.io/server/chasm/lib/workflow" "go.temporal.io/server/common/definition" "go.temporal.io/server/common/log" "go.temporal.io/server/common/metrics" @@ -54,6 +55,7 @@ func (s *nDCEventReapplicationSuite) SetupTest() { metricsHandler := metrics.NoopMetricsHandler s.nDCReapplication = NewEventsReapplier( hsm.NewRegistry(), + chasmworkflow.NewRegistry(), metricsHandler, logger, ) diff --git a/service/history/ndc/workflow_resetter.go b/service/history/ndc/workflow_resetter.go index 8b3ede020b..04d74e855a 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,18 +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 - } - 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 - } + // 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 outcome == cherryPickFallback { + // HSM doesn't own this op (unknown type, or component not in the HSM tree): try the CHASM tree. + outcome, err = cherryPickChasmEvent(ctx, mutableState, chasmWorkflowRegistry, event, resetReapplyExcludeTypes) + if err != nil { + 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 +1058,83 @@ func reapplyEvents( return reappliedEvents, nil } +// cherryPickOutcome is the result of cherry-picking an event against a 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. +// As the last framework tried, an event type it doesn't define is skipped (nothing left to fall back to), and a +// CHASM-defined event on a workflow without CHASM enabled returns an error rather than being silently dropped. +func cherryPickChasmEvent( + ctx context.Context, + mutableState historyi.MutableState, + chasmWorkflowRegistry *chasmworkflow.Registry, + event *historypb.HistoryEvent, + resetReapplyExcludeTypes map[enumspb.ResetReapplyExcludeType]struct{}, +) (cherryPickOutcome, error) { + // The event-type lookup is a cheap, side-effect-free registry check, so it runs before consulting mutable state. + def, ok := chasmWorkflowRegistry.EventDefinitionByEventType(event.GetEventType()) + if !ok { + // HSM was already tried and CHASM doesn't define this event type either: nothing to fall back to, skip it. + return cherryPickSkipped, nil + } + if !mutableState.ChasmEnabled() { + // The event is a CHASM feature event, but the workflow has no CHASM tree to reapply it into. Fail loudly + // rather than dropping it silently: reapplying a CHASM event requires CHASM to be enabled for the workflow. + return cherryPickSkipped, serviceerror.NewInternalf( + "cannot reapply CHASM event %v during reset: CHASM is not enabled for this workflow", event.GetEventType()) + } + 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 11c0296e1d..8a776ec063 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" @@ -100,6 +102,7 @@ func (s *workflowResetterSuite) SetupTest() { tests.NewDynamicConfig(), ) s.mockExecutionMgr = s.mockShard.Resource.ExecutionMgr + s.mockShard.SetChasmWorkflowRegistry(chasmworkflow.NewRegistry()) s.mockTransaction = workflow.NewMockTransaction(s.controller) s.workflowResetter = NewWorkflowResetter( @@ -959,7 +962,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, chasmworkflow.NewRegistry(), tc.events, nil, "", tcReset.isReset) s.NoError(err) }) } @@ -1026,7 +1029,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, chasmworkflow.NewRegistry(), tc.events, nil, "", tcReset.isReset) s.NoError(err) }) } @@ -1261,7 +1264,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, chasmworkflow.NewRegistry(), events, nil, "", tc.isReset) s.NoError(err) s.Equal(tc.expected, appliedEvents) @@ -1332,7 +1335,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, chasmworkflow.NewRegistry(), events, excludes, "", false) s.Empty(reappliedEvents) s.NoError(err) @@ -1358,7 +1361,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, chasmworkflow.NewRegistry(), events, excludes, "", true) s.Empty(reappliedEvents) s.NoError(err) } @@ -1653,7 +1656,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, chasmworkflow.NewRegistry(), events, nil, "", true) s.Error(err) s.Contains(err.Error(), tc.expectedErrorContains) s.Empty(appliedEvents) @@ -1701,7 +1704,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, chasmworkflow.NewRegistry(), events, nil, "", true) s.NoError(err) s.Empty(appliedEvents) // Event should be skipped } @@ -1739,7 +1742,251 @@ 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, chasmworkflow.NewRegistry(), []*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 + wantErrMsg string + }{ + { + // Event type unknown to CHASM is rejected by the registry lookup before mutable state is consulted, so + // ChasmEnabled is never called. CHASM is the last framework tried, so this is skipped, not a fallback. + name: "event type unknown to chasm is skipped", + registry: chasmworkflow.NewRegistry(), + setupMock: func(*historyi.MockMutableState) {}, + wantOutcome: cherryPickSkipped, + }, + { + // A CHASM-defined event on a workflow without CHASM enabled is an error, not a silent skip. + name: "chasm disabled is an error", + registry: newChasmRegistryWithEvent(eventType, nil), + setupMock: func(ms *historyi.MockMutableState) { ms.EXPECT().ChasmEnabled().Return(false) }, + wantOutcome: cherryPickSkipped, + wantErrMsg: "CHASM is not enabled for this workflow", + }, + { + 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) + switch { + case tc.wantErr != nil: + s.ErrorIs(err, tc.wantErr) + case tc.wantErrMsg != "": + s.ErrorContains(err, tc.wantErrMsg) + default: + 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) + + 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 17df0d29a4..5c7f1fea00 100644 --- a/service/history/shard/context_factory.go +++ b/service/history/shard/context_factory.go @@ -3,6 +3,7 @@ package shard import ( otellog "go.opentelemetry.io/otel/log" "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" @@ -62,6 +63,7 @@ type ( StateMachineRegistry *hsm.Registry ChasmRegistry *chasm.Registry + ChasmWorkflowRegistry *chasmworkflow.Registry EndpointRegistry commonnexus.EndpointRegistry HandoverTrackerFactory HandoverTrackerFactory } @@ -107,6 +109,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 65ee4970eb..f68920b621 100644 --- a/service/history/shard/context_impl.go +++ b/service/history/shard/context_impl.go @@ -20,6 +20,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" @@ -150,8 +151,9 @@ type ( stateMachineRegistry *hsm.Registry - chasmRegistry *chasm.Registry - endpointRegistry chasm.EndpointRegistry + chasmRegistry *chasm.Registry + chasmWorkflowRegistry *chasmworkflow.Registry + endpointRegistry chasm.EndpointRegistry businessIDRateLimiters cache.Cache } @@ -2066,6 +2068,7 @@ func newContext( eventsCache events.Cache, stateMachineRegistry *hsm.Registry, chasmRegistry *chasm.Registry, + chasmWorkflowRegistry *chasmworkflow.Registry, endpointRegistry chasm.EndpointRegistry, handoverTrackerFactory HandoverTrackerFactory, ) (*ContextImpl, error) { @@ -2117,6 +2120,7 @@ func newContext( ioSemaphore: locks.NewPrioritySemaphore(ioConcurrency), stateMachineRegistry: stateMachineRegistry, chasmRegistry: chasmRegistry, + chasmWorkflowRegistry: chasmWorkflowRegistry, endpointRegistry: endpointRegistry, businessIDRateLimiters: cache.New( historyConfig.BusinessIDReuseLimiterCacheSize(), @@ -2243,6 +2247,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 9b03a310a6..4bdc9081e9 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 }