diff --git a/chasm/chasmtest/test_engine.go b/chasm/chasmtest/test_engine.go index d45ff69f0b6..53bd196608f 100644 --- a/chasm/chasmtest/test_engine.go +++ b/chasm/chasmtest/test_engine.go @@ -40,6 +40,10 @@ type ( // allExecutions maps (namespaceID, businessID, runID) to any run, for lookups by specific RunID. allExecutions map[runKey]*execution notifier *executionNotifier + // decorateBackend, if set, is applied to each execution's MockNodeBackend at + // creation time, before the tree is built. It lets callers supply handlers the + // default backend leaves unset (e.g. HandleGetNamespaceEntry). + decorateBackend func(*chasm.MockNodeBackend) } execution struct { @@ -73,6 +77,17 @@ func WithTimeSource(ts clock.TimeSource) EngineOption { } } +// WithNodeBackendDecorator registers a function applied to every execution's +// [chasm.MockNodeBackend] at creation time, before its tree is built. Use it to +// supply backend handlers the default leaves unset — most commonly +// HandleGetNamespaceEntry, which components that read namespace-scoped config +// (e.g. the scheduler's Tweakables) require. +func WithNodeBackendDecorator(fn func(*chasm.MockNodeBackend)) EngineOption { + return func(e *Engine) { + e.decorateBackend = fn + } +} + var defaultTransitionOptions = chasm.TransitionOptions{ ReusePolicy: chasm.BusinessIDReusePolicyAllowDuplicate, ConflictPolicy: chasm.BusinessIDConflictPolicyFail, @@ -558,6 +573,9 @@ func (e *Engine) newExecution(key chasm.ExecutionKey) *execution { return changed, nil }, } + if e.decorateBackend != nil { + e.decorateBackend(backend) + } return &execution{ key: key, backend: backend, @@ -572,6 +590,29 @@ func (e *Engine) newExecution(key chasm.ExecutionKey) *execution { } } +// NodeForRef returns the CHASM tree node backing the execution identified by ref. +// It is intended for test drivers that need to invoke node-level task dispatch +// ([chasm.Node.EachPureTask], [chasm.Node.ExecuteSideEffectTask]) or close +// transactions directly, which the [Engine] API does not expose. +func (e *Engine) NodeForRef(ref chasm.ComponentRef) (*chasm.Node, error) { + exec, err := e.executionForRef(ref) + if err != nil { + return nil, err + } + return exec.node, nil +} + +// BackendForRef returns the [chasm.MockNodeBackend] backing the execution +// identified by ref. Test drivers use it to inspect the physical tasks +// ([chasm.MockNodeBackend.TasksByCategory]) the engine has accumulated. +func (e *Engine) BackendForRef(ref chasm.ComponentRef) (*chasm.MockNodeBackend, error) { + exec, err := e.executionForRef(ref) + if err != nil { + return nil, err + } + return exec.backend, nil +} + // executionForRef looks up an execution by the ref's RunID when present, or falls back // to the current run for the business ID when RunID is empty. func (e *Engine) executionForRef(ref chasm.ComponentRef) (*execution, error) { diff --git a/chasm/lib/scheduler/schedulertest/driver.go b/chasm/lib/scheduler/schedulertest/driver.go new file mode 100644 index 00000000000..210ad2fb683 --- /dev/null +++ b/chasm/lib/scheduler/schedulertest/driver.go @@ -0,0 +1,522 @@ +package schedulertest + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + enumspb "go.temporal.io/api/enums/v1" + schedulepb "go.temporal.io/api/schedule/v1" + "go.temporal.io/api/serviceerror" + workflowpb "go.temporal.io/api/workflow/v1" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/server/api/historyservice/v1" + "go.temporal.io/server/api/historyservicemock/v1" + persistencespb "go.temporal.io/server/api/persistence/v1" + "go.temporal.io/server/chasm" + "go.temporal.io/server/chasm/chasmtest" + "go.temporal.io/server/chasm/lib/scheduler" + schedulerpb "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1" + "go.temporal.io/server/common/clock" + "go.temporal.io/server/common/log" + "go.temporal.io/server/common/testing/mockapi/workflowservicemock/v1" + "go.temporal.io/server/common/testing/testlogger" + "go.temporal.io/server/common/testing/testvars" + "go.temporal.io/server/service/history/tasks" + "go.uber.org/mock/gomock" + "google.golang.org/grpc" + "google.golang.org/protobuf/types/known/timestamppb" +) + +// maxKeyFireTime is the sentinel returned by MockNodeBackend.LastDeletePureTaskCall +// when the tree has no pending pure task. It matches the value passed by +// Node.closeTransactionGeneratePhysicalPureTask in that case. +var maxKeyFireTime = tasks.MaximumKey.FireTime + +// Driver builds a real CHASM scheduler on top of a [chasmtest.Engine] and steps +// it forward through every task it schedules for itself, advancing a controllable +// clock. After each settled transition it invokes AfterStep (if set), which is +// where invariant assertions are plugged in. +// +// A "step" fires every pure task due at the current reference time (via +// [chasm.Node.EachPureTask]) and every side-effect task whose visibility time has +// arrived (via [chasm.Node.ExecuteSideEffectTask]), then advances the clock to the +// next scheduled task. A healthy interval schedule never truly quiesces (it always +// re-arms a generator task), so callers bound the run with maxSteps; quiescence is +// reached only for paused/exhausted/closed schedules. +type Driver struct { + T *testing.T + Engine *chasmtest.Engine + Registry *chasm.Registry + TimeSource *clock.EventTimeSource + Logger log.Logger + + Frontend *workflowservicemock.MockWorkflowServiceClient + History *historyservicemock.MockHistoryServiceClient + + ref chasm.ComponentRef + execKey chasm.ExecutionKey + + // dispatchedSideEffects tracks physical side-effect tasks already executed, + // keyed by pointer identity, since MockNodeBackend accumulates tasks append-only. + dispatchedSideEffects map[*tasks.ChasmTask]struct{} + + // AfterStep, if set, is called after every step's CloseTransaction with the + // driver, so invariant checks can observe each settled state. + AfterStep func(d *Driver) +} + +type driverConfig struct { + schedule *schedulepb.Schedule + startTime time.Time + afterStep func(d *Driver) +} + +// DriverOption configures a Driver at construction. +type DriverOption func(*driverConfig) + +// WithSchedule overrides the schedule definition (default: DefaultSchedule). +func WithSchedule(s *schedulepb.Schedule) DriverOption { + return func(c *driverConfig) { c.schedule = s } +} + +// WithStartTime sets the wall-clock instant the engine's clock starts at. +func WithStartTime(t time.Time) DriverOption { + return func(c *driverConfig) { c.startTime = t } +} + +// WithAfterStep installs a hook called after every step. +func WithAfterStep(fn func(d *Driver)) DriverOption { + return func(c *driverConfig) { c.afterStep = fn } +} + +// NewDriver constructs a Driver and starts a scheduler execution via the +// production CreateScheduler path. startTime defaults to a fixed instant for +// determinism. +func NewDriver(t *testing.T, opts ...DriverOption) *Driver { + t.Helper() + d, cfg := newDriverScaffold(t, opts...) + + req := &schedulerpb.CreateScheduleRequest{ + NamespaceId: NamespaceID, + FrontendRequest: &workflowservice.CreateScheduleRequest{ + Namespace: Namespace, + ScheduleId: ScheduleID, + Schedule: cfg.schedule, + RequestId: "create-req", + }, + } + d.startExecution(func(mctx chasm.MutableContext) (*scheduler.Scheduler, error) { + return scheduler.CreateScheduler(mctx, req) + }) + return d +} + +// NewDriverFromMigration constructs a Driver and starts a scheduler execution +// from a migrated V1 state via the production CreateSchedulerFromMigration path. +// This is the V1->V2 import side of a migration round trip. +func NewDriverFromMigration(t *testing.T, req *schedulerpb.CreateFromMigrationStateRequest, opts ...DriverOption) *Driver { + t.Helper() + d, _ := newDriverScaffold(t, opts...) + d.startExecution(func(mctx chasm.MutableContext) (*scheduler.Scheduler, error) { + return scheduler.CreateSchedulerFromMigration(mctx, req) + }) + return d +} + +// newDriverScaffold builds the engine, registry, clients, and time source, but +// does not start an execution. +func newDriverScaffold(t *testing.T, opts ...DriverOption) (*Driver, *driverConfig) { + t.Helper() + + cfg := &driverConfig{ + schedule: DefaultSchedule(), + startTime: time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC), + } + for _, opt := range opts { + opt(cfg) + } + + ctrl := gomock.NewController(t) + logger := testlogger.NewTestLogger(t, testlogger.FailOnExpectedErrorOnly) + + ts := clock.NewEventTimeSource() + ts.Update(cfg.startTime) + + frontend := workflowservicemock.NewMockWorkflowServiceClient(ctrl) + history := historyservicemock.NewMockHistoryServiceClient(ctrl) + // Default happy-path stubs; tests can layer more specific expectations. + frontend.EXPECT().StartWorkflowExecution(gomock.Any(), gomock.Any()). + Return(&workflowservice.StartWorkflowExecutionResponse{RunId: "run-id"}, nil).AnyTimes() + // The callbacks task (used for migrated running workflows) describes each + // running workflow; report it completed so watchers resolve deterministically. + history.EXPECT().DescribeWorkflowExecution(gomock.Any(), gomock.Any()). + DoAndReturn(func(_ context.Context, _ *historyservice.DescribeWorkflowExecutionRequest, _ ...grpc.CallOption) (*historyservice.DescribeWorkflowExecutionResponse, error) { + return &historyservice.DescribeWorkflowExecutionResponse{ + WorkflowExecutionInfo: &workflowpb.WorkflowExecutionInfo{ + Status: enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, + CloseTime: timestamppb.New(ts.Now()), + }, + }, nil + }).AnyTimes() + + specProcessor := NewRealSpecProcessor(ctrl, logger) + + registry := chasm.NewRegistry(logger) + if err := registry.Register(&chasm.CoreLibrary{}); err != nil { + t.Fatalf("failed to register core library: %v", err) + } + if err := registry.Register(NewTestLibrary(DefaultConfig(), logger, specProcessor, frontend, history)); err != nil { + t.Fatalf("failed to register scheduler library: %v", err) + } + + // Scheduler components read namespace-scoped Tweakables via the CHASM context, + // so each execution's backend must answer GetNamespaceEntry. The transition + // counter is held constant (matching helper_test.go); the chasmtest engine's + // default incrementing counter prevents re-armed pure tasks from materializing. + tv := testvars.New(t) + engine := chasmtest.NewEngine(t, registry, + chasmtest.WithTimeSource(ts), + chasmtest.WithNodeBackendDecorator(func(b *chasm.MockNodeBackend) { + b.HandleGetNamespaceEntry = tv.Namespace + b.HandleNextTransitionCount = func() int64 { return 2 } + b.HandleCurrentVersionedTransition = func() *persistencespb.VersionedTransition { + return &persistencespb.VersionedTransition{NamespaceFailoverVersion: 1, TransitionCount: 1} + } + }), + ) + + d := &Driver{ + T: t, + Engine: engine, + Registry: registry, + TimeSource: ts, + Logger: logger, + Frontend: frontend, + History: history, + dispatchedSideEffects: make(map[*tasks.ChasmTask]struct{}), + AfterStep: cfg.afterStep, + } + return d, cfg +} + +// startExecution starts the scheduler execution using the given start function. +func (d *Driver) startExecution(startFn func(chasm.MutableContext) (*scheduler.Scheduler, error)) { + d.T.Helper() + + d.execKey = chasm.ExecutionKey{NamespaceID: NamespaceID, BusinessID: ScheduleID} + engineCtx := chasm.NewEngineContext(context.Background(), d.Engine) + + _, err := chasm.StartExecution(engineCtx, d.execKey, func(mctx chasm.MutableContext, _ *struct{}) (*scheduler.Scheduler, error) { + return startFn(mctx) + }, (*struct{})(nil), chasm.WithRequestID("create-req")) + if err != nil { + d.T.Fatalf("failed to start scheduler execution: %v", err) + } + + d.ref = chasm.NewComponentRef[*scheduler.Scheduler](d.execKey) +} + +// Now returns the engine's current clock time. +func (d *Driver) Now() time.Time { + return d.TimeSource.Now() +} + +// ReadScheduler reads the scheduler root component and invokes fn with it. +func (d *Driver) ReadScheduler(fn func(s *scheduler.Scheduler, ctx chasm.Context)) { + d.T.Helper() + engineCtx := chasm.NewEngineContext(context.Background(), d.Engine) + err := d.Engine.ReadComponent(engineCtx, d.ref, func(ctx chasm.Context, c chasm.Component) error { + s, ok := c.(*scheduler.Scheduler) + if !ok { + d.T.Fatalf("root component is %T, want *scheduler.Scheduler", c) + } + fn(s, ctx) + return nil + }) + if err != nil { + d.T.Fatalf("ReadScheduler failed: %v", err) + } +} + +// node returns the CHASM tree node backing the scheduler execution. +func (d *Driver) node() *chasm.Node { + d.T.Helper() + node, err := d.Engine.NodeForRef(d.ref) + if err != nil { + d.T.Fatalf("NodeForRef failed: %v", err) + } + return node +} + +func (d *Driver) backend() *chasm.MockNodeBackend { + d.T.Helper() + backend, err := d.Engine.BackendForRef(d.ref) + if err != nil { + d.T.Fatalf("BackendForRef failed: %v", err) + } + return backend +} + +// nextFireTime returns the earliest time at which any pending task (pure or +// side-effect) is due, and whether any task is pending at all. +func (d *Driver) nextFireTime() (time.Time, bool) { + backend := d.backend() + + var next time.Time + have := false + consider := func(t time.Time) { + if !have || t.Before(next) { + next = t + have = true + } + } + + // Earliest pending pure task: recorded by closeTransactionGeneratePhysicalPureTask + // via DeleteCHASMPureTasks(earliestPureTaskTime), or MaximumKey.FireTime if none. + if pureTime := backend.LastDeletePureTaskCall(); !pureTime.Equal(maxKeyFireTime) { + consider(pureTime) + } + + // Earliest undispatched side-effect task. + for _, t := range d.pendingSideEffectTasks() { + consider(t.GetVisibilityTime()) + } + + return next, have +} + +// pendingSideEffectTasks returns physical side-effect tasks that have not yet +// been dispatched by this driver. +func (d *Driver) pendingSideEffectTasks() []*tasks.ChasmTask { + backend := d.backend() + var out []*tasks.ChasmTask + for category, categoryTasks := range backend.TasksByCategory { + // Visibility tasks are framework-managed (indexing search attributes) and + // processed by a separate queue, not the CHASM side-effect executor; their + // handler panics if invoked directly. Skip them. + if category.ID() == tasks.CategoryVisibility.ID() { + continue + } + for _, t := range categoryTasks { + ct, ok := t.(*tasks.ChasmTask) + if !ok { + continue + } + if _, done := d.dispatchedSideEffects[ct]; done { + continue + } + out = append(out, ct) + } + } + return out +} + +// Step advances the clock to the next scheduled task and fires all tasks due at +// that time. It returns false if nothing was pending (the schedule is quiescent). +func (d *Driver) Step() bool { + d.T.Helper() + + fireTime, have := d.nextFireTime() + if !have { + return false + } + + // Never move the clock backward; immediate tasks (zero scheduled time) fire now. + now := d.Now() + if fireTime.Before(now) { + fireTime = now + } + d.TimeSource.Update(fireTime) + + node := d.node() + + // Fire all pure tasks due at fireTime, then commit. + pureCtx := chasm.NewEngineContext(context.Background(), d.Engine) + err := node.EachPureTask(fireTime, func(executor chasm.NodePureTask, attrs chasm.TaskAttributes, taskInstance any) (bool, error) { + return executor.ExecutePureTask(pureCtx, attrs, taskInstance) + }) + if err != nil { + d.T.Fatalf("EachPureTask failed: %v", err) + } + if _, err := node.CloseTransaction(); err != nil { + d.T.Fatalf("CloseTransaction after pure tasks failed: %v", err) + } + + // Fire side-effect tasks now due (visibility <= fireTime), each exactly once. + engineCtx := chasm.NewEngineContext(context.Background(), d.Engine) + for _, ct := range d.pendingSideEffectTasks() { + if ct.GetVisibilityTime().After(fireTime) { + continue + } + d.dispatchedSideEffects[ct] = struct{}{} + execErr := node.ExecuteSideEffectTask( + engineCtx, + d.execKey, + ct, + func(chasm.NodeBackend, chasm.Context, chasm.Component) error { return nil }, + ) + if execErr != nil { + // A NotFound means the logical task was superseded/removed before + // dispatch (the backend accumulates physical tasks append-only, so a + // stale one can linger). Production drops these; so do we. + var notFound *serviceerror.NotFound + if errors.As(execErr, ¬Found) { + continue + } + d.T.Fatalf("ExecuteSideEffectTask failed: %v", execErr) + } + } + if _, err := node.CloseTransaction(); err != nil { + d.T.Fatalf("CloseTransaction after side-effect tasks failed: %v", err) + } + + if d.AfterStep != nil { + d.AfterStep(d) + } + return true +} + +// RunToQuiescence steps until no task is pending or maxSteps is reached. It +// returns the number of steps taken and whether the schedule reached quiescence +// (no pending task). A normal running schedule never quiesces and will stop at +// maxSteps with quiescent == false. +func (d *Driver) RunToQuiescence(maxSteps int) (steps int, quiescent bool) { + d.T.Helper() + for steps = 0; steps < maxSteps; steps++ { + if !d.Step() { + return steps, true + } + } + return steps, false +} + +// HasPendingTask reports whether any task (pure or side-effect) is currently +// scheduled to fire. +func (d *Driver) HasPendingTask() bool { + _, have := d.nextFireTime() + return have +} + +// Patch applies a SchedulePatch through the production Scheduler.Patch path in a +// single transaction. It returns the scheduler's error (e.g. ErrClosed when the +// schedule has already closed) without failing the test, so callers driving +// random operations can tolerate rejected patches. On success the AfterStep hook +// runs so invariants are checked at the new settled state. +func (d *Driver) Patch(patch *schedulepb.SchedulePatch) error { + d.T.Helper() + engineCtx := chasm.NewEngineContext(context.Background(), d.Engine) + _, err := d.Engine.UpdateComponent(engineCtx, d.ref, func(mctx chasm.MutableContext, c chasm.Component) error { + s, ok := c.(*scheduler.Scheduler) + if !ok { + return fmt.Errorf("root component is %T, want *scheduler.Scheduler", c) + } + _, perr := s.Patch(mctx, &schedulerpb.PatchScheduleRequest{ + NamespaceId: NamespaceID, + FrontendRequest: &workflowservice.PatchScheduleRequest{ + Namespace: Namespace, + ScheduleId: ScheduleID, + Patch: patch, + }, + }) + return perr + }) + if err == nil && d.AfterStep != nil { + d.AfterStep(d) + } + return err +} + +// Pause pauses the schedule via Patch. +func (d *Driver) Pause() error { + return d.Patch(&schedulepb.SchedulePatch{Pause: "paused by driver"}) +} + +// Unpause unpauses the schedule via Patch. +func (d *Driver) Unpause() error { + return d.Patch(&schedulepb.SchedulePatch{Unpause: "unpaused by driver"}) +} + +// TriggerImmediately requests an immediate action via Patch. +func (d *Driver) TriggerImmediately() error { + return d.Patch(&schedulepb.SchedulePatch{ + TriggerImmediately: &schedulepb.TriggerImmediatelyRequest{}, + }) +} + +// MigrateToWorkflow initiates a V2->V1 migration: it pauses the schedule and +// schedules the SchedulerMigrateToWorkflowTask side-effect task. Step the driver +// afterward to fire the task (which calls the history client's +// StartWorkflowExecution with the exported StartScheduleArgs). +func (d *Driver) MigrateToWorkflow() error { + d.T.Helper() + engineCtx := chasm.NewEngineContext(context.Background(), d.Engine) + _, err := d.Engine.UpdateComponent(engineCtx, d.ref, func(mctx chasm.MutableContext, c chasm.Component) error { + s, ok := c.(*scheduler.Scheduler) + if !ok { + return fmt.Errorf("root component is %T, want *scheduler.Scheduler", c) + } + _, merr := s.MigrateToWorkflow(mctx, &schedulerpb.MigrateToWorkflowRequest{ + NamespaceId: NamespaceID, + ScheduleId: ScheduleID, + }) + return merr + }) + return err +} + +// Snapshot captures the observable scheduler state at a settled point (after a +// step's CloseTransaction). It is the input to invariant checks. +type Snapshot struct { + // Now is the engine clock at capture time. + Now time.Time + // HasPendingTask is true if any task is scheduled to fire. + HasPendingTask bool + + Closed bool + IsSentinel bool + Paused bool + // BackfillerCount is the number of live backfiller components. + BackfillerCount int + // IsHeldOpen mirrors Scheduler.isHeldOpen: the schedule must stay open + // regardless of having no work (paused or pending backfill, non-sentinel). + IsHeldOpen bool + // IdleCloseTime is the armed idle-close deadline, or zero if no idle task is armed. + IdleCloseTime time.Time + + // GeneratorLPT / InvokerLPT are the generator and invoker high-water marks. + GeneratorLPT time.Time + InvokerLPT time.Time +} + +// Snapshot reads current scheduler state into a Snapshot. +func (d *Driver) Snapshot() Snapshot { + d.T.Helper() + snap := Snapshot{ + Now: d.Now(), + HasPendingTask: d.HasPendingTask(), + } + d.ReadScheduler(func(s *scheduler.Scheduler, ctx chasm.Context) { + snap.Closed = s.Closed + snap.IsSentinel = s.IsSentinel() + snap.Paused = s.Schedule.GetState().GetPaused() + snap.BackfillerCount = len(s.Backfillers) + // Mirror Scheduler.isHeldOpen() from public state. + snap.IsHeldOpen = !snap.IsSentinel && (snap.Paused || snap.BackfillerCount > 0) + if s.IdleCloseTime != nil { + snap.IdleCloseTime = s.IdleCloseTime.AsTime() + } + // Sentinels have no generator/invoker sub-components. + if !snap.IsSentinel { + if g := s.Generator.Get(ctx); g != nil && g.LastProcessedTime != nil { + snap.GeneratorLPT = g.LastProcessedTime.AsTime() + } + if inv := s.Invoker.Get(ctx); inv != nil && inv.LastProcessedTime != nil { + snap.InvokerLPT = inv.LastProcessedTime.AsTime() + } + } + }) + return snap +} diff --git a/chasm/lib/scheduler/schedulertest/driver_smoke_test.go b/chasm/lib/scheduler/schedulertest/driver_smoke_test.go new file mode 100644 index 00000000000..4f80dfa219a --- /dev/null +++ b/chasm/lib/scheduler/schedulertest/driver_smoke_test.go @@ -0,0 +1,72 @@ +package schedulertest_test + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.temporal.io/server/chasm" + "go.temporal.io/server/chasm/lib/scheduler" + "go.temporal.io/server/chasm/lib/scheduler/schedulertest" +) + +// TestDriver_RunsIntervalScheduleForward verifies the driver fires the +// scheduler's self-scheduled tasks and advances the clock across several +// interval ticks of a normal (running) schedule. +func TestDriver_RunsIntervalScheduleForward(t *testing.T) { + start := time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC) + d := schedulertest.NewDriver(t, schedulertest.WithStartTime(start)) + + steps, quiescent := d.RunToQuiescence(20) + + // A normal interval schedule re-arms a generator task forever, so it never + // quiesces; it should burn the whole step budget. + require.False(t, quiescent, "running interval schedule should not quiesce") + require.Equal(t, 20, steps) + + // The clock must have advanced as interval ticks fire. + require.True(t, d.Now().After(start.Add(schedulertest.DefaultInterval)), + "clock should advance past one interval, now=%s", d.Now()) + + d.ReadScheduler(func(s *scheduler.Scheduler, _ chasm.Context) { + require.False(t, s.Closed, "schedule should still be open") + }) +} + +// TestDriver_PausedScheduleKeepsTicking documents that a paused interval +// schedule is *held open* and keeps re-arming its generator task to advance the +// high-water mark — it neither quiesces nor closes — but it never buffers a +// start. +func TestDriver_PausedScheduleKeepsTicking(t *testing.T) { + sched := schedulertest.DefaultSchedule() + sched.State.Paused = true + + d := schedulertest.NewDriver(t, schedulertest.WithSchedule(sched)) + + _, quiescent := d.RunToQuiescence(10) + require.False(t, quiescent, "paused interval schedule keeps ticking, not quiescent") + + d.ReadScheduler(func(s *scheduler.Scheduler, _ chasm.Context) { + require.True(t, s.Schedule.State.Paused) + require.False(t, s.Closed, "paused schedule is held open, not closed") + }) +} + +// TestDriver_LimitedActionsExhaustedClosesAndQuiesces verifies that a schedule +// with no remaining actions arms an idle task instead of generating work, and +// after the idle task fires the schedule closes and the driver reaches +// quiescence. +func TestDriver_LimitedActionsExhaustedClosesAndQuiesces(t *testing.T) { + sched := schedulertest.DefaultSchedule() + sched.State.LimitedActions = true + sched.State.RemainingActions = 0 + + d := schedulertest.NewDriver(t, schedulertest.WithSchedule(sched)) + + _, quiescent := d.RunToQuiescence(10) + require.True(t, quiescent, "exhausted schedule should close and quiesce") + + d.ReadScheduler(func(s *scheduler.Scheduler, _ chasm.Context) { + require.True(t, s.Closed, "schedule should be closed after idle task fires") + }) +} diff --git a/chasm/lib/scheduler/schedulertest/library.go b/chasm/lib/scheduler/schedulertest/library.go new file mode 100644 index 00000000000..fcaf1235efe --- /dev/null +++ b/chasm/lib/scheduler/schedulertest/library.go @@ -0,0 +1,170 @@ +// Package schedulertest provides a lifecycle-driving harness for the CHASM +// scheduler. Unlike the per-handler unit tests in package scheduler_test, this +// package builds a real component on top of [chasmtest.Engine] and steps it +// forward through every task it schedules for itself, so that invariants can be +// asserted across sequences of transitions (e.g. "the scheduler never gets +// stuck without a task to drive the next state"). +// +// It deliberately lives in a non-test package so that property tests +// (pgregory.net/rapid) and the V1<->V2 migration round-trip test can import the +// driver. The helpers below mirror chasm/lib/scheduler/helper_test.go, but are +// exported and accept the mock clients the side-effect task handlers need. +package schedulertest + +import ( + "time" + + commonpb "go.temporal.io/api/common/v1" + schedulepb "go.temporal.io/api/schedule/v1" + workflowpb "go.temporal.io/api/workflow/v1" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/server/chasm/lib/scheduler" + "go.temporal.io/server/common/backoff" + "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/log" + "go.temporal.io/server/common/metrics" + "go.temporal.io/server/common/resource" + "go.temporal.io/server/common/searchattribute" + legacyscheduler "go.temporal.io/server/service/worker/scheduler" + "go.uber.org/mock/gomock" + "google.golang.org/protobuf/types/known/durationpb" +) + +const ( + Namespace = "ns" + NamespaceID = "ns-id" + ScheduleID = "sched-id" + + DefaultInterval = 1 * time.Minute + DefaultCatchupWindow = 5 * time.Minute +) + +// DefaultSchedule returns a schedule with a single 1-minute interval and a +// 5-minute catchup window, matching the package's other defaults. +func DefaultSchedule() *schedulepb.Schedule { + return &schedulepb.Schedule{ + Spec: &schedulepb.ScheduleSpec{ + Interval: []*schedulepb.IntervalSpec{ + { + Interval: durationpb.New(DefaultInterval), + Phase: durationpb.New(0), + }, + }, + }, + Action: &schedulepb.ScheduleAction{ + Action: &schedulepb.ScheduleAction_StartWorkflow{ + StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{ + WorkflowId: "scheduled-wf", + WorkflowType: &commonpb.WorkflowType{Name: "scheduled-wf-type"}, + }, + }, + }, + Policies: &schedulepb.SchedulePolicies{ + CatchupWindow: durationpb.New(DefaultCatchupWindow), + }, + State: &schedulepb.ScheduleState{ + Paused: false, + LimitedActions: false, + RemainingActions: 0, + }, + } +} + +// DefaultConfig returns a scheduler config wired with the package defaults. +func DefaultConfig() *scheduler.Config { + return &scheduler.Config{ + Tweakables: func(_ string) scheduler.Tweakables { + return scheduler.DefaultTweakables + }, + ServiceCallTimeout: func() time.Duration { + return 5 * time.Second + }, + EncodeInternalTokenWithEnvelope: func(string) bool { + return true + }, + RetryPolicy: func() backoff.RetryPolicy { + return backoff.NewExponentialRetryPolicy(1 * time.Second) + }, + } +} + +// NewRealSpecProcessor builds a real SpecProcessor with no-op metrics, matching +// production spec processing so generated buffered starts and next-times are +// realistic. +func NewRealSpecProcessor(ctrl *gomock.Controller, logger log.Logger) scheduler.SpecProcessor { + mockMetrics := metrics.NewMockHandler(ctrl) + mockMetrics.EXPECT().Counter(gomock.Any()).Return(metrics.NoopCounterMetricFunc).AnyTimes() + mockMetrics.EXPECT().WithTags(gomock.Any()).Return(mockMetrics).AnyTimes() + mockMetrics.EXPECT().Timer(gomock.Any()).Return(metrics.NoopTimerMetricFunc).AnyTimes() + + return scheduler.NewSpecProcessor( + DefaultConfig(), + mockMetrics, + logger, + legacyscheduler.NewSpecBuilder( + dynamicconfig.SchedulerSpecWarnIterations.Get(dynamicconfig.NewNoopCollection()), + dynamicconfig.SchedulerSpecMaxIterations.Get(dynamicconfig.NewNoopCollection()), + ), + ) +} + +// NewTestLibrary builds a scheduler Library with all task handlers wired to the +// supplied spec processor and mock clients. The clients back the side-effect +// task handlers (execute/callbacks/migrate); pass mocks whose StartWorkflow / +// Terminate / Cancel expectations the test controls. +func NewTestLibrary( + config *scheduler.Config, + logger log.Logger, + specProcessor scheduler.SpecProcessor, + frontendClient workflowservice.WorkflowServiceClient, + historyClient resource.HistoryClient, +) *scheduler.Library { + specBuilder := legacyscheduler.NewSpecBuilder( + dynamicconfig.SchedulerSpecWarnIterations.Get(dynamicconfig.NewNoopCollection()), + dynamicconfig.SchedulerSpecMaxIterations.Get(dynamicconfig.NewNoopCollection()), + ) + invokerOpts := scheduler.InvokerTaskHandlerOptions{ + Config: config, + MetricsHandler: metrics.NoopMetricsHandler, + BaseLogger: logger, + SpecProcessor: specProcessor, + HistoryClient: historyClient, + FrontendClient: frontendClient, + } + return scheduler.NewLibrary( + config, + nil, + scheduler.NewSchedulerIdleTaskHandler(scheduler.SchedulerIdleTaskHandlerOptions{ + Config: config, + MetricsHandler: metrics.NoopMetricsHandler, + BaseLogger: logger, + }), + scheduler.NewSchedulerCallbacksTaskHandler(scheduler.SchedulerCallbacksTaskHandlerOptions{ + Config: config, + HistoryClient: historyClient, + FrontendClient: frontendClient, + }), + scheduler.NewGeneratorTaskHandler(scheduler.GeneratorTaskHandlerOptions{ + Config: config, + MetricsHandler: metrics.NoopMetricsHandler, + BaseLogger: logger, + SpecProcessor: specProcessor, + SpecBuilder: specBuilder, + }), + scheduler.NewInvokerExecuteTaskHandler(invokerOpts), + scheduler.NewInvokerProcessBufferTaskHandler(invokerOpts), + scheduler.NewBackfillerTaskHandler(scheduler.BackfillerTaskHandlerOptions{ + Config: config, + MetricsHandler: metrics.NoopMetricsHandler, + BaseLogger: logger, + SpecProcessor: specProcessor, + }), + scheduler.NewSchedulerMigrateToWorkflowTaskHandler(scheduler.SchedulerMigrateToWorkflowTaskHandlerOptions{ + Config: config, + MetricsHandler: metrics.NoopMetricsHandler, + BaseLogger: logger, + HistoryClient: historyClient, + SaMapperProvider: searchattribute.NewTestMapperProvider(nil), + }), + ) +} diff --git a/service/worker/scheduler/testdata/replay_migration_v1_to_v2.json.gz b/service/worker/scheduler/testdata/replay_migration_v1_to_v2.json.gz new file mode 100644 index 00000000000..7cc141ea49a Binary files /dev/null and b/service/worker/scheduler/testdata/replay_migration_v1_to_v2.json.gz differ diff --git a/service/worker/scheduler/workflow.go b/service/worker/scheduler/workflow.go index 1d9b60728ec..699a2ee59da 100644 --- a/service/worker/scheduler/workflow.go +++ b/service/worker/scheduler/workflow.go @@ -66,6 +66,10 @@ const ( LimitMemoSpecSize = 11 // trigger immediately timestamp is added to the PatchRequest TriggerImmediatelyTimestamp = 12 + // reconcile running-workflow status before evaluating CHASM migration + // eligibility, so an actively-firing schedule can observe an empty + // RunningWorkflows window and migrate instead of deferring forever + RefreshBeforeMigrationCheck = 13 ) const ( @@ -215,7 +219,7 @@ var ( ReuseTimer: true, NextTimeCacheV2Size: 14, // see note below SpecFieldLengthLimit: 10, - Version: TriggerImmediatelyTimestamp, + Version: RefreshBeforeMigrationCheck, } // Note on NextTimeCacheV2Size: This value must be > FutureActionCountForList. Each @@ -334,6 +338,25 @@ func (s *scheduler) run() error { ) } + // Reconcile running-workflow status before evaluating migration + // eligibility. processTimeRange (above) sets NeedRefresh when it buffers a + // new action, but the refresh itself only runs inside processBuffer, which + // is after this check and also starts the replacement action. Without this, + // an actively-firing schedule's RunningWorkflows is stale-non-empty here + // every iteration (the just-completed action is never pruned in time), so + // len(RunningWorkflows)==0 is never observed and, with the default + // MigrateWithRunningWorkflows=false guard, migration is deferred forever. + // Refreshing here lets the check see the genuine post-completion window and + // migrate before the next action is started. Version-gated because + // refreshWorkflows issues (recorded) local activities: reordering it would + // otherwise break replay of pre-existing histories. + if s.hasMinVersion(RefreshBeforeMigrationCheck) && + s.tweakables.EnableCHASMMigration && !s.State.PendingMigration && + s.State.NeedRefresh { + s.refreshWorkflows(slices.Clone(s.Info.RunningWorkflows)) + s.State.NeedRefresh = false + } + if !s.State.PendingMigration && s.tweakables.EnableCHASMMigration && (s.tweakables.MigrateWithRunningWorkflows || len(s.Info.RunningWorkflows) == 0) { s.State.PendingMigration = true diff --git a/service/worker/scheduler/workflow_test.go b/service/worker/scheduler/workflow_test.go index af190fdcd67..085abad0460 100644 --- a/service/worker/scheduler/workflow_test.go +++ b/service/worker/scheduler/workflow_test.go @@ -2306,6 +2306,85 @@ func (s *workflowSuite) TestMigrateSuccess() { s.NoError(s.env.GetWorkflowError()) } +// TestAutoMigrateReconcilesRunningWorkflowBeforeCheck is a regression test for +// the read-ordering bug fixed by the RefreshBeforeMigrationCheck version. An +// actively-firing V1 schedule tracks a workflow in RunningWorkflows that has +// actually already completed; NeedRefresh is set (as processTimeRange does when +// it buffers a new action). Automatic migration is enabled with the default +// MigrateWithRunningWorkflows=false guard, and -- unlike the other TestMigrate* +// tests -- NO migrate-to-chasm signal is sent, so migration relies solely on +// the automatic-eligibility check. +// +// Before the fix, that check read RunningWorkflows before processBuffer's +// refresh, so the already-completed workflow was still counted as running and +// migration was deferred. The fix reconciles workflow status before the check, +// so the stale entry is pruned and migration proceeds within the same idle +// iteration -- before the schedule's next action is even due. We assert +// migration happened before the first scheduled action time to pin that down. +// (The full continuously-firing reproduction lives in the integration test +// TestScheduleMigrationV1ToV2_AutomaticMigrationForActiveSchedule.) +func (s *workflowSuite) TestAutoMigrateReconcilesRunningWorkflowBeforeCheck() { + staleWID := "myid-2022-06-01T00:00:00Z" + + // The refresh watcher reports the stale workflow as already completed. + s.env.OnActivity(new(activities).WatchWorkflow, mock.Anything, mock.Anything).Return( + &schedulespb.WatchWorkflowResponse{Status: enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED}, nil) + + // No action should ever be started: migration must happen during the idle + // window before the first action fires. + s.env.OnActivity(new(activities).StartWorkflow, mock.Anything, mock.Anything).Times(0).Maybe().Return( + func(_ context.Context, req *schedulespb.StartWorkflowRequest) (*schedulespb.StartWorkflowResponse, error) { + s.Failf("unexpected start", "for %s at %s", req.Request.WorkflowId, s.now()) + return nil, nil + }) + + var migratedAt time.Time + migrated := false + s.env.OnActivity(new(activities).MigrateScheduleToChasm, mock.Anything, mock.Anything).Once().Return( + func(context.Context, *schedulerpb.CreateFromMigrationStateRequest) error { + migrated = true + migratedAt = s.now() + return nil + }) + + CurrentTweakablePolicies.IterationsBeforeContinueAsNew = 100 + s.env.SetStartTime(baseStartTime) + s.env.ExecuteWorkflow(func(ctx workflow.Context, args *schedulespb.StartScheduleArgs) error { + // enableCHASMMigration=true, migrateWithRunningWorkflows=false (guard on). + return schedulerWorkflowWithSpecBuilder(ctx, args, newSpecBuilderForTest(0, 0), + func() bool { return true }, func() bool { return false }) + }, &schedulespb.StartScheduleArgs{ + Schedule: &schedulepb.Schedule{ + Spec: &schedulepb.ScheduleSpec{ + Interval: []*schedulepb.IntervalSpec{{ + Interval: durationpb.New(1 * time.Hour), + }}, + }, + Action: s.defaultAction("myid"), + }, + Info: &schedulepb.ScheduleInfo{ + RunningWorkflows: []*commonpb.WorkflowExecution{{WorkflowId: staleWID}}, + }, + State: &schedulespb.InternalState{ + Namespace: "myns", + NamespaceId: "mynsid", + ScheduleId: "myschedule", + ConflictToken: InitialConflictToken, + LastProcessedTime: timestamppb.New(baseStartTime), + // Mimics processTimeRange having just buffered an action: production + // sets NeedRefresh in that path, and it is what previously ran only + // inside processBuffer, after the eligibility check. + NeedRefresh: true, + }, + }) + + s.True(s.env.IsWorkflowCompleted()) + s.Require().NoError(s.env.GetWorkflowError(), "schedule should migrate (complete), not defer/CAN") + s.True(migrated, "MigrateScheduleToChasm should have been called via auto-eligibility") + s.True(migratedAt.Before(baseStartTime.Add(time.Hour)), + "migration should occur in the idle window before the first action fires, at %s", migratedAt) +} + func (s *workflowSuite) TestMigrateFailure() { // Mock MigrateSchedule activity to always fail. Migration is retried // each iteration since PendingMigration is persisted in State. diff --git a/tests/schedule_migration_replay_fixture_gen_test.go b/tests/schedule_migration_replay_fixture_gen_test.go new file mode 100644 index 00000000000..88b98d722fc --- /dev/null +++ b/tests/schedule_migration_replay_fixture_gen_test.go @@ -0,0 +1,147 @@ +package tests + +import ( + "bytes" + "compress/gzip" + "os" + "path/filepath" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" + schedulepb "go.temporal.io/api/schedule/v1" + taskqueuepb "go.temporal.io/api/taskqueue/v1" + workflowpb "go.temporal.io/api/workflow/v1" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/sdk/workflow" + "go.temporal.io/server/api/historyservice/v1" + "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/testing/await" + "go.temporal.io/server/service/worker/scheduler" + "go.temporal.io/server/tests/testcore" + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/types/known/durationpb" +) + +// TestGenerateMigrationReplayFixture regenerates the checked-in replay fixture +// service/worker/scheduler/testdata/replay_migration_v1_to_v2.json.gz, which +// guards the RefreshBeforeMigrationCheck (v13) run-loop change against future +// nondeterministic edits. It captures the history of a real V1 scheduler +// workflow that auto-migrated to CHASM (exercising the early-refresh + +// eligibility branch), and writes it in the same protojson format +// client.HistoryFromJSON expects. +// +// It is inert unless GENERATE_REPLAY_FIXTURE is set, so it never runs in CI: +// +// GENERATE_REPLAY_FIXTURE=1 go test -tags=test_dep ./tests/ \ +// -run TestGenerateMigrationReplayFixture -count=1 -v +// +// After regenerating, run `go test ./service/worker/scheduler/ -run TestReplays` +// to confirm the new fixture replays cleanly. +func TestGenerateMigrationReplayFixture(t *testing.T) { + if os.Getenv("GENERATE_REPLAY_FIXTURE") == "" { + t.Skip("set GENERATE_REPLAY_FIXTURE=1 to regenerate the migration replay fixture") + } + + env := testcore.NewEnv(t, testcore.WithWorkerService("V1 scheduler")) + + ctx := testcore.NewContext() + sid := testcore.RandomizeStr("sched-migrate-fixture") + wid := testcore.RandomizeStr("sched-migrate-fixture-wf") + wt := testcore.RandomizeStr("sched-migrate-fixture-wt") + nsName := env.Namespace().String() + + env.OverrideDynamicConfig(dynamicconfig.EnableChasm, false) + + // Short-lived workflow that completes well within the interval, so the + // schedule is a continuous stream of quick actions -- the shape that + // exercises the read-ordering fix. + env.SdkWorker().RegisterWorkflowWithOptions(func(ctx workflow.Context) error { + return workflow.Sleep(ctx, 200*time.Millisecond) + }, workflow.RegisterOptions{Name: wt}) + + sched := &schedulepb.Schedule{ + Spec: &schedulepb.ScheduleSpec{ + Interval: []*schedulepb.IntervalSpec{{Interval: durationpb.New(1 * time.Second)}}, + }, + Action: &schedulepb.ScheduleAction{ + Action: &schedulepb.ScheduleAction_StartWorkflow{ + StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{ + WorkflowId: wid, + WorkflowType: &commonpb.WorkflowType{Name: wt}, + TaskQueue: &taskqueuepb.TaskQueue{Name: env.WorkerTaskQueue(), Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, + }, + }, + }, + } + _, err := env.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{ + Namespace: nsName, + ScheduleId: sid, + Schedule: sched, + Identity: "test", + RequestId: uuid.NewString(), + }) + require.NoError(t, err) + + // Let it become an established, actively-firing V1 schedule. + await.RequireTrue(t, func() bool { + descResp, err := env.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{ + Namespace: nsName, ScheduleId: sid, + }) + return err == nil && len(descResp.GetInfo().GetRecentActions()) > 0 + }, 15*time.Second, 500*time.Millisecond) + + // Enable automatic migration; the running-workflow guard stays at its default. + env.OverrideDynamicConfig(dynamicconfig.EnableChasm, true) + env.OverrideDynamicConfig(dynamicconfig.EnableCHASMSchedulerMigration, true) + env.OverrideDynamicConfig(dynamicconfig.CHASMSchedulerMigrationRolloutPercent, 100) + + // The V1 scheduler workflow completes only when migration succeeds. + v1WorkflowID := scheduler.WorkflowIDPrefix + sid + await.RequireTruef(t, func() bool { + desc, err := env.GetTestCluster().HistoryClient().DescribeWorkflowExecution(ctx, + &historyservice.DescribeWorkflowExecutionRequest{ + NamespaceId: env.NamespaceID().String(), + Request: &workflowservice.DescribeWorkflowExecutionRequest{ + Namespace: nsName, + Execution: &commonpb.WorkflowExecution{WorkflowId: v1WorkflowID}, + }, + }) + return err == nil && desc.GetWorkflowExecutionInfo().GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED + }, 30*time.Second, 1*time.Second, "V1 schedule should auto-migrate to V2") + + // Capture the full history of the migrated V1 scheduler workflow. + var events []*historypb.HistoryEvent + var token []byte + for { + resp, err := env.FrontendClient().GetWorkflowExecutionHistory(ctx, &workflowservice.GetWorkflowExecutionHistoryRequest{ + Namespace: nsName, + Execution: &commonpb.WorkflowExecution{WorkflowId: v1WorkflowID}, + NextPageToken: token, + }) + require.NoError(t, err) + events = append(events, resp.GetHistory().GetEvents()...) + token = resp.GetNextPageToken() + if len(token) == 0 { + break + } + } + require.NotEmpty(t, events) + + data, err := protojson.Marshal(&historypb.History{Events: events}) + require.NoError(t, err) + + outPath := filepath.Join("..", "service", "worker", "scheduler", "testdata", "replay_migration_v1_to_v2.json.gz") + var buf bytes.Buffer + gw, _ := gzip.NewWriterLevel(&buf, gzip.BestCompression) + _, err = gw.Write(data) + require.NoError(t, err) + require.NoError(t, gw.Close()) + require.NoError(t, os.WriteFile(outPath, buf.Bytes(), 0o644)) + + t.Logf("wrote %s (%d events)", outPath, len(events)) +} diff --git a/tests/schedule_migration_v1_to_v2_callback_compat_test.go b/tests/schedule_migration_v1_to_v2_callback_compat_test.go new file mode 100644 index 00000000000..5052780c993 --- /dev/null +++ b/tests/schedule_migration_v1_to_v2_callback_compat_test.go @@ -0,0 +1,334 @@ +package tests + +import ( + "context" + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + schedulepb "go.temporal.io/api/schedule/v1" + "go.temporal.io/api/serviceerror" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/sdk/workflow" + "go.temporal.io/server/api/adminservice/v1" + "go.temporal.io/server/api/historyservice/v1" + "go.temporal.io/server/chasm" + schedulerpb "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1" + "go.temporal.io/server/common/dynamicconfig" + "go.temporal.io/server/common/testing/await" + "go.temporal.io/server/service/worker/scheduler" + "go.temporal.io/server/tests/testcore" + "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/durationpb" +) + +// requireNoChasmSentinel asserts that no CHASM entity -- sentinel or otherwise +// -- exists yet for scheduleID under the scheduler archetype. Used right after +// creating a V1 schedule to confirm the test's "no sentinel gets written" +// assumption directly (rather than only inferring it from the EnableChasm +// value passed to CreateSchedule), since a stray/unexpired sentinel would +// invisibly gate migration behind chasm/lib/scheduler/config.go's +// SentinelIdleTime (15 minutes) and make an otherwise-passing test hang or +// flake for the wrong reason. +func requireNoChasmSentinel(ctx context.Context, t *testing.T, env *testcore.TestEnv, scheduleID string) { + t.Helper() + + resp, err := env.AdminClient().DescribeMutableState(ctx, &adminservice.DescribeMutableStateRequest{ + Namespace: env.Namespace().String(), + Execution: &commonpb.WorkflowExecution{WorkflowId: scheduleID}, + Archetype: string(chasm.SchedulerArchetype), + }) + if err != nil { + // NotFound means nothing at all was written to the CHASM key space for + // this schedule ID yet -- definitely no sentinel. Any other error is + // unexpected and should fail the test. + var notFoundErr *serviceerror.NotFound + require.ErrorAs(t, err, ¬FoundErr, "unexpected error checking for a CHASM sentinel") + return + } + + node := resp.GetDatabaseMutableState().GetChasmNodes()[""] + require.NotNil(t, node, "CHASM execution exists for %q but has no root node", scheduleID) + + var state schedulerpb.SchedulerState + require.NoError(t, proto.Unmarshal(node.GetData().GetData(), &state)) + require.False(t, state.GetSentinel(), + "a CHASM sentinel exists for schedule %q -- it would block migration for up to "+ + "SentinelIdleTime (chasm/lib/scheduler/config.go), invalidating this test's timing", scheduleID) +} + +// createV1Schedule creates a V1 (workflow-backed) schedule with CHASM disabled, +// then asserts no CHASM sentinel was written. initialPatch may be nil. +func createV1Schedule( + ctx context.Context, + t *testing.T, + env *testcore.TestEnv, + scheduleID string, + sched *schedulepb.Schedule, + initialPatch *schedulepb.SchedulePatch, +) { + t.Helper() + + // EnableChasm is unset at creation time, so no CHASM sentinel gets written -- + // a sentinel would block migration for 15 minutes (chasm/lib/scheduler/config.go + // SentinelIdleTime), which these tests can't afford to wait out. + env.OverrideDynamicConfig(dynamicconfig.EnableChasm, false) + + _, err := env.FrontendClient().CreateSchedule(ctx, &workflowservice.CreateScheduleRequest{ + Namespace: env.Namespace().String(), + ScheduleId: scheduleID, + Schedule: sched, + InitialPatch: initialPatch, + Identity: "test", + RequestId: uuid.NewString(), + }) + require.NoError(t, err) + requireNoChasmSentinel(ctx, t, env, scheduleID) +} + +// awaitRunningAction waits until the schedule has fired an action whose workflow +// is still RUNNING, and returns that workflow's ID. +func awaitRunningAction(ctx context.Context, t *testing.T, env *testcore.TestEnv, scheduleID string) string { + t.Helper() + + var runningWfID string + await.RequireTrue(t, func() bool { + descResp, err := env.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{ + Namespace: env.Namespace().String(), + ScheduleId: scheduleID, + }) + if err != nil || len(descResp.GetInfo().GetRecentActions()) == 0 { + return false + } + a := descResp.Info.RecentActions[0] + if a.GetStartWorkflowStatus() != enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING { + return false + } + runningWfID = a.GetStartWorkflowResult().GetWorkflowId() + return true + }, 15*time.Second, 500*time.Millisecond) + require.NotEmpty(t, runningWfID) + return runningWfID +} + +// awaitAnyAction waits until the schedule has fired at least one action +// (regardless of the fired workflow's status) and returns that workflow's ID. +func awaitAnyAction(ctx context.Context, t *testing.T, env *testcore.TestEnv, scheduleID string) string { + t.Helper() + + var wfID string + await.RequireTrue(t, func() bool { + descResp, err := env.FrontendClient().DescribeSchedule(ctx, &workflowservice.DescribeScheduleRequest{ + Namespace: env.Namespace().String(), + ScheduleId: scheduleID, + }) + if err != nil || len(descResp.GetInfo().GetRecentActions()) == 0 { + return false + } + wfID = descResp.Info.RecentActions[0].GetStartWorkflowResult().GetWorkflowId() + return wfID != "" + }, 15*time.Second, 500*time.Millisecond) + require.NotEmpty(t, wfID) + return wfID +} + +// awaitV1SchedulerCompleted waits until the V1 scheduler workflow reaches +// COMPLETED. The V1 scheduler workflow only completes when executeMigration() +// succeeds, so its completion is a reliable "migration happened" signal. +func awaitV1SchedulerCompleted(ctx context.Context, t *testing.T, env *testcore.TestEnv, scheduleID string) { + t.Helper() + + v1WorkflowID := scheduler.WorkflowIDPrefix + scheduleID + await.RequireTruef(t, func() bool { + desc, err := env.GetTestCluster().HistoryClient().DescribeWorkflowExecution(ctx, &historyservice.DescribeWorkflowExecutionRequest{ + NamespaceId: env.NamespaceID().String(), + Request: &workflowservice.DescribeWorkflowExecutionRequest{ + Namespace: env.Namespace().String(), + Execution: &commonpb.WorkflowExecution{WorkflowId: v1WorkflowID}, + }, + }) + return err == nil && desc.GetWorkflowExecutionInfo().GetStatus() == enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED + }, 30*time.Second, 1*time.Second, "V1 scheduler workflow should complete once migration succeeds") +} + +// requireNoOptionsUpdatedEvent asserts the workflow's history does not contain +// EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED. V1 never emits this event, so +// older/community SDKs (whose vendored protobuf predates it) cannot decode it +// and crash -- permanently stalling the workflow (see +// repros/scheduler-migration-bug-evidence.md). +func requireNoOptionsUpdatedEvent(ctx context.Context, t *testing.T, env *testcore.TestEnv, workflowID string) { + t.Helper() + + history, err := env.FrontendClient().GetWorkflowExecutionHistory(ctx, &workflowservice.GetWorkflowExecutionHistoryRequest{ + Namespace: env.Namespace().String(), + Execution: &commonpb.WorkflowExecution{WorkflowId: workflowID}, + }) + require.NoError(t, err) + for _, event := range history.GetHistory().GetEvents() { + require.NotEqual(t, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED, event.GetEventType(), + "workflow %q gained %s (event id %d) during migration -- older SDKs cannot decode this "+ + "event type and will permanently stall on it (see repros/scheduler-migration-bug-evidence.md)", + workflowID, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED, event.GetEventId()) + } +} + +// requireV2ScheduleExists asserts a V2 (CHASM) schedule exists for scheduleID. +func requireV2ScheduleExists(ctx context.Context, t *testing.T, env *testcore.TestEnv, scheduleID string) { + t.Helper() + + _, err := env.GetTestCluster().SchedulerClient().DescribeSchedule(ctx, &schedulerpb.DescribeScheduleRequest{ + NamespaceId: env.NamespaceID().String(), + FrontendRequest: &workflowservice.DescribeScheduleRequest{Namespace: env.Namespace().String(), ScheduleId: scheduleID}, + }) + require.NoError(t, err) +} + +// TestScheduleMigrationV1ToV2_AdminMigratePreservesRunningWorkflowHistory is a +// regression guard for a customer-reported crash on the admin (tdbg/CLI) +// MigrateSchedule path. +// +// When a V1 schedule is migrated to V2 via the admin MigrateSchedule RPC while +// one of its fired workflows is still running, CHASM's Invoker retroactively +// attaches a Nexus completion callback to that already-running workflow +// (chasm/lib/scheduler/scheduler_tasks.go, WorkflowIdConflictPolicy USE_EXISTING +// + OnConflictOptions.AttachCompletionCallbacks). Server-side this appends a +// brand-new EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED event into the middle +// of that workflow's existing history. V1 never emits this event, so a workflow +// that was already running before migration never expected to see it, and older +// SDKs (e.g. coinbase/temporal-ruby) crash decoding it -- permanently stalling +// the workflow and, via the schedule's overlap policy, potentially the whole +// schedule (see repros/scheduler-migration-bug-evidence.md). +// +// Unlike the dynamic-config rollout path -- which defers migration until no +// workflow is running (see TestScheduleMigrationV1ToV2_RolloutMigration) -- the +// admin MigrateSchedule RPC migrates immediately regardless of running +// workflows, so this bug is still live on that path. Skipped until the +// retroactive callback-attach is fixed. +func TestScheduleMigrationV1ToV2_AdminMigratePreservesRunningWorkflowHistory(t *testing.T) { + // TODO: file a tracking issue for the admin-path callback-attach bug and reference it here. + t.Skip("admin MigrateSchedule path still injects EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED " + + "into a running workflow's history; remove this skip when the retroactive callback-attach is fixed") + + env := testcore.NewEnv(t, testcore.WithWorkerService("V1 scheduler")) + ctx := testcore.NewContext() + + sid := testcore.RandomizeStr("sched-admin-migrate") + wid := testcore.RandomizeStr("sched-admin-migrate-wf") + wt := testcore.RandomizeStr("sched-admin-migrate-wt") + + // A workflow that blocks until signaled, so it is guaranteed to still be + // running at the moment migration happens. + resumeSignal := "resume" + env.SdkWorker().RegisterWorkflowWithOptions(func(ctx workflow.Context) error { + workflow.GetSignalChannel(ctx, resumeSignal).Receive(ctx, nil) + return nil + }, workflow.RegisterOptions{Name: wt}) + + sched := &schedulepb.Schedule{ + Spec: &schedulepb.ScheduleSpec{Interval: []*schedulepb.IntervalSpec{{Interval: durationpb.New(1 * time.Hour)}}}, + Action: startWorkflowAction(env, wid, wt), + } + createV1Schedule(ctx, t, env, sid, sched, &schedulepb.SchedulePatch{ + TriggerImmediately: &schedulepb.TriggerImmediatelyRequest{}, + }) + + runningWfID := awaitRunningAction(ctx, t, env, sid) + + // Ensure the workflow gets unblocked at the end regardless of outcome, so a + // failing assertion doesn't leak a permanently-running execution. + defer func() { + _, _ = env.FrontendClient().SignalWorkflowExecution(testcore.NewContext(), &workflowservice.SignalWorkflowExecutionRequest{ + Namespace: env.Namespace().String(), + WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: runningWfID}, + SignalName: resumeSignal, + }) + }() + + // Sanity: V1 never emits this event type before migration. + requireNoOptionsUpdatedEvent(ctx, t, env, runningWfID) + + // Migrate via the admin RPC while the fired workflow is still running. + env.OverrideDynamicConfig(dynamicconfig.EnableChasm, true) + _, err := env.AdminClient().MigrateSchedule(ctx, &adminservice.MigrateScheduleRequest{ + Namespace: env.Namespace().String(), + ScheduleId: sid, + Target: adminservice.MigrateScheduleRequest_SCHEDULER_TARGET_CHASM, + Identity: "test", + RequestId: testcore.RandomizeStr("request-id"), + }) + require.NoError(t, err) + + // Migration is fully applied once the V1 scheduler workflow completes, so its + // side effects (including any retroactive callback attach) are already + // reflected in the fired workflow's history by this point. + awaitV1SchedulerCompleted(ctx, t, env, sid) + + requireNoOptionsUpdatedEvent(ctx, t, env, runningWfID) +} + +// TestScheduleMigrationV1ToV2_RolloutMigration verifies the dynamic-config +// rollout migration path for a normal, actively-firing schedule, covering two +// properties at once: +// +// 1. It migrates. A continuously-firing schedule must eventually move from V1 +// to V2 once migration is enabled via dynamic config. Regression guard for +// the eligibility-check ordering fix (RefreshBeforeMigrationCheck): the check +// previously read len(RunningWorkflows) before the same run-loop iteration's +// processBuffer() reconciled it, so a busy schedule -- which always started a +// replacement action before any later iteration observed the previous one +// complete -- never saw an idle window and never auto-migrated. +// 2. It stays IDL-safe. Because EnableCHASMSchedulerMigrationWithRunningWorkflows +// defaults to false, migration only proceeds once no fired workflow is +// running, so it never retroactively attaches a Nexus completion callback and +// never injects EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED (which older +// SDKs cannot decode -- see repros/scheduler-migration-bug-evidence.md). +// +// Migration here is driven purely by dynamic config -- no admin MigrateSchedule +// RPC and no migrate-to-chasm signal. Contrast the admin path, which migrates +// even with a workflow still running and is still broken (see +// TestScheduleMigrationV1ToV2_AdminMigratePreservesRunningWorkflowHistory). +func TestScheduleMigrationV1ToV2_RolloutMigration(t *testing.T) { + env := testcore.NewEnv(t, testcore.WithWorkerService("V1 scheduler")) + ctx := testcore.NewContext() + + sid := testcore.RandomizeStr("sched-rollout-migrate") + wid := testcore.RandomizeStr("sched-rollout-migrate-wf") + wt := testcore.RandomizeStr("sched-rollout-migrate-wt") + + // A short-lived workflow that completes well within the interval, so the + // schedule is a continuous stream of quick actions -- the shape that both + // exercises the ordering fix and leaves brief idle windows for the + // (default-off) MigrateWithRunningWorkflows guard to migrate in. + env.SdkWorker().RegisterWorkflowWithOptions(func(ctx workflow.Context) error { + return workflow.Sleep(ctx, 200*time.Millisecond) + }, workflow.RegisterOptions{Name: wt}) + + sched := &schedulepb.Schedule{ + Spec: &schedulepb.ScheduleSpec{Interval: []*schedulepb.IntervalSpec{{Interval: durationpb.New(1 * time.Second)}}}, + Action: startWorkflowAction(env, wid, wt), + } + createV1Schedule(ctx, t, env, sid, sched, nil) + + // Establish it as an actively-firing V1 schedule (at least one action) before + // enabling migration, matching an operator turning on rollout for schedules + // that are already running. + firedWfID := awaitAnyAction(ctx, t, env, sid) + + // Enable automatic migration via dynamic config only. MigrateWithRunningWorkflows + // stays at its default (false). + env.OverrideDynamicConfig(dynamicconfig.EnableChasm, true) + env.OverrideDynamicConfig(dynamicconfig.EnableCHASMSchedulerMigration, true) + env.OverrideDynamicConfig(dynamicconfig.CHASMSchedulerMigrationRolloutPercent, 100) + + // (1) Migration completes, and the resulting V2 schedule exists. + awaitV1SchedulerCompleted(ctx, t, env, sid) + requireV2ScheduleExists(ctx, t, env, sid) + + // (2) The fired workflow never gained the unsupported event: migration waited + // (guard default false) until it was no longer running, so no callback was + // retroactively attached. + requireNoOptionsUpdatedEvent(ctx, t, env, firedWfID) +}