add activity ts feature flag

Author: feiyang3cat

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 {
@@ -69,7 +75,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),
+		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/tree.go

@@ -1730,6 +1730,8 @@ 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
 	}
@@ -2197,29 +2199,6 @@ func (n *Node) closeTransactionHandleTimeSkipping() error {
 	return nil
 }
 
-// validateSerializedTask deserializes a serialized component task on this node and runs its validator
-// against current component state, returning true only if the task is still valid. It mirrors the
-// per-task validation in closeTransactionCleanupInvalidTasks and is used by the time-skipping paths
-// (calculateTimeSkippingTransition, regenerateTasksForTimeSkipping) so that an invalidated task — e.g. a
-// timeout whose activity already executed — is neither chosen as a skip target nor re-stamped. The component
-// value is hydrated first (prepareComponentValue is idempotent).
-func (n *Node) validateSerializedTask(
-	ctx Context,
-	task *persistencespb.ChasmComponentAttributes_Task,
-) (bool, error) {
-	if err := n.prepareComponentValue(ctx); err != nil {
-		return false, err
-	}
-	taskInstance, err := n.deserializeComponentTask(task)
-	if err != nil {
-		return false, err
-	}
-	return n.validateTask(ctx, TaskAttributes{
-		ScheduledTime: task.GetScheduledTime().AsTime(),
-		Destination:   task.GetDestination(),
-	}, taskInstance)
-}
-
 func (n *Node) defaultFindNextTargetTime(transition *TimeSkippingTransition) error {
 	now := transition.CurrentTime
 	for _, node := range n.andAllChildren() {
@@ -2246,16 +2225,6 @@ func (n *Node) defaultFindNextTargetTime(transition *TimeSkippingTransition) err
 	return nil
 }
 
-// regenerateChasmTasksForTimeSkipping re-emits physical timer tasks after a time-skipping transition
-// so their wall-clock visibility reflects the new accumulated skip.
-//
-// "Regenerate all" by default: every CategoryTimer task (pure and side-effect) across the tree is
-// reset to physicalTaskStatusNone and re-generated. BOTH pure and side-effect tasks can be
-// future-scheduled (CategoryTimer) tasks that need re-stamping.Immediate Transfer/Outbound/Visibility
-// tasks are left untouched.
-//
-// Invalidated tasks are excluded: each task is validated (validateSerializedTask) before being reset or
-// re-stamped, so a task whose validator now rejects it is left alone rather than re-emitted.
 func (n *Node) regenerateChasmTasksForTimeSkipping(ctx Context) error {
 	archetypeID := n.ArchetypeID()
 
@@ -2272,25 +2241,11 @@ func (n *Node) regenerateChasmTasksForTimeSkipping(ctx Context) error {
 			if taskCategory(sideEffectTask) != tasks.CategoryTimer {
 				continue
 			}
-			valid, err := node.validateSerializedTask(ctx, sideEffectTask)
-			if err != nil {
-				return err
-			}
-			if !valid {
-				continue
-			}
 			sideEffectTask.PhysicalTaskStatus = physicalTaskStatusNone
 			node.closeTransactionGeneratePhysicalSideEffectTask(sideEffectTask, nodePath, archetypeID)
 		}
 
 		for _, pureTask := range componentAttr.GetPureTasks() {
-			valid, err := node.validateSerializedTask(ctx, pureTask)
-			if err != nil {
-				return err
-			}
-			if !valid {
-				continue
-			}
 			pureTask.PhysicalTaskStatus = physicalTaskStatusNone
 			if firstPureTask == nil || comparePureTasks(pureTask, firstPureTask) < 0 {
 				firstPureTask = pureTask

service/history/workflow/timeskipping.go

@@ -302,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() {
@@ -486,10 +486,9 @@ func (ms *MutableStateImpl) closeTransactionRegenTimerTasksForWorkflowTimeSkippi
 
 // 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. It is the archetype-agnostic core shared by the workflow
-// event path (ApplyWorkflowExecutionTimeSkippingTransitionedEvent) and the CHASM close-transaction
-// path (RecordTimeSkippingTransition); the workflow path records the transition as a history event while
-// CHASM (which has no history) applies it directly, but the TSI mutation is identical.
+// 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
@@ -558,9 +557,9 @@ func (ms *MutableStateImpl) SetTimeSkippingConfig(config *commonpb.TimeSkippingC
 }
 
 // RecordTimeSkippingTransition records a single time-skipping transition for the execution. It is the
-// archetype-agnostic apply sink shared by both time-skipping front-ends: the CHASM framework's
-// closeTransactionHandleTimeSkipping (which builds the transition for non-workflow CHASM executions)
-// and the workflow event path (closeTransactionHandleTimeSkipping -> shouldExecuteTimeSkipping). 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.
 //

tests/activity_standalone_test.go

@@ -146,8 +146,8 @@ func (s *standaloneActivityTestSuite) TestTimeSkipping_StartDelayAndRetryBackoff
 	env := s.newTestEnv()
 	t := s.T()
 
-	// Standalone activities only opt into time skipping when the namespace flag is on.
-	env.GetTestCluster().OverrideDynamicConfig(t, dynamicconfig.TimeSkippingEnabled, []dynamicconfig.ConstrainedValue{
+	// 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},
 	})