diff --git a/chasm/context.go b/chasm/context.go index 8eb17021408..90f31f4d69a 100644 --- a/chasm/context.go +++ b/chasm/context.go @@ -97,6 +97,10 @@ type MutableContext interface { // SetUserMetadata replaces the user metadata attached to the given component. SetUserMetadata(Component, *sdkpb.UserMetadata) error + // TimeSkippingConfigurator is the framework-provided surface for managing this execution's + // time-skipping configuration. It takes no Component argument as the configuration is execution-scoped. + TimeSkippingConfigurator + // Get a Ref for the component // This ref to the component state at the end of the transition // Same as Ref(Component) method in Context, @@ -277,6 +281,10 @@ func (c *mutableCtx) SetUserMetadata(component Component, md *sdkpb.UserMetadata return c.root.setComponentUserMetadata(component, md) } +func (c *mutableCtx) SetTimeSkippingConfig(config *commonpb.TimeSkippingConfig) { + c.root.backend.SetTimeSkippingConfig(config) +} + func (c *mutableCtx) withValue(key any, value any) Context { return &mutableCtx{ immutableCtx: ContextWithValue(c.immutableCtx, key, value), diff --git a/chasm/context_mock.go b/chasm/context_mock.go index 92b05f780b7..61d231edfe3 100644 --- a/chasm/context_mock.go +++ b/chasm/context_mock.go @@ -187,6 +187,13 @@ type MockMutableContext struct { Tasks []MockTask LinksByRequest map[Component]map[string][]*commonpb.Link UserMetadataByComponent map[Component]*sdkpb.UserMetadata + TimeSkippingConfig *commonpb.TimeSkippingConfig +} + +func (c *MockMutableContext) SetTimeSkippingConfig(config *commonpb.TimeSkippingConfig) { + c.mu.Lock() + defer c.mu.Unlock() + c.TimeSkippingConfig = config } func (c *MockMutableContext) AddTask(component Component, attributes TaskAttributes, payload any) { diff --git a/chasm/lib/activity/activity.go b/chasm/lib/activity/activity.go index 3efce2027c9..d2cad8a6c8e 100644 --- a/chasm/lib/activity/activity.go +++ b/chasm/lib/activity/activity.go @@ -55,6 +55,7 @@ var ( var _ chasm.VisibilitySearchAttributesProvider = (*Activity)(nil) var _ callback.CompletionSource = (*Activity)(nil) +var _ chasm.TimeSkippingRuntimeGate = (*Activity)(nil) type ActivityStore interface { // RecordCompleted applies the provided function to record activity completion @@ -124,6 +125,22 @@ func (a *Activity) LifecycleState(_ chasm.Context) chasm.LifecycleState { } } +// IsExecutionSkippable implements chasm.TimeSkippingRuntimeGate. The activity is skippable only when it is +// SCHEDULED (any other status — running, or a terminal activity still delivering completion callbacks, +// including while backing off between retries — is real work whose delivery resolves on the target's +// readiness, not this execution's virtual clock) and its next attempt dispatch is in the (virtual) +// future (a dispatch at or before now means an attempt is being dispatched or running, i.e. in flight). +func (a *Activity) IsExecutionSkippable(ctx chasm.Context) bool { + if a.GetStatus() != activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED { + return false + } + nextDispatch := a.attemptScheduleTime(a.LastAttempt.Get(ctx)) + if nextDispatch == nil || !ctx.Now(a).Before(nextDispatch.AsTime()) { + return false + } + return true +} + func (a *Activity) ContextMetadata(_ chasm.Context) map[string]string { md := make(map[string]string, 2) if actType := a.GetActivityType().GetName(); actType != "" { diff --git a/chasm/lib/activity/config.go b/chasm/lib/activity/config.go index b21d7df65d8..7b7d9fc5e30 100644 --- a/chasm/lib/activity/config.go +++ b/chasm/lib/activity/config.go @@ -39,6 +39,12 @@ var ( false, `Allows attaching completion callbacks to standalone activity executions.`, ) + + TimeSkippingEnabled = dynamicconfig.NewNamespaceBoolSetting( + "activity.timeSkippingEnabled", + false, + `Allows setting time_skipping_config on standalone activity executions. This is a standalone-activity-specific feature flag, separate from the workflow frontend.TimeSkippingEnabled flag.`, + ) ) type Config struct { @@ -53,6 +59,7 @@ type Config struct { MaxCallbacksPerExecution dynamicconfig.IntPropertyFnWithNamespaceFilter DefaultActivityRetryPolicy dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings] StartDelayEnabled dynamicconfig.BoolPropertyFnWithNamespaceFilter + TimeSkippingEnabled dynamicconfig.BoolPropertyFnWithNamespaceFilter VisibilityMaxPageSize dynamicconfig.IntPropertyFnWithNamespaceFilter } @@ -68,6 +75,7 @@ func ConfigProvider(dc *dynamicconfig.Collection) *Config { LongPollTimeout: LongPollTimeout.Get(dc), MaxIDLengthLimit: dynamicconfig.MaxIDLengthLimit.Get(dc), StartDelayEnabled: StartDelayEnabled.Get(dc), + TimeSkippingEnabled: TimeSkippingEnabled.Get(dc), MaxCallbacksPerExecution: callback.MaxPerExecution.Get(dc), VisibilityMaxPageSize: dynamicconfig.FrontendVisibilityMaxPageSize.Get(dc), } diff --git a/chasm/lib/activity/frontend.go b/chasm/lib/activity/frontend.go index 749e4b08033..c11f7ea395f 100644 --- a/chasm/lib/activity/frontend.go +++ b/chasm/lib/activity/frontend.go @@ -92,6 +92,12 @@ func (h *frontendHandler) StartActivityExecution(ctx context.Context, req *workf if !h.config.Enabled(req.GetNamespace()) { return nil, ErrStandaloneActivityDisabled } + if req.GetTimeSkippingConfig() != nil && !h.config.TimeSkippingEnabled(req.GetNamespace()) { + return nil, serviceerror.NewUnimplementedf( + "The Time-Skipping feature is not enabled for namespace %s", + req.GetNamespace(), + ) + } namespaceID, err := h.namespaceRegistry.GetNamespaceID(namespace.Name(req.GetNamespace())) if err != nil { diff --git a/chasm/lib/activity/handler.go b/chasm/lib/activity/handler.go index 33e7ed15296..3ffe750da25 100644 --- a/chasm/lib/activity/handler.go +++ b/chasm/lib/activity/handler.go @@ -95,6 +95,10 @@ func (h *handler) StartActivityExecution(ctx context.Context, req *activitypb.St } } + if request.GetTimeSkippingConfig() != nil { + mutableContext.SetTimeSkippingConfig(request.GetTimeSkippingConfig()) + } + err = TransitionScheduled.Apply(newActivity, mutableContext, nil) if err != nil { return nil, err diff --git a/chasm/node_backend_mock.go b/chasm/node_backend_mock.go index 0fcc7faba92..7c87e944f2e 100644 --- a/chasm/node_backend_mock.go +++ b/chasm/node_backend_mock.go @@ -5,6 +5,7 @@ import ( "sync" "time" + commonpb "go.temporal.io/api/common/v1" enumspb "go.temporal.io/api/enums/v1" historypb "go.temporal.io/api/history/v1" enumsspb "go.temporal.io/server/api/enums/v1" @@ -22,23 +23,25 @@ var _ NodeBackend = (*MockNodeBackend)(nil) // fields (thread-safe). type MockNodeBackend struct { // Optional function overrides. If nil, methods return zero-values. - HandleGetExecutionState func() *persistencespb.WorkflowExecutionState - HandleGetExecutionInfo func() *persistencespb.WorkflowExecutionInfo - HandleGetCurrentVersion func() int64 - HandleNextTransitionCount func() int64 - HandleGetApproximatePersistedSize func() int - HandleCurrentVersionedTransition func() *persistencespb.VersionedTransition - HandleGetWorkflowKey func() definition.WorkflowKey - HandleUpdateWorkflowStateStatus func(state enumsspb.WorkflowExecutionState, status enumspb.WorkflowExecutionStatus) (bool, error) - HandleIsWorkflow func() bool - HandleGetNexusCompletion func(ctx context.Context, requestID string) (nexusrpc.CompleteOperationOptions, error) - HandleGetNexusUpdateCompletion func(ctx context.Context, updateID string, requestID string) (nexusrpc.CompleteOperationOptions, error) - HandleAddHistoryEvent func(t enumspb.EventType, setAttributes func(*historypb.HistoryEvent)) *historypb.HistoryEvent - HandleLoadHistoryEvent func(ctx context.Context, token []byte) (*historypb.HistoryEvent, error) - HandleGenerateEventLoadToken func(event *historypb.HistoryEvent) ([]byte, error) - HandleHasAnyBufferedEvent func(filter func(*historypb.HistoryEvent) bool) bool - HandleGetNamespaceEntry func() *namespace.Namespace - HandleEndpointRegistry func() EndpointRegistry + HandleGetExecutionState func() *persistencespb.WorkflowExecutionState + HandleGetExecutionInfo func() *persistencespb.WorkflowExecutionInfo + HandleGetCurrentVersion func() int64 + HandleNextTransitionCount func() int64 + HandleGetApproximatePersistedSize func() int + HandleCurrentVersionedTransition func() *persistencespb.VersionedTransition + HandleGetWorkflowKey func() definition.WorkflowKey + HandleUpdateWorkflowStateStatus func(state enumsspb.WorkflowExecutionState, status enumspb.WorkflowExecutionStatus) (bool, error) + HandleIsWorkflow func() bool + HandleGetNexusCompletion func(ctx context.Context, requestID string) (nexusrpc.CompleteOperationOptions, error) + HandleGetNexusUpdateCompletion func(ctx context.Context, updateID string, requestID string) (nexusrpc.CompleteOperationOptions, error) + HandleAddHistoryEvent func(t enumspb.EventType, setAttributes func(*historypb.HistoryEvent)) *historypb.HistoryEvent + HandleLoadHistoryEvent func(ctx context.Context, token []byte) (*historypb.HistoryEvent, error) + HandleGenerateEventLoadToken func(event *historypb.HistoryEvent) ([]byte, error) + HandleHasAnyBufferedEvent func(filter func(*historypb.HistoryEvent) bool) bool + HandleGetNamespaceEntry func() *namespace.Namespace + HandleEndpointRegistry func() EndpointRegistry + HandleSetTimeSkippingConfig func(config *commonpb.TimeSkippingConfig) + HandleRecordTimeSkippingTransition func(ctx context.Context, transition TimeSkippingTransition, archetype ArchetypeID) error // Recorded calls (protected by mu). mu sync.Mutex @@ -111,6 +114,19 @@ func (m *MockNodeBackend) AddTasks(ts ...tasks.Task) { } } +func (m *MockNodeBackend) SetTimeSkippingConfig(config *commonpb.TimeSkippingConfig) { + if m.HandleSetTimeSkippingConfig != nil { + m.HandleSetTimeSkippingConfig(config) + } +} + +func (m *MockNodeBackend) RecordTimeSkippingTransition(ctx context.Context, transition TimeSkippingTransition, archetype ArchetypeID) error { + if m.HandleRecordTimeSkippingTransition != nil { + return m.HandleRecordTimeSkippingTransition(ctx, transition, archetype) + } + return nil +} + func (m *MockNodeBackend) DeleteCHASMPureTasks(maxScheduledTime time.Time) { m.mu.Lock() defer m.mu.Unlock() diff --git a/chasm/timeskipping_test.go b/chasm/timeskipping_test.go new file mode 100644 index 00000000000..e25ca3f820a --- /dev/null +++ b/chasm/timeskipping_test.go @@ -0,0 +1,115 @@ +package chasm + +import ( + "context" + "time" + + persistencespb "go.temporal.io/server/api/persistence/v1" + "go.temporal.io/server/service/history/tasks" + "go.uber.org/mock/gomock" + "google.golang.org/protobuf/types/known/timestamppb" +) + +// timeSkippingTaskNodes builds a tree with three CategoryTimer tasks already materialized +// (PhysicalTaskStatus == Created): a side-effect timer on the root, a side-effect timer on a child, +// and a pure timer on the root. The scheduled times are in the (virtual) future and carry no +// destination, so taskCategory classifies the side-effect tasks as CategoryTimer. +func (s *nodeSuite) timeSkippingTaskNodes() map[string]*persistencespb.ChasmNode { + now := s.timeSource.Now() + return map[string]*persistencespb.ChasmNode{ + "": { + Metadata: &persistencespb.ChasmNodeMetadata{ + InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1}, + LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1}, + Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{ + ComponentAttributes: &persistencespb.ChasmComponentAttributes{ + TypeId: testComponentTypeID, + PureTasks: []*persistencespb.ChasmComponentAttributes_Task{ + { + TypeId: testPureTaskTypeID, + ScheduledTime: timestamppb.New(now.Add(time.Hour)), + VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1}, + VersionedTransitionOffset: 1, + PhysicalTaskStatus: physicalTaskStatusCreated, + }, + }, + SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{ + { + TypeId: testSideEffectTaskTypeID, + ScheduledTime: timestamppb.New(now.Add(time.Hour)), + VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1}, + VersionedTransitionOffset: 2, + PhysicalTaskStatus: physicalTaskStatusCreated, + }, + }, + }, + }, + }, + }, + "SubComponent1": { + Metadata: &persistencespb.ChasmNodeMetadata{ + InitialVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1}, + LastUpdateVersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1}, + Attributes: &persistencespb.ChasmNodeMetadata_ComponentAttributes{ + ComponentAttributes: &persistencespb.ChasmComponentAttributes{ + TypeId: testSubComponent1TypeID, + SideEffectTasks: []*persistencespb.ChasmComponentAttributes_Task{ + { + TypeId: testSideEffectTaskTypeID, + ScheduledTime: timestamppb.New(now.Add(2 * time.Hour)), + VersionedTransition: &persistencespb.VersionedTransition{TransitionCount: 1}, + VersionedTransitionOffset: 3, + PhysicalTaskStatus: physicalTaskStatusCreated, + }, + }, + }, + }, + }, + }, + } +} + +// TestRegenerateTasksForTimeSkipping_RestampsEachTimerExactlyOnce verifies a time-skipping +// regeneration re-emits exactly one physical task per logical CategoryTimer task — two side-effect +// timers plus a single pure timer (the earliest across the tree) — and not duplicates. This is the +// "generated only once" guarantee: even though every task starts already Created, the re-stamp emits +// each one exactly once. +func (s *nodeSuite) TestRegenerateTasksForTimeSkipping_RestampsEachTimerExactlyOnce() { + root, err := s.newTestTree(s.timeSkippingTaskNodes()) + s.NoError(err) + + // regenerateTasksForTimeSkipping validates each task before re-stamping it; all three test timers + // are valid. + s.testLibrary.mockSideEffectTaskHandler.EXPECT(). + Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). + Return(true, nil).AnyTimes() + s.testLibrary.mockPureTaskHandler.EXPECT(). + Validate(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()). + Return(true, nil).AnyTimes() + + err = root.regenerateChasmTasksForTimeSkipping(NewContext(context.Background(), root)) + s.NoError(err) + + // 2 side-effect CategoryTimer tasks (root + SubComponent1) + 1 pure timer (earliest) = 3, all + // CategoryTimer. Not 6 (would indicate double-emission) and not 2 (would indicate the no-break + // re-stamp skipped already-Created tasks). + s.Equal(3, s.nodeBackend.NumTasksAdded()) + s.Len(s.nodeBackend.TasksByCategory[tasks.CategoryTimer], 3) +} + +// TestCloseTransaction_DoesNotRegenerateAlreadyMaterializedTimers verifies the complementary +// property: a plain CloseTransaction (no time-skipping transition, no user-state change) does NOT +// re-emit timer tasks that are already materialized. This is what keeps the active cluster from +// generating new timer tasks on every transaction — only a skip (via regenerateTasksForTimeSkipping) +// re-stamps them. +func (s *nodeSuite) TestCloseTransaction_DoesNotRegenerateAlreadyMaterializedTimers() { + root, err := s.newTestTree(s.timeSkippingTaskNodes()) + s.NoError(err) + + // Time skipping is not enabled (the mock backend's GetExecutionInfo returns nil, so TimeSkippingInfo + // is nil) and nothing was mutated, so the normal generation loop must skip the already-Created timer + // tasks. + _, err = root.CloseTransaction() + s.NoError(err) + s.Equal(0, s.nodeBackend.NumTasksAdded()) +} diff --git a/chasm/timeskippping.go b/chasm/timeskippping.go new file mode 100644 index 00000000000..0ef9a5badbb --- /dev/null +++ b/chasm/timeskippping.go @@ -0,0 +1,115 @@ +package chasm + +import ( + "time" + + commonpb "go.temporal.io/api/common/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" +) + +// Time skipping is a contract between the CHASM framework and a root component, split along the +// configuration / runtime axis: +// +// - TimeSkippingConfigurator (PROVIDED BY the framework, CALLED BY the component) is the +// configuration-time surface. It is embedded in MutableContext and lets the component opt in / +// adjust the config via SetTimeSkippingConfig. +// - TimeSkippingRuntimeGate (component-implemented, MANDATORY) is the runtime surface the framework +// consults each transaction to decide whether the execution is idle enough to skip. +// - TimeSkippingRuntimeTargetProvider (component-implemented, OPTIONAL) lets a component nominate the +// next skip target itself, in place of the framework's default timer-task tree scan. +// +// The two runtime interfaces are ROOT COMPONENT ONLY — the framework consults only the root component. + +// TimeSkippingConfigurator is the framework-provided, configuration-time surface (embedded in +// MutableContext) through which a root component opts into and adjusts time skipping for the whole +// execution. +type TimeSkippingConfigurator interface { + // SetTimeSkippingConfig sets the execution's time-skipping config: the first call establishes it, + // later calls update it in place (preserving the accumulated skipped duration). + SetTimeSkippingConfig(config *commonpb.TimeSkippingConfig) +} + +// TimeSkippingRuntimeGate is the mandatory, runtime half of the time-skipping contract: the framework +// asks it, once per transaction, whether the execution may be skipped at all. +type TimeSkippingRuntimeGate interface { + // IsExecutionSkippable reports whether the execution may have its virtual clock fast-forwarded right + // now. It returns true only when the execution is idle. The framework skips no time while it returns + // false, so in-flight work runs at real speed. + // + // A recommended implementation composes: + // 1. status check (optional): only states that can be idle are skip candidates; any active or + // terminal-but-still-working status returns false. + // 2. in-flight work check (mandatory): false whenever real work is dispatched or running (e.g. a + // scheduled attempt whose dispatch time is at or before now). + // 3. other execution-specific preconditions, as needed. + IsExecutionSkippable(ctx Context) bool +} + +// TimeSkippingRuntimeTargetProvider is the optional, by default the framework scans the tree's +// all pure and side-effect tasks that are future-scheduled and takes the earliest one as the skip target. +// If the execution needs a customized strategy, the root component can implement this interface +// and it will override the default strategy. +type TimeSkippingRuntimeTargetProvider interface { + // FindNextTargetTime provides a customized strategy to find the next skip target. + // The framework will provided a initilalized transition, and the implementation should just call + // transition.TrackEarliestFutureTime to set the target time, and it is allowed to leave the transition unchanged. + FindNextTargetTime(ctx Context, transition *TimeSkippingTransition) +} + +// TimeSkippingTransition is a time-skipping decision and is shared between the CHASM framework and +// the history substrate. +type TimeSkippingTransition struct { + CurrentTime time.Time + TargetTime time.Time + DisabledAfterFastForward bool +} + +// NewTimeSkippingTransition creates a new time-skipping transition with the given current time. +// A valid transition always has a non-zero CurrentTime; the fold methods below no-op without it. +func NewTimeSkippingTransition(currentTime time.Time) *TimeSkippingTransition { + return &TimeSkippingTransition{CurrentTime: currentTime} +} + +// IsValid reports whether the transition is worth applying: a real skip target, or a bare disable +// signal. Nil-safe. A transition without a current time is never valid — every meaningful field is +// derived relative to the current time, so without it there is nothing to apply. +func (t *TimeSkippingTransition) IsValid() bool { + return t != nil && !t.CurrentTime.IsZero() && (!t.TargetTime.IsZero() || t.DisabledAfterFastForward) +} + +// TrackEarliestFutureTime folds a business wake candidate (virtual time) in, keeping the earliest one at +// or after CurrentTime. Zero or past candidates (and a nil/uninitialized transition) are ignored. It +// never touches DisabledAfterFastForward — only the fast-forward budget disables time skipping. +func (t *TimeSkippingTransition) TrackEarliestFutureTime(candidate time.Time) { + if t == nil || t.CurrentTime.IsZero() || candidate.IsZero() || candidate.Before(t.CurrentTime) { + return + } + if t.TargetTime.IsZero() || candidate.Before(t.TargetTime) { + t.TargetTime = candidate + } +} + +// GateByFastForward folds the fast-forward target (the skip budget) in and decides whether +// reaching it disables time skipping. It MUST be called AFTER all business candidates have been folded, +// because the budget is a ceiling. +func (t *TimeSkippingTransition) GateByFastForward(ff *persistencespb.FastForwardInfo) { + if t == nil || t.CurrentTime.IsZero() { + return + } + if ff == nil || ff.GetHasReached() || ff.GetTargetTime() == nil || + ff.GetTargetTime().AsTime().IsZero() { + return + } + ffTargetTime := ff.GetTargetTime().AsTime() + // If a real candidate is scheduled strictly before the fast-forward target, we skip to + // that and the fast-forward budget is not yet exhausted — leave time skipping enabled. + if !t.TargetTime.IsZero() && t.TargetTime.Before(ffTargetTime) { + return + } + // Otherwise the fast-forward target is the earliest target: skip to it (clamped to the + // present by TrackEarliestFutureTime) and disable time skipping — the budget is reached. + // This is what lets the budget cap a chain of runs: a run with no earlier candidate + // consumes the remaining budget by skipping to the fast-forward and disabling. + t.TrackEarliestFutureTime(ffTargetTime) + t.DisabledAfterFastForward = true +} diff --git a/chasm/tree.go b/chasm/tree.go index b9076ff179f..919053ee978 100644 --- a/chasm/tree.go +++ b/chasm/tree.go @@ -234,6 +234,12 @@ type ( requestID string, ) (nexusrpc.CompleteOperationOptions, error) EndpointRegistry() EndpointRegistry + SetTimeSkippingConfig(config *commonpb.TimeSkippingConfig) + RecordTimeSkippingTransition(ctx context.Context, transition TimeSkippingTransition, archetype ArchetypeID) error + } + + nowProvider interface { + Now() time.Time } // NodePathEncoder is an interface for encoding and decoding node paths. @@ -1642,7 +1648,15 @@ func (n *Node) dataNodePath( func (n *Node) Now( _ Component, ) time.Time { + // Delegate to the backend when it owns a (time-skipping-aware) clock so that CHASM and the rest + // of the execution share a single clock; mutable state's clock reflects the execution's + // time-skipping offset. Fall back to the tree's own timeSource for backends that don't provide + // one (e.g. bare test mocks). + // // TODO: Now() could be different for components after we support Pause for CHASM components. + if np, ok := n.backend.(nowProvider); ok { + return np.Now() + } return n.timeSource.Now() } @@ -1716,6 +1730,12 @@ func (n *Node) CloseTransaction() (NodesMutation, error) { return NodesMutation{}, err } + // time skippign should be called at the end of the transaction, + // so that all invalid tasks are deleted. + if err := n.closeTransactionHandleTimeSkipping(); err != nil { + return NodesMutation{}, err + } + // Both user & system data mutation need to be returned and persisted. maps.Copy(n.mutation.UpdatedNodes, n.systemMutation.UpdatedNodes) maps.Copy(n.mutation.DeletedNodes, n.systemMutation.DeletedNodes) @@ -2085,6 +2105,158 @@ func (n *Node) closeTransactionUpdateComponentTasks( ) } +func (n *Node) closeTransactionHandleTimeSkipping() error { + + // step1: time skipping only works on the root node + // todo@time-skipping: it seems CloseTransaction is only a method reasonable for the root node + // but currently the framework doesn't enforce this contract + if !softassert.That(n.logger, n.parent == nil, "closeTransactionHandleTimeSkipping must run on the root node") { + return nil + } + if n.ArchetypeID() == WorkflowArchetypeID { + // todo: remove after workflows get migrated to CHASM + return nil + } + tsi := n.backend.GetExecutionInfo().GetTimeSkippingInfo() + if !tsi.GetConfig().GetEnabled() { + return nil + } + + // The chasm framework distinguishes active vs passive via isActiveStateDirty: it is set only for + // active-cluster user-data mutations and is never set by replication (ApplyMutation/ApplySnapshot) + // — see its declaration and the same "active cluster" gate in closeTransactionUpdateComponentTasks. + // + // NOTE: isActiveStateDirty tracks chasm component user-data changes only. It does NOT track + // TimeSkippingInfo changes — TimeSkippingInfo lives in mutable state's WorkflowExecutionInfo (not a + // chasm node) and is tracked separately by MutableStateImpl.timeSkippingInfoUpdated. That is fine + // here: the decision is driven by the component work that schedules a future task worth skipping to + // (which sets isActiveStateDirty), and the skip the decision records marks the TimeSkippingInfo + // dirty via its own flag. The one implication is that a transaction which only changes + // TimeSkippingInfo (e.g. SetTimeSkippingConfig) without touching chasm state won't trigger the + // decision; it is picked up on the next active transaction. + if !n.isActiveStateDirty { + // Passive cluster: the active cluster already made the time-skipping decision and replicated the + // resulting TimeSkippingInfo (including the new accumulatedSkippedDuration). We must NOT re-run + // the decision here — recording another skip on top of the replicated one would diverge from + // the active cluster. + // + // TODO(time-skipping/chasm): on the passive cluster, detect that the replicated + // accumulatedSkippedDuration changed since the physical timer tasks were last generated and, if + // so, re-stamp them via regenerateTasksForTimeSkipping (the same no-break CategoryTimer + // regeneration the active path uses) — comparing the current accumulatedSkippedDuration against + // a baseline captured at tree load and refreshed each CloseTransaction. Until then, passive + // timer tasks are only re-stamped on the full-refresh replication path (taskRefresher.Refresh -> + // ChasmTree().RefreshTasks()); the incremental path (PartialRefresh, a no-op for CHASM) leaves + // them stale. State-based time-skipping replication is currently out of scope. + return nil + } + + // Active cluster: make and apply the time-skipping decision. + immutableContext := NewContext(context.Background(), n) + rootComponent, err := n.Component(immutableContext, ComponentRef{}) + if err != nil { + return err + } + tsRoot, ok := rootComponent.(TimeSkippingRuntimeGate) + if !ok { + // todo: add a metric for alert in real implementation + n.logger.Error( + "root component does not implement TimeSkippingRuntimeGate when executions have turned on time skipping", + tag.Error(fmt.Errorf("type: %s", reflect.TypeOf(rootComponent).Name()))) + return nil + } + + // step2: bail unless the execution is idle. + if !tsRoot.IsExecutionSkippable(immutableContext) { + return nil + } + + // step3: find the next skip target. A root component MAY implement TimeSkippingRuntimeTargetProvider + // to nominate the target itself; otherwise the framework scans the tree's timer tasks. Either way the + // framework owns the transition: it constructs it here, then applies the fast-forward budget and the + // validity gate below. + transition := NewTimeSkippingTransition(n.Now(nil)) + // if provider, ok := rootComponent.(TimeSkippingRuntimeTargetProvider); ok { + // provider.FindNextTargetTime(immutableContext, transition) + // } else { + n.defaultFindNextTargetTime(transition) + // } + transition.GateByFastForward(tsi.GetFastForwardInfo()) + if !transition.IsValid() { + return nil + } + + // step4: record and regenerate + // todo: a bad design of current implementation is that RecordTimeSkippingTransition called regenerateChasmFastForwardWakeTask + // for chasm-based executions, and this is a special treatment. And we also need this special treatment for passive cluster. + // The reason is current chasm MutableContext.AddTask requires a component. + if err := n.backend.RecordTimeSkippingTransition(context.Background(), *transition, n.ArchetypeID()); err != nil { + return err + } + if err := n.regenerateChasmTasksForTimeSkipping(immutableContext); err != nil { + return err + } + return nil +} + +func (n *Node) defaultFindNextTargetTime(transition *TimeSkippingTransition) error { + now := transition.CurrentTime + for _, node := range n.andAllChildren() { + componentAttr := node.serializedNode.GetMetadata().GetComponentAttributes() + if componentAttr == nil { + continue + } + for _, taskList := range [][]*persistencespb.ChasmComponentAttributes_Task{ + componentAttr.GetPureTasks(), + componentAttr.GetSideEffectTasks(), + } { + for _, task := range taskList { + if taskCategory(task) != tasks.CategoryTimer { + continue + } + scheduledTime := task.GetScheduledTime().AsTime() + if !scheduledTime.After(now) { + continue + } + transition.TrackEarliestFutureTime(scheduledTime) + } + } + } + return nil +} + +func (n *Node) regenerateChasmTasksForTimeSkipping(ctx Context) error { + archetypeID := n.ArchetypeID() + + var firstPureTask *persistencespb.ChasmComponentAttributes_Task + var firstPureTaskNode *Node + + for nodePath, node := range n.andAllChildren() { + componentAttr := node.serializedNode.GetMetadata().GetComponentAttributes() + if componentAttr == nil { + continue + } + + for _, sideEffectTask := range componentAttr.GetSideEffectTasks() { + if taskCategory(sideEffectTask) != tasks.CategoryTimer { + continue + } + sideEffectTask.PhysicalTaskStatus = physicalTaskStatusNone + node.closeTransactionGeneratePhysicalSideEffectTask(sideEffectTask, nodePath, archetypeID) + } + + for _, pureTask := range componentAttr.GetPureTasks() { + pureTask.PhysicalTaskStatus = physicalTaskStatusNone + if firstPureTask == nil || comparePureTasks(pureTask, firstPureTask) < 0 { + firstPureTask = pureTask + firstPureTaskNode = node + } + } + } + + return n.closeTransactionGeneratePhysicalPureTask(firstPureTask, firstPureTaskNode, archetypeID) +} + func (n *Node) deserializeComponentTask( componentTask *persistencespb.ChasmComponentAttributes_Task, ) (any, error) { diff --git a/go.mod b/go.mod index 64032ba8611..a0cd5ba178c 100644 --- a/go.mod +++ b/go.mod @@ -64,7 +64,7 @@ require ( go.opentelemetry.io/otel/sdk v1.43.0 go.opentelemetry.io/otel/sdk/metric v1.43.0 go.opentelemetry.io/otel/trace v1.43.0 - go.temporal.io/api v1.63.0 + go.temporal.io/api v1.62.15-0.20260701034110-a7b499228361 go.temporal.io/auto-scaled-workers v0.0.0-20260622220320-9b1e3849116d go.temporal.io/sdk v1.41.1 go.uber.org/fx v1.24.0 diff --git a/go.sum b/go.sum index 36513b04183..b987aff262d 100644 --- a/go.sum +++ b/go.sum @@ -471,6 +471,10 @@ go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0 h1:R go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0/go.mod h1:I89cynRj8y+383o7tEQVg2SVA6SRgDVIouWPUVXjx0U= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN6Z8jsUeYv8J0lXRvygALXIzsmAeCcZE0= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4= +go.temporal.io/api v1.62.15-0.20260617232320-bf832cc7a07b h1:m4476YxxwyyK2v7+lCvXo4Fea22wXbpyvxKnfwN9gHU= +go.temporal.io/api v1.62.15-0.20260617232320-bf832cc7a07b/go.mod h1:0k75tRljEuELWGeXjEZZO7zYqBln4+1FrG6+IMOMy7Q= +go.temporal.io/api v1.62.15-0.20260701034110-a7b499228361 h1:51Ol1sND0K6BQi+Cxtk2pk0woTTUCXem8ZE8a7tIyPI= +go.temporal.io/api v1.62.15-0.20260701034110-a7b499228361/go.mod h1:0k75tRljEuELWGeXjEZZO7zYqBln4+1FrG6+IMOMy7Q= go.temporal.io/api v1.63.0 h1:YZFOTA0/thRUIUC4qunAWdHhPh/IG4vy/+WjfEvT+ZE= go.temporal.io/api v1.63.0/go.mod h1:0k75tRljEuELWGeXjEZZO7zYqBln4+1FrG6+IMOMy7Q= go.temporal.io/auto-scaled-workers v0.0.0-20260622220320-9b1e3849116d h1:f7+FCJHSrYWz9zvJp2OxKo8Fu/dsBUdnZZA+m5CEOS0= diff --git a/service/history/interfaces/mutable_state.go b/service/history/interfaces/mutable_state.go index a0887a58bd3..463c73b3a95 100644 --- a/service/history/interfaces/mutable_state.go +++ b/service/history/interfaces/mutable_state.go @@ -424,5 +424,8 @@ type ( AddWorkflowExecutionTimeSkippingTransitionedEvent( ctx context.Context, targetTime time.Time, disabledAfterFastForward bool) (*historypb.HistoryEvent, error) ApplyWorkflowExecutionTimeSkippingTransitionedEvent(ctx context.Context, event *historypb.HistoryEvent) error + // RecordTimeSkippingTransition is the archetype-aware time-skipping sink: workflows record a + // history event, other archetypes apply the transition to the TimeSkippingInfo directly. + RecordTimeSkippingTransition(ctx context.Context, transition chasm.TimeSkippingTransition, archetype chasm.ArchetypeID) error } ) diff --git a/service/history/interfaces/mutable_state_mock.go b/service/history/interfaces/mutable_state_mock.go index 7b12ad207dd..4d4478c7708 100644 --- a/service/history/interfaces/mutable_state_mock.go +++ b/service/history/interfaces/mutable_state_mock.go @@ -797,6 +797,20 @@ func (mr *MockMutableStateMockRecorder) AddWorkflowExecutionTimeSkippingTransiti return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddWorkflowExecutionTimeSkippingTransitionedEvent", reflect.TypeOf((*MockMutableState)(nil).AddWorkflowExecutionTimeSkippingTransitionedEvent), ctx, targetTime, disabledAfterFastForward) } +// RecordTimeSkippingTransition mocks base method. +func (m *MockMutableState) RecordTimeSkippingTransition(ctx context.Context, transition chasm.TimeSkippingTransition, archetype chasm.ArchetypeID) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "RecordTimeSkippingTransition", ctx, transition, archetype) + ret0, _ := ret[0].(error) + return ret0 +} + +// RecordTimeSkippingTransition indicates an expected call of RecordTimeSkippingTransition. +func (mr *MockMutableStateMockRecorder) RecordTimeSkippingTransition(ctx, transition, archetype any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecordTimeSkippingTransition", reflect.TypeOf((*MockMutableState)(nil).RecordTimeSkippingTransition), ctx, transition, archetype) +} + // AddWorkflowExecutionUnpausedEvent mocks base method. func (m *MockMutableState) AddWorkflowExecutionUnpausedEvent(identity, reason, requestID string) (*history.HistoryEvent, error) { m.ctrl.T.Helper() diff --git a/service/history/timer_queue_active_task_executor.go b/service/history/timer_queue_active_task_executor.go index dc07530e4fd..b96355e756c 100644 --- a/service/history/timer_queue_active_task_executor.go +++ b/service/history/timer_queue_active_task_executor.go @@ -3,7 +3,6 @@ package history import ( "context" "fmt" - "time" "github.com/google/uuid" commonpb "go.temporal.io/api/common/v1" @@ -937,9 +936,13 @@ func (t *timerQueueActiveTaskExecutor) executeTimeSkippingTimerTask( return errNoTimerFired } - _, err = mutableState.AddWorkflowExecutionTimeSkippingTransitionedEvent( - ctx, time.Time{}, true) - if err != nil { + // Disable time skipping through the archetype-aware sink: workflows record a history event; CHASM + // (and other archetypes) apply the disable-only transition directly to the TimeSkippingInfo. + if err = mutableState.RecordTimeSkippingTransition( + ctx, + chasm.TimeSkippingTransition{CurrentTime: mutableState.Now(), DisabledAfterFastForward: true}, + mutableState.ChasmTree().ArchetypeID(), + ); err != nil { return err } return t.updateWorkflowExecution(ctx, weContext, mutableState, false) diff --git a/service/history/workflow/mutable_state_impl.go b/service/history/workflow/mutable_state_impl.go index a5fa564e6fc..11038e8ac2f 100644 --- a/service/history/workflow/mutable_state_impl.go +++ b/service/history/workflow/mutable_state_impl.go @@ -7688,6 +7688,8 @@ func (ms *MutableStateImpl) closeTransaction( ms.chasmNodeSizes[nodePath] = newSize } + // todo@time-skipping: chasm close transaction logic is after isStateDirty, + // need to check if this impacts chasm time skipping logic if isStateDirty { if err := ms.closeTransactionUpdateTransitionHistory( transactionPolicy, @@ -8140,6 +8142,7 @@ func (ms *MutableStateImpl) closeTransactionPrepareTasks( if err := ms.closeTransactionHandleActivityUserTimerTasks(transactionPolicy); err != nil { return err } + // todo: this is only for workflows not for chasm-based executions if regenerateTimerTasksForTimeSkipping { if err := ms.closeTransactionRegenTimerTasksForWorkflowTimeSkipping(transactionPolicy); err != nil { return err diff --git a/service/history/workflow/timeskipping.go b/service/history/workflow/timeskipping.go index 2953d69cc02..c0d1e7e158d 100644 --- a/service/history/workflow/timeskipping.go +++ b/service/history/workflow/timeskipping.go @@ -9,6 +9,7 @@ import ( historypb "go.temporal.io/api/history/v1" "go.temporal.io/api/serviceerror" persistencespb "go.temporal.io/server/api/persistence/v1" + "go.temporal.io/server/chasm" "go.temporal.io/server/common" "go.temporal.io/server/common/clock" "go.temporal.io/server/common/log/tag" @@ -301,7 +302,7 @@ func (ms *MutableStateImpl) isWorkflowSkippable() bool { } // A pending activity blocks time skipping unless it has failed and is still // waiting out its retry backoff (next attempt strictly in the future) — that one - // is a skip target, not in-flight work (see calculateTimeSkippingTransition). The + // is a skip target, not in-flight work (see findNextSkipTarget). The // strict future check is what keeps a just-scheduled or already-due activity (next // attempt <= now) blocking. for _, ai := range ms.GetPendingActivityInfos() { @@ -393,6 +394,14 @@ func (ms *MutableStateImpl) closeTransactionHandleWorkflowTimeSkipping( ctx context.Context, transactionPolicy historyi.TransactionPolicy, ) (needRegenTasks bool) { + // This is the workflow (history-event) time-skipping path. CHASM executions (e.g. standalone + // activities) run their own decision in the CHASM tree's closeTransactionHandleTimeSkipping; + // running this path for them would scan only workflow-level targets (timers, pending activities, + // run timeout) — none of which exist for a CHASM component — and its GateByFastForward would then + // skip straight to the fast-forward and disable time skipping before the CHASM path ever runs. + if !ms.IsWorkflow() { + return false + } switch transactionPolicy { case historyi.TransactionPolicyActive: // 1. gate: only a running, time-skipping-enabled, idle workflow may skip time @@ -478,3 +487,127 @@ func (ms *MutableStateImpl) closeTransactionRegenTimerTasksForWorkflowTimeSkippi return serviceerror.NewInternalf("unknown transaction policy: %v", transactionPolicy) } } + +// ============================================================================= +// Time Skipping for CHASM-based Executions +// ============================================================================= + +// applyTimeSkippingTransition mutates the execution's TimeSkippingInfo for a single skip transition: +// it advances AccumulatedSkippedDuration by (TargetTime - CurrentTime), toggles Config.Enabled, and +// marks the fast-forward target reached. This is the CHASM (non-workflow) apply path, invoked by +// RecordTimeSkippingTransition; the workflow path applies the equivalent mutation from its history +// event in ApplyWorkflowExecutionTimeSkippingTransitionedEvent. +// +// A zero TargetTime means "no skip" (only valid together with DisabledAfterFastForward, e.g. a bound +// hit exactly). transition.CurrentTime must be in the virtual frame (ms.Now() / the event's virtual +// EventTime). +func (ms *MutableStateImpl) applyTimeSkippingTransition(transition chasm.TimeSkippingTransition) error { + opTag := tag.WorkflowActionWorkflowExecutionTimeSkippingTransitioned + invalidTransitionError := serviceerror.NewInternal("TimeSkippingTransitionedEvent failed to apply") + tsi := ms.executionInfo.GetTimeSkippingInfo() + if tsi == nil { + ms.logError("TimeSkippingTransitionedEvent failed to apply: TimeSkippingInfo is nil", opTag) + return invalidTransitionError + } + if transition.TargetTime.IsZero() && !transition.DisabledAfterFastForward { + ms.logError("TimeSkippingTransitionedEvent failed to apply: TargetTime is nil and disabled after fast-forward is false", opTag) + return invalidTransitionError + } + + if tsi.GetAccumulatedSkippedDuration() == nil { + tsi.AccumulatedSkippedDuration = durationpb.New(0) + } + accumulatedSkippedDuration := tsi.GetAccumulatedSkippedDuration().AsDuration() + if !transition.TargetTime.IsZero() { + accumulatedSkippedDuration += transition.TargetTime.Sub(transition.CurrentTime) + } + tsi.AccumulatedSkippedDuration = durationpb.New(accumulatedSkippedDuration) + tsi.Config.Enabled = !transition.DisabledAfterFastForward + + if transition.DisabledAfterFastForward && tsi.GetFastForwardInfo() != nil { + tsi.FastForwardInfo.HasReached = true + } + + ms.timeSkippingInfoUpdated = true + + // Re-stamp the CHASM fast-forward wake so its wall-clock fire time tracks the new accumulated skip + // (real fire = TargetTime - accumulatedSkippedDuration). Bundled with the accumulated update here + // rather than routed through the chasm task pipeline; no-op for workflows and once the target is + // reached. + ms.regenerateChasmFastForwardWakeTask() + return nil +} + +// chasmNoEventID is the sentinel event ID used when (re)computing time-skipping info for CHASM +// executions, which have no history events to anchor the configuration to. Real event IDs start at +// 1, so -1 is unambiguous. For CHASM the fast-forward wake is emitted as a CHASM task during +// CloseTransaction (anchored on the VersionedTransition), so this stub only ends up in the unused +// FastForwardInfo.SourceEventId field. +// +// TODO(time-skipping/chasm): emit the fast-forward wake as a ChasmTaskPure on the root component and +// anchor its staleness check on the VersionedTransition (as ChasmTaskInfo does) rather than an +// event ID. +const chasmNoEventID int64 = -1 + +// SetTimeSkippingConfig sets the execution's time-skipping configuration for a CHASM execution. It +// backs chasm.MutableContext.SetTimeSkippingConfig (via the chasm.NodeBackend interface): the first +// call initializes the TimeSkippingInfo, and subsequent calls update the config in place (preserving +// the accumulated skipped duration). It reuses the workflow path's initTimeSkippingInfo / +// updateTimeSkippingInfo, which are archetype-aware (no physical workflow timer task for CHASM — see +// applyFastForward). CHASM executions have no history events to anchor to (see chasmNoEventID) and no +// propagated state on this path. +func (ms *MutableStateImpl) SetTimeSkippingConfig(config *commonpb.TimeSkippingConfig) { + if ms.executionInfo.GetTimeSkippingInfo() == nil { + ms.initTimeSkippingInfo(config, nil, chasmNoEventID) + } else { + ms.updateTimeSkippingInfo(config, chasmNoEventID) + } +} + +// RecordTimeSkippingTransition records a single time-skipping transition for the execution. It is the +// archetype-aware apply sink: the CHASM framework's closeTransactionHandleTimeSkipping calls it for +// non-workflow executions after building the transition, and the fast-forward wake timer executor +// calls it (for either archetype) to disable time skipping once the fast-forward target is hit. The +// caller is responsible for computing the final transition — including the min of the component +// candidate and the fast-forward target; this method only records it. +// +// archetype selects how the skip is recorded: +// - the workflow archetype records a history event (AddWorkflowExecutionTimeSkippingTransitionedEvent), +// which in turn applies the transition to the TimeSkippingInfo; +// - all other archetypes have no history, so the transition is applied directly to the TSI. +// +// For CHASM executions, after this returns the physical timer tasks regenerated later in the same +// CloseTransaction pick up the new accumulated offset via AddTasks -> ToRealTime, so no separate +// timer-task regeneration is needed. +func (ms *MutableStateImpl) RecordTimeSkippingTransition( + ctx context.Context, + transition chasm.TimeSkippingTransition, + archetype chasm.ArchetypeID, +) error { + if !transition.IsValid() { + return nil + } + switch archetype { + case chasm.WorkflowArchetypeID: + // workflows require applying the transition through a history event + _, err := ms.AddWorkflowExecutionTimeSkippingTransitionedEvent( + ctx, transition.TargetTime, transition.DisabledAfterFastForward) + return err + default: + return ms.applyTimeSkippingTransition(transition) + } +} +func (ms *MutableStateImpl) regenerateChasmFastForwardWakeTask() { + if ms.IsWorkflow() { + return + } + ff := ms.executionInfo.GetTimeSkippingInfo().GetFastForwardInfo() + if ff == nil || ff.GetHasReached() || ff.GetTargetTime() == nil { + return + } + ms.AddTasks(&tasks.TimeSkippingTimerTask{ + WorkflowKey: ms.GetWorkflowKey(), + VisibilityTimestamp: ff.GetTargetTime().AsTime(), // virtual; AddTasks -> ToRealTime shifts to wall clock + EventID: chasmNoEventID, + }) +} diff --git a/service/history/workflow/timeskipping_test.go b/service/history/workflow/timeskipping_test.go index fb438df8abe..17b5e7fe9b9 100644 --- a/service/history/workflow/timeskipping_test.go +++ b/service/history/workflow/timeskipping_test.go @@ -11,10 +11,12 @@ import ( historypb "go.temporal.io/api/history/v1" enumsspb "go.temporal.io/server/api/enums/v1" persistencespb "go.temporal.io/server/api/persistence/v1" + "go.temporal.io/server/chasm" "go.temporal.io/server/common" "go.temporal.io/server/common/clock" "go.temporal.io/server/components/nexusoperations" historyi "go.temporal.io/server/service/history/interfaces" + "go.temporal.io/server/service/history/tasks" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" @@ -143,6 +145,54 @@ func (s *mutableStateSuite) TestPropagateTimeSkippingToNextRun_FastForwardInfo() }) } +// TestChasmFastForwardWake_ShortenedBySkip verifies that, for a CHASM (non-workflow) execution, the +// fast-forward wake timer is kept in mutable state and its wall-clock fire time is shortened by each +// skip: init emits the wake at the fast-forward target (no skip yet); a 10m skip re-emits it 10m +// sooner (real fire = target - accumulatedSkippedDuration). +func (s *mutableStateSuite) TestChasmFastForwardWake_ShortenedBySkip() { + // Force a CHASM (non-workflow) archetype so the wake stays in mutable state and is re-stamped here + // (rather than via the workflow event path). + mockChasmTree := historyi.NewMockChasmTree(s.controller) + s.mutableState.chasmTree = mockChasmTree + mockChasmTree.EXPECT().ArchetypeID().Return(chasm.WorkflowArchetypeID + 1).AnyTimes() + + // Init the fast-forward — emits the wake at the target (accumulated == 0, so real fire == target). + s.mutableState.SetTimeSkippingConfig(&commonpb.TimeSkippingConfig{ + Enabled: true, + FastForward: durationpb.New(time.Hour), + }) + ffTarget := s.mutableState.executionInfo.GetTimeSkippingInfo().GetFastForwardInfo().GetTargetTime().AsTime() + + initialWake := s.lastTimeSkippingTimerTask() + s.Require().NotNil(initialWake) + s.Equal(ffTarget, initialWake.GetVisibilityTime(), "initial wake fires at the fast-forward target") + + // Skip 10m of virtual time. + const skipped = 10 * time.Minute + base := s.mutableState.timeSource.Now() + s.NoError(s.mutableState.applyTimeSkippingTransition(chasm.TimeSkippingTransition{ + CurrentTime: base, + TargetTime: base.Add(skipped), + })) + s.Equal(skipped, s.mutableState.executionInfo.GetTimeSkippingInfo().GetAccumulatedSkippedDuration().AsDuration()) + + // The wake is re-stamped 10m sooner — the fast-forward timer is shortened by the skip. + skippedWake := s.lastTimeSkippingTimerTask() + s.Require().NotNil(skippedWake) + s.Equal(ffTarget.Add(-skipped), skippedWake.GetVisibilityTime(), "wake shortened by the skipped duration") + s.Equal(skipped, initialWake.GetVisibilityTime().Sub(skippedWake.GetVisibilityTime())) +} + +func (s *mutableStateSuite) lastTimeSkippingTimerTask() *tasks.TimeSkippingTimerTask { + var last *tasks.TimeSkippingTimerTask + for _, t := range s.mutableState.InsertTasks[tasks.CategoryTimer] { + if tst, ok := t.(*tasks.TimeSkippingTimerTask); ok { + last = tst + } + } + return last +} + func (s *mutableStateSuite) TestSnapshotTimeSkippingInfo_ForChildWorkflows() { newSource := func() *persistencespb.WorkflowExecutionInfo { s.mutableState.timeSource = clock.NewEventTimeSource() diff --git a/tests/activity_standalone_test.go b/tests/activity_standalone_test.go index 7d893028672..97693f100f5 100644 --- a/tests/activity_standalone_test.go +++ b/tests/activity_standalone_test.go @@ -23,6 +23,7 @@ import ( "go.temporal.io/sdk/temporal" "go.temporal.io/server/chasm/lib/activity" "go.temporal.io/server/chasm/lib/callback" + "go.temporal.io/server/common" "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/headers" "go.temporal.io/server/common/log" @@ -30,6 +31,7 @@ import ( "go.temporal.io/server/common/nexus/nexusrpc" "go.temporal.io/server/common/payload" "go.temporal.io/server/common/payloads" + "go.temporal.io/server/common/persistence" "go.temporal.io/server/common/searchattribute/sadefs" "go.temporal.io/server/common/tasktoken" "go.temporal.io/server/common/testing/parallelsuite" @@ -116,6 +118,137 @@ func (s *standaloneActivityTestSuite) newTestEnv(opts ...testcore.TestOption) *s return env } +// getMutableState loads the persisted mutable state of a standalone activity execution, so tests can +// inspect server-internal state such as the time-skipping info. +func (env *standaloneActivityEnv) getMutableState(t *testing.T, activityID, runID string) *persistence.GetWorkflowExecutionResponse { + shardID := common.WorkflowIDToHistoryShard( + env.NamespaceID().String(), + activityID, + env.GetTestClusterConfig().HistoryConfig.NumHistoryShards, + ) + ms, err := env.GetTestCluster().ExecutionManager().GetWorkflowExecution(testcore.NewContext(), &persistence.GetWorkflowExecutionRequest{ + ShardID: shardID, + NamespaceID: env.NamespaceID().String(), + WorkflowID: activityID, + RunID: runID, + ArchetypeID: activity.ArchetypeID, + }) + require.NoError(t, err) + return ms +} + +// TestTimeSkipping_StartDelayAndRetryBackoff verifies that a standalone activity, which opts into +// time skipping, fast-forwards both its start delay and its retry backoff while idle. With a 1h +// fast-forward budget, a 20m start delay plus a 20m retry backoff (~40m of virtual time) are skipped, +// so the run completes in seconds of wall clock and the accumulated skipped duration reflects the +// skipped delays. +func (s *standaloneActivityTestSuite) TestTimeSkipping_StartDelayAndRetryBackoff() { + env := s.newTestEnv() + t := s.T() + + // Standalone activities only opt into time skipping when the SAA-specific namespace flag is on. + env.GetTestCluster().OverrideDynamicConfig(t, activity.TimeSkippingEnabled, []dynamicconfig.ConstrainedValue{ + {Constraints: dynamicconfig.Constraints{Namespace: env.Namespace().String()}, Value: true}, + }) + + activityID := testcore.RandomizeStr(t.Name()) + taskQueue := testcore.RandomizeStr(t.Name()) + + const startDelay = 20 * time.Minute + const retryBackoff = 20 * time.Minute + + wallStart := time.Now() + + startResp, err := env.FrontendClient().StartActivityExecution(s.Context(), &workflowservice.StartActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + ActivityType: env.Tv().ActivityType(), + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue}, + Input: defaultInput, + StartToCloseTimeout: durationpb.New(time.Minute), + ScheduleToCloseTimeout: durationpb.New(2 * time.Hour), // comfortably exceeds delay + backoff + StartDelay: durationpb.New(startDelay), + RetryPolicy: &commonpb.RetryPolicy{ + InitialInterval: durationpb.New(retryBackoff), + BackoffCoefficient: 1.0, + MaximumAttempts: 2, + }, + // The 1h FastForward budget never binds here (both skip targets — the 20m start delay and + // the 20m retry backoff — are well under 1h, so time skipping advances to them, not to the + // budget). It is kept deliberately as a regression guard: it is precisely what surfaces the + // bug where the workflow (history-event) time-skipping path runs for a CHASM execution and, + // finding no workflow-level target, skips straight to the fast-forward and disables time + // skipping before the CHASM path runs (fixed by the !ms.IsWorkflow() gate in + // closeTransactionHandleWorkflowTimeSkipping). Without a FastForward that path has nothing to + // skip to and the regression would not reproduce. + TimeSkippingConfig: &commonpb.TimeSkippingConfig{ + Enabled: true, + FastForward: durationpb.New(time.Hour), + }, + }) + require.NoError(t, err) + runID := startResp.RunId + + pollTaskQueue := func() *workflowservice.PollActivityTaskQueueResponse { + resp, pollErr := env.FrontendClient().PollActivityTaskQueue(s.Context(), &workflowservice.PollActivityTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, + }) + require.NoError(t, pollErr) + return resp + } + + // Attempt 1's dispatch is deferred 20m by the start delay. Time skipping fast-forwards virtual + // time to the dispatch, so the task becomes pollable within seconds of wall clock rather than + // after 20m. (Without skipping this poll would block out its long-poll window and return empty.) + attempt1 := pollTaskQueue() + require.EqualValues(t, 1, attempt1.Attempt) + + // Fail attempt 1 retryably; the 20m retry backoff is likewise fast-forwarded. + _, err = env.FrontendClient().RespondActivityTaskFailed(s.Context(), &workflowservice.RespondActivityTaskFailedRequest{ + Namespace: env.Namespace().String(), + TaskToken: attempt1.TaskToken, + Failure: &failurepb.Failure{ + Message: "retryable failure", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{NonRetryable: false}}, + }, + }) + require.NoError(t, err) + + attempt2 := pollTaskQueue() + require.EqualValues(t, 2, attempt2.Attempt) + + _, err = env.FrontendClient().RespondActivityTaskCompleted(s.Context(), &workflowservice.RespondActivityTaskCompletedRequest{ + Namespace: env.Namespace().String(), + TaskToken: attempt2.TaskToken, + Result: defaultResult, + }) + require.NoError(t, err) + + wallElapsed := time.Since(wallStart) + + // The run spanned ~40m of virtual time (start delay + retry backoff) but finished in seconds of + // wall clock — proof that both delays were skipped rather than waited out. + require.Less(t, wallElapsed, startDelay, "run should finish in well under the virtual start delay; took %v", wallElapsed) + + // The activity completed on its second attempt. + desc, err := env.FrontendClient().DescribeActivityExecution(s.Context(), &workflowservice.DescribeActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: runID, + }) + require.NoError(t, err) + require.Equal(t, enumspb.ACTIVITY_EXECUTION_STATUS_COMPLETED, desc.GetInfo().GetStatus()) + require.EqualValues(t, 2, desc.GetInfo().GetAttempt()) + + // The accumulated skipped duration reflects the skipped start delay + retry backoff (~40m). Time + // skipping then ends because the activity is closed (no further pending timers to skip to). + ms := env.getMutableState(t, activityID, runID) + accumulated := ms.State.GetExecutionInfo().GetTimeSkippingInfo().GetAccumulatedSkippedDuration().AsDuration() + require.InDelta(t, float64(startDelay+retryBackoff), float64(accumulated), float64(time.Minute), + "accumulated skipped duration %v should be ~%v", accumulated, startDelay+retryBackoff) +} + func (s *standaloneActivityTestSuite) TestIDReusePolicy() { env := s.newTestEnv() t := s.T()