[don't merge] Time-skipping POC for CHASM with SAA

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),

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) {

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 != "" {

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),
 	}

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 {

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

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()

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())
+}