timeskipping poc with saa

Author: feiyang3cat

chasm/context.go

@@ -97,6 +97,12 @@ type MutableContext interface {
 	// SetUserMetadata replaces the user metadata attached to the given component.
 	SetUserMetadata(Component, *sdkpb.UserMetadata) error
 
+	// TimeSkippingController is the framework-provided surface for managing this execution's
+	// time-skipping configuration (Init/Update/Get). A root component opts in by calling
+	// InitTimeSkippingConfig when the execution is created. These take no Component argument because
+	// the configuration is execution-scoped, like the ExecutionKey()/ExecutionInfo() accessors.
+	TimeSkippingController
+
 	// 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,
@@ -272,6 +278,18 @@ func (c *mutableCtx) SetUserMetadata(component Component, md *sdkpb.UserMetadata
 	return c.root.setComponentUserMetadata(component, md)
 }
 
+func (c *mutableCtx) InitTimeSkippingConfig(config *commonpb.TimeSkippingConfig) {
+	c.root.backend.InitTimeSkippingConfig(config)
+}
+
+func (c *mutableCtx) UpdateTimeSkippingConfig(config *commonpb.TimeSkippingConfig) {
+	c.root.backend.UpdateTimeSkippingConfig(config)
+}
+
+func (c *mutableCtx) GetTimeSkippingConfig() *commonpb.TimeSkippingConfig {
+	return c.root.backend.GetTimeSkippingInfo().GetConfig()
+}
+
 func (c *mutableCtx) withValue(key any, value any) Context {
 	return &mutableCtx{
 		immutableCtx: ContextWithValue(c.immutableCtx, key, value),

chasm/context_mock.go

@@ -183,10 +183,31 @@ func (c *MockContext) withValue(key any, value any) Context {
 type MockMutableContext struct {
 	MockContext
 
-	mu                      sync.Mutex
-	Tasks                   []MockTask
-	LinksByRequest          map[Component]map[string][]*commonpb.Link
-	UserMetadataByComponent map[Component]*sdkpb.UserMetadata
+	mu                       sync.Mutex
+	Tasks                    []MockTask
+	LinksByRequest           map[Component]map[string][]*commonpb.Link
+	UserMetadataByComponent  map[Component]*sdkpb.UserMetadata
+	TimeSkippingConfig       *commonpb.TimeSkippingConfig
+	TimeSkippingConfigInited bool
+}
+
+func (c *MockMutableContext) InitTimeSkippingConfig(config *commonpb.TimeSkippingConfig) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.TimeSkippingConfig = config
+	c.TimeSkippingConfigInited = true
+}
+
+func (c *MockMutableContext) UpdateTimeSkippingConfig(config *commonpb.TimeSkippingConfig) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.TimeSkippingConfig = config
+}
+
+func (c *MockMutableContext) GetTimeSkippingConfig() *commonpb.TimeSkippingConfig {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.TimeSkippingConfig
 }
 
 func (c *MockMutableContext) AddTask(component Component, attributes TaskAttributes, payload any) {

chasm/lib/activity/activity.go

@@ -124,6 +124,53 @@ func (a *Activity) LifecycleState(_ chasm.Context) chasm.LifecycleState {
 	}
 }
 
+// HasInflightWork implements chasm.TimeSkippable. It reports whether advancing virtual time would
+// skip past real work, gating the framework's per-transaction time-skipping decision.
+//
+// Only the activity attempt is considered: it is in flight unless the activity is SCHEDULED and
+// still waiting for its next dispatch — out of a start delay or retry backoff with the next-attempt
+// time strictly in the (virtual) future. Once that time arrives (dispatch pushed to matching,
+// waiting for a poller), the attempt is running (STARTED) or being cancelled (CANCEL_REQUESTED), or
+// the activity has closed, this returns true. This mirrors the workflow time-skipping rule for
+// activities, and means a closed activity is never time-skipped.
+func (a *Activity) HasInflightWork(ctx chasm.Context) bool {
+	if a.GetStatus() != activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED {
+		return true
+	}
+	nextDispatch := a.attemptScheduleTime(a.LastAttempt.Get(ctx))
+	if nextDispatch == nil || !ctx.Now(a).Before(nextDispatch.AsTime()) {
+		return true
+	}
+	return false
+}
+
+// FindNextTargetTime implements chasm.TimeSkippable. The framework calls it after HasInflightWork has
+// reported the activity is idle; the activity contributes its candidate virtual time by calling
+// targetTime, leaving it untouched if it has nothing of its own to skip to. Today that is only the next
+// attempt dispatch (start-delay end on the first attempt, retry-backoff end on a retry) while the
+// activity is SCHEDULED-and-waiting, gated by the schedule-to-close deadline (the execution timeout) so
+// we never skip past it.
+//
+// The framework pre-seeds targetTime with the configured fast-forward target, so the activity only
+// contributes its own next-wake time; CompareAndSet keeps the minimum and the framework decides whether
+// the fast-forward budget was reached.
+//
+// TODO(time-skipping): completion callback retry backoff
+func (a *Activity) FindNextTargetTime(
+	ctx chasm.Context,
+	targetTime *chasm.TimeSkippingTargetTime,
+) {
+	nextDispatch := a.attemptScheduleTime(a.LastAttempt.Get(ctx))
+	if nextDispatch == nil || !ctx.Now(a).Before(nextDispatch.AsTime()) {
+		return
+	}
+	targetTime.CompareAndSet(nextDispatch.AsTime())
+	// Gate by the schedule-to-close deadline (execution timeout) so we never skip past it.
+	if deadline := a.scheduleToCloseDeadline(); !deadline.IsZero() {
+		targetTime.GateByExecutionTimeout(deadline)
+	}
+}
+
 func (a *Activity) ContextMetadata(_ chasm.Context) map[string]string {
 	md := make(map[string]string, 2)
 	if actType := a.GetActivityType().GetName(); actType != "" {

chasm/lib/activity/config.go

@@ -53,6 +53,7 @@ type Config struct {
 	MaxCallbacksPerExecution    dynamicconfig.IntPropertyFnWithNamespaceFilter
 	DefaultActivityRetryPolicy  dynamicconfig.TypedPropertyFnWithNamespaceFilter[retrypolicy.DefaultRetrySettings]
 	StartDelayEnabled           dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	TimeSkippingEnabled         dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	VisibilityMaxPageSize       dynamicconfig.IntPropertyFnWithNamespaceFilter
 }
 
@@ -68,6 +69,7 @@ func ConfigProvider(dc *dynamicconfig.Collection) *Config {
 		LongPollTimeout:             LongPollTimeout.Get(dc),
 		MaxIDLengthLimit:            dynamicconfig.MaxIDLengthLimit.Get(dc),
 		StartDelayEnabled:           StartDelayEnabled.Get(dc),
+		TimeSkippingEnabled:         dynamicconfig.TimeSkippingEnabled.Get(dc),
 		MaxCallbacksPerExecution:    callback.MaxPerExecution.Get(dc),
 		VisibilityMaxPageSize:       dynamicconfig.FrontendVisibilityMaxPageSize.Get(dc),
 	}

chasm/lib/activity/handler.go

@@ -95,6 +95,10 @@ func (h *handler) StartActivityExecution(ctx context.Context, req *activitypb.St
 				}
 			}
 
+			if request.GetTimeSkippingConfig() != nil {
+				mutableContext.InitTimeSkippingConfig(request.GetTimeSkippingConfig())
+			}
+
 			err = TransitionScheduled.Apply(newActivity, mutableContext, nil)
 			if err != nil {
 				return nil, err

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,27 @@ 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
+	HandleInitTimeSkippingConfig       func(config *commonpb.TimeSkippingConfig)
+	HandleUpdateTimeSkippingConfig     func(config *commonpb.TimeSkippingConfig)
+	HandleGetTimeSkippingInfo          func() *persistencespb.TimeSkippingInfo
+	HandleRecordTimeSkippingTransition func(ctx context.Context, transition TimeSkippingTransition, archetype ArchetypeID) error
 
 	// Recorded calls (protected by mu).
 	mu                  sync.Mutex
@@ -111,6 +116,32 @@ func (m *MockNodeBackend) AddTasks(ts ...tasks.Task) {
 	}
 }
 
+func (m *MockNodeBackend) InitTimeSkippingConfig(config *commonpb.TimeSkippingConfig) {
+	if m.HandleInitTimeSkippingConfig != nil {
+		m.HandleInitTimeSkippingConfig(config)
+	}
+}
+
+func (m *MockNodeBackend) UpdateTimeSkippingConfig(config *commonpb.TimeSkippingConfig) {
+	if m.HandleUpdateTimeSkippingConfig != nil {
+		m.HandleUpdateTimeSkippingConfig(config)
+	}
+}
+
+func (m *MockNodeBackend) GetTimeSkippingInfo() *persistencespb.TimeSkippingInfo {
+	if m.HandleGetTimeSkippingInfo != nil {
+		return m.HandleGetTimeSkippingInfo()
+	}
+	return nil
+}
+
+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()

chasm/timeskipping_test.go

@@ -0,0 +1,103 @@
+package chasm
+
+import (
+	"time"
+
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/service/history/tasks"
+	"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)
+
+	err = root.regenerateTasksForTimeSkipping()
+	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 GetTimeSkippingInfo returns 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())
+}