From 9c4f9da3c01e02e3a97646fcbad2cedf49b6a588 Mon Sep 17 00:00:00 2001 From: "alex.stanfield" <13949480+chaptersix@users.noreply.github.com> Date: Wed, 19 Aug 2026 03:55:47 -0500 Subject: [PATCH 1/3] Separate CHASM scheduler buffer planning --- chasm/lib/scheduler/buffer_planner.go | 209 ++++++++++ chasm/lib/scheduler/config.go | 8 + chasm/lib/scheduler/export_test.go | 14 + .../lib/scheduler/internal/buffer_planner.go | 296 ++++++++++++++ .../scheduler/internal/buffer_planner_test.go | 144 +++++++ chasm/lib/scheduler/invoker.go | 8 +- .../invoker_process_buffer_task_test.go | 362 ++++++++++++++++++ chasm/lib/scheduler/invoker_tasks.go | 53 ++- tests/schedule_test.go | 3 + 9 files changed, 1076 insertions(+), 21 deletions(-) create mode 100644 chasm/lib/scheduler/buffer_planner.go create mode 100644 chasm/lib/scheduler/internal/buffer_planner.go create mode 100644 chasm/lib/scheduler/internal/buffer_planner_test.go diff --git a/chasm/lib/scheduler/buffer_planner.go b/chasm/lib/scheduler/buffer_planner.go new file mode 100644 index 00000000000..26f1b687fe1 --- /dev/null +++ b/chasm/lib/scheduler/buffer_planner.go @@ -0,0 +1,209 @@ +package scheduler + +import ( + "maps" + "slices" + "time" + + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + schedulespb "go.temporal.io/server/api/schedule/v1" + "go.temporal.io/server/chasm" + schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal" +) + +type appliedBufferPlan struct { + result processBufferResult + decisions []schedulerinternal.BufferDecision + staleDecisions int64 +} + +func newBufferProcessingSnapshot(invoker *Invoker, scheduler *Scheduler, catchupWindow time.Duration) schedulerinternal.BufferProcessingSnapshot { + state := scheduler.Schedule.GetState() + snapshot := schedulerinternal.BufferProcessingSnapshot{ + Starts: make([]schedulerinternal.BufferedStartSnapshot, 0, len(invoker.GetBufferedStarts())), + DefaultOverlapPolicy: scheduler.overlapPolicy(), + CatchupWindow: catchupWindow, + MinimumCatchupWindow: startWorkflowMinDeadline, + Paused: state.GetPaused(), + LimitedActions: state.GetLimitedActions(), + RemainingActions: state.GetRemainingActions(), + } + for _, start := range invoker.GetBufferedStarts() { + projected := projectBufferedStart(start) + snapshot.Starts = append(snapshot.Starts, projected) + if projected.RunID != "" && !projected.Completed { + snapshot.RunningWorkflows = append(snapshot.RunningWorkflows, schedulerinternal.WorkflowExecutionSnapshot{ + WorkflowID: projected.WorkflowID, + RunID: projected.RunID, + }) + } + } + return snapshot +} + +func projectBufferedStart(start *schedulespb.BufferedStart) schedulerinternal.BufferedStartSnapshot { + return schedulerinternal.BufferedStartSnapshot{ + RequestID: start.GetRequestId(), + WorkflowID: start.GetWorkflowId(), + RunID: start.GetRunId(), + Attempt: start.GetAttempt(), + Manual: start.GetManual(), + OverlapPolicy: start.GetOverlapPolicy(), + ActualTime: start.GetActualTime().AsTime(), + DesiredTime: start.GetDesiredTime().AsTime(), + Completed: start.GetCompleted() != nil, + } +} + +func applyBufferPlan( + ctx chasm.MutableContext, + scheduler *Scheduler, + invoker *Invoker, + plan schedulerinternal.BufferPlan, +) appliedBufferPlan { + applied := newAppliedBufferPlan() + currentSnapshot := newBufferProcessingSnapshot(invoker, scheduler, plan.Snapshot.CatchupWindow) + if !bufferProcessingSnapshotsEqual(plan.Snapshot, currentSnapshot) { + return staleBufferPlan(plan, applied) + } + startsByRequestID := make(map[string][]*schedulespb.BufferedStart) + for _, start := range invoker.GetBufferedStarts() { + startsByRequestID[start.GetRequestId()] = append(startsByRequestID[start.GetRequestId()], start) + } + + for _, decision := range plan.Decisions { + if !decision.MutatesState() { + applied.decisions = append(applied.decisions, decision) + continue + } + start, ok := popMatchingBufferedStart(startsByRequestID, decision) + if !ok { + applied.recordStaleDecision(decision) + continue + } + + if decision.Action == schedulerinternal.BufferDecisionExecute && !start.GetManual() && !scheduler.consumeScheduledAction() { + applied.recordStaleDecision(decision) + continue + } + + applyBufferDecision(&applied.result, decision, start) + applied.decisions = append(applied.decisions, decision) + } + if applied.staleDecisions == 0 { + applied.result.overlapSkipped = plan.OverlapSkipped + applied.result.overlapSkippedByPolicy = maps.Clone(plan.OverlapSkippedByPolicy) + } + + for _, target := range plan.TerminateWorkflows { + if currentRunningWorkflow(invoker, target) { + applied.result.terminateWorkflows = append(applied.result.terminateWorkflows, workflowExecutionFromSnapshot(target)) + } + } + for _, target := range plan.CancelWorkflows { + if currentRunningWorkflow(invoker, target) { + applied.result.cancelWorkflows = append(applied.result.cancelWorkflows, workflowExecutionFromSnapshot(target)) + } + } + + var totalMissedCatchup int64 + for _, count := range applied.result.missedCatchupByActionRunning { + totalMissedCatchup += count + } + scheduler.recordActionResult(&schedulerActionResult{ + overlapSkipped: applied.result.overlapSkipped, + missedCatchupWindow: totalMissedCatchup, + }) + invoker.recordProcessBufferResult(ctx, &applied.result) + return applied +} + +func newAppliedBufferPlan() appliedBufferPlan { + return appliedBufferPlan{result: processBufferResult{ + overlapSkippedByPolicy: make(map[enumspb.ScheduleOverlapPolicy]int64), + missedCatchupByActionRunning: make(map[bool]int64), + processedStarts: make(map[string]bool), + }} +} + +func staleBufferPlan(plan schedulerinternal.BufferPlan, applied appliedBufferPlan) appliedBufferPlan { + for _, decision := range plan.Decisions { + if decision.MutatesState() { + applied.recordStaleDecision(decision) + } else { + applied.decisions = append(applied.decisions, decision) + } + } + return applied +} + +func (a *appliedBufferPlan) recordStaleDecision(decision schedulerinternal.BufferDecision) { + decision.Action = schedulerinternal.BufferDecisionRetain + decision.Reason = schedulerinternal.BufferDecisionReasonStale + a.staleDecisions++ + a.decisions = append(a.decisions, decision) +} + +func popMatchingBufferedStart( + startsByRequestID map[string][]*schedulespb.BufferedStart, + decision schedulerinternal.BufferDecision, +) (*schedulespb.BufferedStart, bool) { + matches := startsByRequestID[decision.RequestID] + for index, start := range matches { + if projectBufferedStart(start) == decision.Expected { + startsByRequestID[decision.RequestID] = append(matches[:index], matches[index+1:]...) + return start, true + } + } + return nil, false +} + +func applyBufferDecision(result *processBufferResult, decision schedulerinternal.BufferDecision, start *schedulespb.BufferedStart) { + result.processedStarts[decision.RequestID] = true + switch decision.Action { + case schedulerinternal.BufferDecisionExecute: + result.startWorkflows = append(result.startWorkflows, start) + case schedulerinternal.BufferDecisionDiscard: + result.discardStarts = append(result.discardStarts, start) + recordAppliedDiscard(result, decision) + default: + } +} + +func recordAppliedDiscard(result *processBufferResult, decision schedulerinternal.BufferDecision) { + switch decision.Reason { + case schedulerinternal.BufferDecisionReasonMissedCatchupWindow: + result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedMissedCatchup) + case schedulerinternal.BufferDecisionReasonPausedOrLimited: + result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedPausedOrLimited) + default: + } + if decision.MissedCatchupMetric { + result.missedCatchupByActionRunning[decision.MissedCatchupActionRunning]++ + } +} + +func bufferProcessingSnapshotsEqual(left, right schedulerinternal.BufferProcessingSnapshot) bool { + return left.DefaultOverlapPolicy == right.DefaultOverlapPolicy && + left.CatchupWindow == right.CatchupWindow && + left.MinimumCatchupWindow == right.MinimumCatchupWindow && + left.Paused == right.Paused && + left.LimitedActions == right.LimitedActions && + left.RemainingActions == right.RemainingActions && + slices.Equal(left.Starts, right.Starts) && + slices.Equal(left.RunningWorkflows, right.RunningWorkflows) +} + +func currentRunningWorkflow(invoker *Invoker, target schedulerinternal.WorkflowExecutionSnapshot) bool { + for _, start := range invoker.GetBufferedStarts() { + if start.GetWorkflowId() == target.WorkflowID && start.GetRunId() == target.RunID && start.GetCompleted() == nil { + return true + } + } + return false +} + +func workflowExecutionFromSnapshot(execution schedulerinternal.WorkflowExecutionSnapshot) *commonpb.WorkflowExecution { + return &commonpb.WorkflowExecution{WorkflowId: execution.WorkflowID, RunId: execution.RunID} +} diff --git a/chasm/lib/scheduler/config.go b/chasm/lib/scheduler/config.go index f90ce47af4d..14bcc895838 100644 --- a/chasm/lib/scheduler/config.go +++ b/chasm/lib/scheduler/config.go @@ -29,6 +29,7 @@ type ( ServiceCallTimeout dynamicconfig.DurationPropertyFn RetryPolicy func() backoff.RetryPolicy EncodeInternalTokenWithEnvelope dynamicconfig.BoolPropertyFnWithNamespaceFilter + EnableBufferPlanner dynamicconfig.BoolPropertyFnWithNamespaceFilter } ) @@ -80,6 +81,12 @@ var ( `The upper bound on how long a service call can take before being timed out.`, ) + EnableBufferPlanner = dynamicconfig.NewNamespaceBoolSetting( + "scheduler.enableBufferPlanner", + false, + `Use the pure planner to process CHASM scheduler buffered starts. The legacy processor remains the fallback while this setting is disabled.`, + ) + // SentinelIdleTime is how long a CHASM sentinel reserves the schedule ID // before auto-closing via the idle task mechanism. Matches the dummy // workflow's duration. @@ -104,6 +111,7 @@ func ConfigProvider(dc *dynamicconfig.Collection) *Config { Tweakables: CurrentTweakables.Get(dc), ServiceCallTimeout: ServiceCallTimeout.Get(dc), EncodeInternalTokenWithEnvelope: callback.EncodeInternalTokenWithEnvelope.Get(dc), + EnableBufferPlanner: EnableBufferPlanner.Get(dc), RetryPolicy: func() backoff.RetryPolicy { return backoff.NewExponentialRetryPolicy( RetryPolicyInitialInterval.Get(dc)(), diff --git a/chasm/lib/scheduler/export_test.go b/chasm/lib/scheduler/export_test.go index 8880a4efd63..286be6943da 100644 --- a/chasm/lib/scheduler/export_test.go +++ b/chasm/lib/scheduler/export_test.go @@ -7,6 +7,7 @@ import ( schedulespb "go.temporal.io/server/api/schedule/v1" "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1" + schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal" "go.temporal.io/server/common/log" legacyscheduler "go.temporal.io/server/service/worker/scheduler" ) @@ -72,6 +73,19 @@ func (i *Invoker) RecordExecuteResult( }) } +func (h *InvokerProcessBufferTaskHandler) PlanBufferProcessingForTest( + invoker *Invoker, + scheduler *Scheduler, + now time.Time, +) func(chasm.MutableContext, *Scheduler, *Invoker) int64 { + tweakables := h.config.Tweakables(scheduler.Namespace) + snapshot := newBufferProcessingSnapshot(invoker, scheduler, catchupWindow(scheduler, tweakables)) + plan := schedulerinternal.PlanBufferProcessing(snapshot, now) + return func(ctx chasm.MutableContext, scheduler *Scheduler, invoker *Invoker) int64 { + return applyBufferPlan(ctx, scheduler, invoker, plan).staleDecisions + } +} + func (b *BackfillerTaskHandler) ProcessBackfill( scheduler *Scheduler, backfiller *Backfiller, diff --git a/chasm/lib/scheduler/internal/buffer_planner.go b/chasm/lib/scheduler/internal/buffer_planner.go new file mode 100644 index 00000000000..958e6d47a47 --- /dev/null +++ b/chasm/lib/scheduler/internal/buffer_planner.go @@ -0,0 +1,296 @@ +package internal + +import ( + "slices" + "time" + + enumspb "go.temporal.io/api/enums/v1" +) + +type BufferDecisionAction int + +const ( + BufferDecisionRetain BufferDecisionAction = iota + BufferDecisionExecute + BufferDecisionDiscard + BufferDecisionDefer + BufferDecisionRetry + BufferDecisionRunning + BufferDecisionCompleted +) + +type BufferDecisionReason int + +const ( + BufferDecisionReasonNone BufferDecisionReason = iota + BufferDecisionReasonAlreadyProcessed + BufferDecisionReasonOverlapPolicy + BufferDecisionReasonCancelOrTerminate + BufferDecisionReasonMissedCatchupWindow + BufferDecisionReasonPausedOrLimited + BufferDecisionReasonStale +) + +type BufferedStartSnapshot struct { + RequestID string + WorkflowID string + RunID string + Attempt int64 + Manual bool + OverlapPolicy enumspb.ScheduleOverlapPolicy + ActualTime time.Time + DesiredTime time.Time + Completed bool +} + +func (s BufferedStartSnapshot) GetOverlapPolicy() enumspb.ScheduleOverlapPolicy { + return s.OverlapPolicy +} + +type WorkflowExecutionSnapshot struct { + WorkflowID string + RunID string +} + +type BufferProcessingSnapshot struct { + Starts []BufferedStartSnapshot + RunningWorkflows []WorkflowExecutionSnapshot + DefaultOverlapPolicy enumspb.ScheduleOverlapPolicy + CatchupWindow time.Duration + MinimumCatchupWindow time.Duration + Paused bool + LimitedActions bool + RemainingActions int64 +} + +type BufferDecision struct { + RequestID string + Expected BufferedStartSnapshot + Action BufferDecisionAction + Reason BufferDecisionReason + OverlapPolicy enumspb.ScheduleOverlapPolicy + OverlapSkipped bool + MissedCatchupMetric bool + MissedCatchupActionRunning bool +} + +func (d BufferDecision) MutatesState() bool { + return d.Action == BufferDecisionExecute || d.Action == BufferDecisionDiscard || d.Action == BufferDecisionDefer +} + +type BufferPlan struct { + Snapshot BufferProcessingSnapshot + Decisions []BufferDecision + CancelWorkflows []WorkflowExecutionSnapshot + TerminateWorkflows []WorkflowExecutionSnapshot + OverlapSkipped int64 + OverlapSkippedByPolicy map[enumspb.ScheduleOverlapPolicy]int64 + MissedCatchupByActionRunning map[bool]int64 +} + +type overlapAction struct { + OverlappingStarts []BufferedStartSnapshot + NonOverlappingStart BufferedStartSnapshot + NewBuffer []BufferedStartSnapshot + NeedCancel bool + NeedTerminate bool + OverlapSkipped int64 + OverlapSkippedByPolicy map[enumspb.ScheduleOverlapPolicy]int64 +} + +func PlanBufferProcessing(snapshot BufferProcessingSnapshot, now time.Time) BufferPlan { + plan := BufferPlan{ + Snapshot: cloneBufferProcessingSnapshot(snapshot), + OverlapSkippedByPolicy: make(map[enumspb.ScheduleOverlapPolicy]int64), + MissedCatchupByActionRunning: make(map[bool]int64), + } + resolveOverlapPolicy := func(policy enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy { + if policy == enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED { + return snapshot.DefaultOverlapPolicy + } + return policy + } + + pending := make([]BufferedStartSnapshot, 0, len(snapshot.Starts)) + for _, start := range snapshot.Starts { + if start.Attempt == 0 || + (start.Attempt == -1 && resolveOverlapPolicy(start.OverlapPolicy) == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE) { + pending = append(pending, start) + } + } + + action := planOverlapActions(pending, len(snapshot.RunningWorkflows) > 0, resolveOverlapPolicy) + plan.OverlapSkipped = action.OverlapSkipped + plan.OverlapSkippedByPolicy = action.OverlapSkippedByPolicy + if action.NeedTerminate { + plan.TerminateWorkflows = append(plan.TerminateWorkflows, snapshot.RunningWorkflows...) + } else if action.NeedCancel { + plan.CancelWorkflows = append(plan.CancelWorkflows, snapshot.RunningWorkflows...) + } + + deferred := make(map[string]struct{}, len(action.NewBuffer)) + for _, start := range action.NewBuffer { + deferred[start.RequestID] = struct{}{} + } + ready := make(map[string]struct{}, len(action.OverlappingStarts)+1) + for _, start := range action.OverlappingStarts { + ready[start.RequestID] = struct{}{} + } + if action.NonOverlappingStart.RequestID != "" { + ready[action.NonOverlappingStart.RequestID] = struct{}{} + } + + remainingActions := snapshot.RemainingActions + for _, start := range snapshot.Starts { + isPending := start.Attempt == 0 || + (start.Attempt == -1 && resolveOverlapPolicy(start.OverlapPolicy) == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE) + if !isPending { + plan.Decisions = append(plan.Decisions, alreadyProcessedBufferDecision(start)) + continue + } + decision := planPendingBufferDecision( + start, + snapshot, + now, + deferred, + ready, + action.NeedCancel || action.NeedTerminate, + resolveOverlapPolicy, + &remainingActions, + ) + if decision.MissedCatchupMetric { + plan.MissedCatchupByActionRunning[decision.MissedCatchupActionRunning]++ + } + plan.Decisions = append(plan.Decisions, decision) + } + return plan +} + +func planOverlapActions( + buffer []BufferedStartSnapshot, + isRunning bool, + resolve func(enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy, +) overlapAction { + action := overlapAction{OverlapSkippedByPolicy: make(map[enumspb.ScheduleOverlapPolicy]int64)} + for _, start := range buffer { + overlapPolicy := resolve(start.OverlapPolicy) + if overlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL { + action.OverlappingStarts = append(action.OverlappingStarts, start) + continue + } + if !isRunning && action.NonOverlappingStart.RequestID == "" { + action.NonOverlappingStart = start + continue + } + switch overlapPolicy { + case enumspb.SCHEDULE_OVERLAP_POLICY_SKIP: + action.OverlapSkipped++ + action.OverlapSkippedByPolicy[overlapPolicy]++ + case enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE: + if len(action.NewBuffer) == 0 { + action.NewBuffer = append(action.NewBuffer, start) + } else { + action.OverlapSkipped++ + action.OverlapSkippedByPolicy[overlapPolicy]++ + } + case enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL: + action.NewBuffer = append(action.NewBuffer, start) + case enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER: + if isRunning { + action.NeedCancel = true + action.NewBuffer = append(action.NewBuffer, start) + } else { + action.NonOverlappingStart = start + } + case enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER: + if isRunning { + action.NeedTerminate = true + action.NewBuffer = append(action.NewBuffer, start) + } else { + action.NonOverlappingStart = start + } + default: + } + } + if action.NeedCancel || action.NeedTerminate { + action.OverlappingStarts = nil + } + return action +} + +func alreadyProcessedBufferDecision(start BufferedStartSnapshot) BufferDecision { + decision := BufferDecision{ + RequestID: start.RequestID, + Expected: start, + Action: BufferDecisionRetain, + Reason: BufferDecisionReasonAlreadyProcessed, + } + switch { + case start.Completed: + decision.Action = BufferDecisionCompleted + case start.RunID != "": + decision.Action = BufferDecisionRunning + case start.Attempt > 1: + decision.Action = BufferDecisionRetry + default: + } + return decision +} + +func planPendingBufferDecision( + start BufferedStartSnapshot, + snapshot BufferProcessingSnapshot, + now time.Time, + deferred map[string]struct{}, + ready map[string]struct{}, + needCancelOrTerminate bool, + resolveOverlapPolicy func(enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy, + remainingActions *int64, +) BufferDecision { + decision := BufferDecision{RequestID: start.RequestID, Expected: start} + if _, ok := deferred[start.RequestID]; ok { + decision.Action = BufferDecisionDefer + decision.Reason = BufferDecisionReasonOverlapPolicy + return decision + } + if _, ok := ready[start.RequestID]; !ok { + decision.Action = BufferDecisionDiscard + decision.Reason = BufferDecisionReasonOverlapPolicy + if needCancelOrTerminate { + decision.Reason = BufferDecisionReasonCancelOrTerminate + return decision + } + decision.OverlapPolicy = resolveOverlapPolicy(start.OverlapPolicy) + decision.OverlapSkipped = decision.OverlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_SKIP || + decision.OverlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE + return decision + } + + deadline := start.ActualTime.Add(max(snapshot.CatchupWindow, snapshot.MinimumCatchupWindow)) + canTakeScheduledAction := !snapshot.Paused && (!snapshot.LimitedActions || *remainingActions > 0) + if !start.Manual && now.After(deadline) { + decision.Action = BufferDecisionDiscard + decision.Reason = BufferDecisionReasonMissedCatchupWindow + if canTakeScheduledAction { + decision.MissedCatchupMetric = true + decision.MissedCatchupActionRunning = len(snapshot.RunningWorkflows) > 0 || start.DesiredTime.After(deadline) + } + return decision + } + if !start.Manual && !canTakeScheduledAction { + decision.Action = BufferDecisionDiscard + decision.Reason = BufferDecisionReasonPausedOrLimited + return decision + } + if !start.Manual && snapshot.LimitedActions { + *remainingActions-- + } + decision.Action = BufferDecisionExecute + return decision +} + +func cloneBufferProcessingSnapshot(snapshot BufferProcessingSnapshot) BufferProcessingSnapshot { + snapshot.Starts = slices.Clone(snapshot.Starts) + snapshot.RunningWorkflows = slices.Clone(snapshot.RunningWorkflows) + return snapshot +} diff --git a/chasm/lib/scheduler/internal/buffer_planner_test.go b/chasm/lib/scheduler/internal/buffer_planner_test.go new file mode 100644 index 00000000000..3899089cc61 --- /dev/null +++ b/chasm/lib/scheduler/internal/buffer_planner_test.go @@ -0,0 +1,144 @@ +package internal + +import ( + "reflect" + "testing" + "time" + + "github.com/stretchr/testify/require" + enumspb "go.temporal.io/api/enums/v1" +) + +func TestPlanBufferProcessing_OverlapPolicies(t *testing.T) { + now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC) + tests := []struct { + name string + policy enumspb.ScheduleOverlapPolicy + action BufferDecisionAction + reason BufferDecisionReason + cancelCount int + terminateCount int + overlapSkipped int64 + }{ + {name: "allow all", policy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, action: BufferDecisionExecute}, + {name: "skip", policy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, action: BufferDecisionDiscard, reason: BufferDecisionReasonOverlapPolicy, overlapSkipped: 1}, + {name: "buffer one", policy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, action: BufferDecisionDefer, reason: BufferDecisionReasonOverlapPolicy}, + {name: "buffer all", policy: enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, action: BufferDecisionDefer, reason: BufferDecisionReasonOverlapPolicy}, + {name: "cancel other", policy: enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER, action: BufferDecisionDefer, reason: BufferDecisionReasonOverlapPolicy, cancelCount: 1}, + {name: "terminate other", policy: enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER, action: BufferDecisionDefer, reason: BufferDecisionReasonOverlapPolicy, terminateCount: 1}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + snapshot := BufferProcessingSnapshot{ + Starts: []BufferedStartSnapshot{{ + RequestID: "request", + OverlapPolicy: test.policy, + ActualTime: now, + DesiredTime: now, + }}, + RunningWorkflows: []WorkflowExecutionSnapshot{{WorkflowID: "running", RunID: "run"}}, + DefaultOverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, + CatchupWindow: time.Hour, + MinimumCatchupWindow: 5 * time.Second, + } + + plan := PlanBufferProcessing(snapshot, now) + + require.Len(t, plan.Decisions, 1) + require.Equal(t, test.action, plan.Decisions[0].Action) + require.Equal(t, test.reason, plan.Decisions[0].Reason) + require.Len(t, plan.CancelWorkflows, test.cancelCount) + require.Len(t, plan.TerminateWorkflows, test.terminateCount) + require.Equal(t, test.overlapSkipped, plan.OverlapSkipped) + }) + } +} + +func TestPlanBufferProcessing_TimeAndCapacity(t *testing.T) { + now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC) + snapshot := BufferProcessingSnapshot{ + Starts: []BufferedStartSnapshot{ + {RequestID: "expired", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now.Add(-2 * time.Hour), DesiredTime: now.Add(-2 * time.Hour)}, + {RequestID: "automatic", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now, DesiredTime: now}, + {RequestID: "limited", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now, DesiredTime: now}, + {RequestID: "manual", Manual: true, OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now.Add(-24 * time.Hour), DesiredTime: now}, + }, + DefaultOverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, + CatchupWindow: time.Hour, + MinimumCatchupWindow: 5 * time.Second, + LimitedActions: true, + RemainingActions: 1, + } + + plan := PlanBufferProcessing(snapshot, now) + + require.Equal(t, []BufferDecisionAction{ + BufferDecisionDiscard, + BufferDecisionExecute, + BufferDecisionDiscard, + BufferDecisionExecute, + }, decisionActions(plan.Decisions)) + require.Equal(t, BufferDecisionReasonMissedCatchupWindow, plan.Decisions[0].Reason) + require.Equal(t, BufferDecisionReasonPausedOrLimited, plan.Decisions[2].Reason) + require.Equal(t, int64(1), plan.MissedCatchupByActionRunning[false]) +} + +func TestPlanBufferProcessing_PausedAndProcessedStates(t *testing.T) { + now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC) + snapshot := BufferProcessingSnapshot{ + Starts: []BufferedStartSnapshot{ + {RequestID: "automatic", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now}, + {RequestID: "manual", Manual: true, OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now}, + {RequestID: "backing-off", Attempt: 2, ActualTime: now}, + {RequestID: "running", Attempt: 1, RunID: "run", ActualTime: now}, + {RequestID: "completed", Attempt: 1, RunID: "run", Completed: true, ActualTime: now}, + }, + DefaultOverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, + CatchupWindow: time.Hour, + MinimumCatchupWindow: 5 * time.Second, + Paused: true, + } + + plan := PlanBufferProcessing(snapshot, now) + + require.Equal(t, []BufferDecisionAction{ + BufferDecisionDiscard, + BufferDecisionExecute, + BufferDecisionRetry, + BufferDecisionRunning, + BufferDecisionCompleted, + }, decisionActions(plan.Decisions)) +} + +func TestPlanBufferProcessing_DoesNotMutateOrAliasInput(t *testing.T) { + now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC) + snapshot := BufferProcessingSnapshot{ + Starts: []BufferedStartSnapshot{{ + RequestID: "request", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now, + }}, + DefaultOverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, + CatchupWindow: time.Hour, + MinimumCatchupWindow: 5 * time.Second, + } + want := BufferProcessingSnapshot{ + Starts: append([]BufferedStartSnapshot(nil), snapshot.Starts...), + DefaultOverlapPolicy: snapshot.DefaultOverlapPolicy, + CatchupWindow: snapshot.CatchupWindow, + MinimumCatchupWindow: snapshot.MinimumCatchupWindow, + } + + plan := PlanBufferProcessing(snapshot, now) + + require.True(t, reflect.DeepEqual(want, snapshot)) + plan.Decisions[0].Expected.RequestID = "changed" + plan.Snapshot.Starts[0].RequestID = "changed-again" + require.Equal(t, "request", snapshot.Starts[0].RequestID) +} + +func decisionActions(decisions []BufferDecision) []BufferDecisionAction { + actions := make([]BufferDecisionAction, 0, len(decisions)) + for _, decision := range decisions { + actions = append(actions, decision.Action) + } + return actions +} diff --git a/chasm/lib/scheduler/invoker.go b/chasm/lib/scheduler/invoker.go index 91bdc663e16..b7277e26c30 100644 --- a/chasm/lib/scheduler/invoker.go +++ b/chasm/lib/scheduler/invoker.go @@ -12,6 +12,7 @@ import ( "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1" schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal" + "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/util" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -93,6 +94,11 @@ type processBufferResult struct { // Number of buffered starts dropped from missing the catchup window, // bucketed by whether a running action contributed to the miss. missedCatchupByActionRunning map[bool]int64 + bufferedStartDropReasons []metrics.ReasonString + + // processedStarts limits lifecycle transitions to identities covered by a + // revalidated plan. A nil map preserves the legacy processor's behavior. + processedStarts map[string]bool } // recordProcessBufferResult updates the Invoker's internal state based on result, as well as the @@ -120,7 +126,7 @@ func (i *Invoker) recordProcessBufferResult(ctx chasm.MutableContext, result *pr if ready[start.RequestId] && start.Attempt < 1 { schedulerinternal.MarkStartReady(start) readiedStarts++ - } else if start.Attempt == 0 { + } else if start.Attempt == 0 && (result.processedStarts == nil || result.processedStarts[start.RequestId]) { // Start was processed but deferred (e.g., BUFFER_ONE policy with running workflow). // Mark as deferred (-1) to distinguish from newly-enqueued starts so addTasks // won't schedule an immediate ProcessBuffer task for them - they wait on diff --git a/chasm/lib/scheduler/invoker_process_buffer_task_test.go b/chasm/lib/scheduler/invoker_process_buffer_task_test.go index 87735003005..bf3e22c9492 100644 --- a/chasm/lib/scheduler/invoker_process_buffer_task_test.go +++ b/chasm/lib/scheduler/invoker_process_buffer_task_test.go @@ -2,13 +2,17 @@ package scheduler_test import ( "context" + "fmt" "slices" + "sort" + "strings" "testing" "time" "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" schedulespb "go.temporal.io/server/api/schedule/v1" "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/chasmtest" @@ -16,8 +20,10 @@ import ( "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1" "go.temporal.io/server/common/clock" "go.temporal.io/server/common/metrics" + "go.temporal.io/server/common/metrics/metricstest" "go.temporal.io/server/common/util" "go.temporal.io/server/service/history/tasks" + "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/durationpb" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -678,6 +684,362 @@ func TestProcessBufferTask_PausedDropsAutomatedKeepsManual(t *testing.T) { }) } +type bufferProcessingComparisonInput struct { + name string + schedule *schedulepb.Schedule + bufferedStarts []*schedulespb.BufferedStart + cancelWorkflows []*commonpb.WorkflowExecution + terminateWorkflows []*commonpb.WorkflowExecution + lastProcessedTime time.Time + workflowMigration bool + initialConflictToken int64 +} + +type normalizedBufferProcessing struct { + schedulerState *schedulerpb.SchedulerState + invokerState *schedulerpb.InvokerState + tasks []string + actionRequests []string + metrics []string + outcomes []string + remainingDelta int64 + conflictTokenDelta int64 +} + +func TestProcessBufferTask_LegacyAndPlannerDifferentialCorpus(t *testing.T) { + now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC) + makeStart := func(id string, policy enumspb.ScheduleOverlapPolicy) *schedulespb.BufferedStart { + return &schedulespb.BufferedStart{ + NominalTime: timestamppb.New(now), + ActualTime: timestamppb.New(now), + DesiredTime: timestamppb.New(now), + RequestId: id, + WorkflowId: "workflow-" + id, + OverlapPolicy: policy, + } + } + running := makeStart("running", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + running.Attempt = 1 + running.RunId = "running-run" + + var corpus []bufferProcessingComparisonInput + for _, policy := range []enumspb.ScheduleOverlapPolicy{ + enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, + enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, + enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, + enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER, + enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER, + enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, + } { + corpus = append(corpus, bufferProcessingComparisonInput{ + name: "overlap " + policy.String(), + schedule: defaultSchedule(), + bufferedStarts: []*schedulespb.BufferedStart{running, makeStart("pending", policy)}, + lastProcessedTime: now, + }) + } + + pausedSchedule := defaultSchedule() + pausedSchedule.State.Paused = true + pausedManual := makeStart("manual", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + pausedManual.Manual = true + corpus = append(corpus, bufferProcessingComparisonInput{ + name: "paused automatic and manual", schedule: pausedSchedule, + bufferedStarts: []*schedulespb.BufferedStart{ + makeStart("automatic", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL), pausedManual, + }, lastProcessedTime: now, + }) + for _, remaining := range []int64{0, 1, 3} { + limited := defaultSchedule() + limited.State.LimitedActions = true + limited.State.RemainingActions = remaining + corpus = append(corpus, bufferProcessingComparisonInput{ + name: fmt.Sprintf("limited actions %d", remaining), schedule: limited, + bufferedStarts: []*schedulespb.BufferedStart{ + makeStart("first", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL), + makeStart("second", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL), + }, lastProcessedTime: now, initialConflictToken: 17, + }) + } + unlimitedManual := makeStart("manual", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + unlimitedManual.Manual = true + corpus = append(corpus, bufferProcessingComparisonInput{ + name: "unlimited and manual", schedule: defaultSchedule(), + bufferedStarts: []*schedulespb.BufferedStart{ + makeStart("automatic", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL), unlimitedManual, + }, lastProcessedTime: now, + }) + expired := makeStart("expired", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + expired.ActualTime = timestamppb.New(now.Add(-2 * defaultCatchupWindow)) + corpus = append(corpus, bufferProcessingComparisonInput{ + name: "expired catchup window", schedule: defaultSchedule(), + bufferedStarts: []*schedulespb.BufferedStart{expired}, lastProcessedTime: now, + }) + deferred := makeStart("deferred", enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE) + deferred.Attempt = -1 + corpus = append(corpus, bufferProcessingComparisonInput{ + name: "running and deferred", schedule: defaultSchedule(), + bufferedStarts: []*schedulespb.BufferedStart{running, deferred}, lastProcessedTime: now, + }) + for _, backoffDelta := range []time.Duration{-time.Second, 0, time.Second} { + retry := makeStart(fmt.Sprintf("retry-%s", backoffDelta), enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + retry.Attempt = 2 + retry.BackoffTime = timestamppb.New(now.Add(backoffDelta)) + corpus = append(corpus, bufferProcessingComparisonInput{ + name: fmt.Sprintf("retry boundary %s", backoffDelta), schedule: defaultSchedule(), + bufferedStarts: []*schedulespb.BufferedStart{retry}, lastProcessedTime: now, + }) + } + completed := makeStart("duplicate", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + completed.Attempt = 1 + completed.RunId = "completed-run" + completed.Completed = &schedulespb.CompletedResult{} + corpus = append(corpus, bufferProcessingComparisonInput{ + name: "duplicate and completed entries", schedule: defaultSchedule(), + bufferedStarts: []*schedulespb.BufferedStart{completed, proto.Clone(completed).(*schedulespb.BufferedStart)}, + lastProcessedTime: now, + }) + duplicatePending := makeStart("duplicate-pending", enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE) + corpus = append(corpus, bufferProcessingComparisonInput{ + name: "duplicate pending entries", schedule: defaultSchedule(), + bufferedStarts: []*schedulespb.BufferedStart{ + running, + duplicatePending, + proto.Clone(duplicatePending).(*schedulespb.BufferedStart), + }, + lastProcessedTime: now, + }) + corpus = append(corpus, + bufferProcessingComparisonInput{ + name: "existing cancel work", schedule: defaultSchedule(), + bufferedStarts: []*schedulespb.BufferedStart{running}, + cancelWorkflows: []*commonpb.WorkflowExecution{{WorkflowId: "cancel", RunId: "cancel-run"}}, + lastProcessedTime: now, + }, + bufferProcessingComparisonInput{ + name: "existing terminate work", schedule: defaultSchedule(), + bufferedStarts: []*schedulespb.BufferedStart{running}, + terminateWorkflows: []*commonpb.WorkflowExecution{{WorkflowId: "terminate", RunId: "terminate-run"}}, + lastProcessedTime: now, + }, + bufferProcessingComparisonInput{ + name: "migration state", schedule: defaultSchedule(), + bufferedStarts: []*schedulespb.BufferedStart{makeStart("migration", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL)}, + lastProcessedTime: now, workflowMigration: true, + }, + ) + completionRace := makeStart("completion-race", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + completionRace.Attempt = 1 + completionRace.Completed = &schedulespb.CompletedResult{} + corpus = append(corpus, bufferProcessingComparisonInput{ + name: "completion before start result", schedule: defaultSchedule(), + bufferedStarts: []*schedulespb.BufferedStart{completionRace}, lastProcessedTime: now, + }) + + for _, input := range corpus { + t.Run(input.name, func(t *testing.T) { + compareBufferProcessing(t, input) + }) + } +} + +func TestApplyBufferPlan_RevalidatesRequestBeforeMutation(t *testing.T) { + schedule := defaultSchedule() + schedule.State.LimitedActions = true + schedule.State.RemainingActions = 1 + env := newSchedulerTestEngine(t, schedule) + now := env.timeSource.Now() + config := defaultConfig() + handler := scheduler.NewInvokerProcessBufferTaskHandler(scheduler.InvokerTaskHandlerOptions{ + Config: config, MetricsHandler: metrics.NoopMetricsHandler, BaseLogger: env.logger, + }) + var apply func(chasm.MutableContext, *scheduler.Scheduler, *scheduler.Invoker) int64 + var initialConflictToken int64 + require.NoError(t, env.updateScheduler(func(s *scheduler.Scheduler, ctx chasm.MutableContext) error { + invoker := s.Invoker.Get(ctx) + invoker.BufferedStarts = []*schedulespb.BufferedStart{{ + NominalTime: timestamppb.New(now), ActualTime: timestamppb.New(now), DesiredTime: timestamppb.New(now), + RequestId: "request", WorkflowId: "workflow", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, + }} + invoker.LastProcessedTime = timestamppb.New(now) + apply = handler.PlanBufferProcessingForTest(invoker, s, now) + initialConflictToken = s.ConflictToken + return nil + })) + + var staleDecisions int64 + require.NoError(t, env.updateScheduler(func(s *scheduler.Scheduler, ctx chasm.MutableContext) error { + invoker := s.Invoker.Get(ctx) + invoker.BufferedStarts[0].Attempt = 1 + staleDecisions = apply(ctx, s, invoker) + return nil + })) + + require.Equal(t, int64(1), staleDecisions) + require.NoError(t, env.readScheduler(func(s *scheduler.Scheduler, ctx chasm.Context) error { + require.Equal(t, int64(1), s.Invoker.Get(ctx).BufferedStarts[0].GetAttempt()) + require.Equal(t, int64(1), s.Schedule.State.GetRemainingActions()) + require.Equal(t, initialConflictToken, s.ConflictToken) + return nil + })) +} + +func compareBufferProcessing(t *testing.T, input bufferProcessingComparisonInput) { + t.Helper() + legacy := runBufferProcessing(t, input, false) + planned := runBufferProcessing(t, input, true) + require.True(t, proto.Equal(legacy.schedulerState, planned.schedulerState), "Scheduler state mismatch\nlegacy: %v\nplanned: %v", legacy.schedulerState, planned.schedulerState) + require.True(t, proto.Equal(legacy.invokerState, planned.invokerState), "Invoker state mismatch\nlegacy: %v\nplanned: %v", legacy.invokerState, planned.invokerState) + require.Equal(t, legacy.tasks, planned.tasks) + require.Equal(t, legacy.actionRequests, planned.actionRequests) + require.Equal(t, legacy.metrics, planned.metrics) + require.Equal(t, legacy.outcomes, planned.outcomes) + require.Equal(t, legacy.remainingDelta, planned.remainingDelta) + require.Equal(t, legacy.conflictTokenDelta, planned.conflictTokenDelta) +} + +func runBufferProcessing(t *testing.T, input bufferProcessingComparisonInput, planner bool) normalizedBufferProcessing { + t.Helper() + env := newTestEnv(t) + env.TimeSource.Update(input.lastProcessedTime) + ctx := env.MutableContext() + env.Scheduler.Schedule = proto.Clone(input.schedule).(*schedulepb.Schedule) + env.Scheduler.Info.CreateTime = timestamppb.New(input.lastProcessedTime) + initialConflictToken := input.initialConflictToken + if initialConflictToken == 0 { + initialConflictToken = env.Scheduler.ConflictToken + } + env.Scheduler.ConflictToken = initialConflictToken + if input.workflowMigration { + env.Scheduler.WorkflowMigration = &schedulerpb.WorkflowMigrationState{} + } + invoker := env.Scheduler.Invoker.Get(ctx) + invoker.InvokerState = &schedulerpb.InvokerState{ + BufferedStarts: cloneBufferedStarts(input.bufferedStarts), + CancelWorkflows: cloneWorkflowExecutions(input.cancelWorkflows), + TerminateWorkflows: cloneWorkflowExecutions(input.terminateWorkflows), + LastProcessedTime: timestamppb.New(input.lastProcessedTime), + } + env.NodeBackend.TasksByCategory = nil + + config := defaultConfig() + config.EnableBufferPlanner = func(string) bool { return planner } + recorder := metricstest.NewCaptureHandler() + capture := recorder.StartCapture() + defer recorder.StopCapture(capture) + handler := scheduler.NewInvokerProcessBufferTaskHandler(scheduler.InvokerTaskHandlerOptions{ + Config: config, MetricsHandler: recorder, BaseLogger: env.Logger, + }) + initialRemaining := env.Scheduler.Schedule.State.GetRemainingActions() + require.NoError(t, handler.Execute(ctx, invoker, chasm.TaskAttributes{}, &schedulerpb.InvokerProcessBufferTask{})) + require.NoError(t, env.CloseTransaction()) + + return normalizedBufferProcessing{ + schedulerState: proto.Clone(env.Scheduler.SchedulerState).(*schedulerpb.SchedulerState), + invokerState: proto.Clone(invoker.InvokerState).(*schedulerpb.InvokerState), + tasks: normalizeTasks(env.NodeBackend.TasksByCategory), + actionRequests: normalizeActionRequests(invoker), + metrics: normalizeMetrics(capture.Snapshot()), + outcomes: normalizeBufferOutcomes(input.bufferedStarts, invoker.GetBufferedStarts()), + remainingDelta: env.Scheduler.Schedule.State.GetRemainingActions() - initialRemaining, + conflictTokenDelta: env.Scheduler.ConflictToken - initialConflictToken, + } +} + +func cloneBufferedStarts(starts []*schedulespb.BufferedStart) []*schedulespb.BufferedStart { + cloned := make([]*schedulespb.BufferedStart, 0, len(starts)) + for _, start := range starts { + cloned = append(cloned, proto.Clone(start).(*schedulespb.BufferedStart)) + } + return cloned +} + +func cloneWorkflowExecutions(executions []*commonpb.WorkflowExecution) []*commonpb.WorkflowExecution { + cloned := make([]*commonpb.WorkflowExecution, 0, len(executions)) + for _, execution := range executions { + cloned = append(cloned, proto.Clone(execution).(*commonpb.WorkflowExecution)) + } + return cloned +} + +func normalizeTasks(tasksByCategory map[tasks.Category][]tasks.Task) []string { + var categories []tasks.Category + for category := range tasksByCategory { + categories = append(categories, category) + } + sort.Slice(categories, func(i, j int) bool { return categories[i].ID() < categories[j].ID() }) + var normalized []string + for _, category := range categories { + for index, task := range tasksByCategory[category] { + identity := "" + if chasmTask, ok := task.(*tasks.ChasmTask); ok { + identity = fmt.Sprintf("path=%s,type=%d,data=%x", strings.Join(chasmTask.Info.GetPath(), "/"), chasmTask.Info.GetTypeId(), chasmTask.Info.GetData().GetData()) + } + normalized = append(normalized, fmt.Sprintf("%d:%d:%s:%s:%s", category.ID(), index, task.GetType(), task.GetVisibilityTime().UTC().Format(time.RFC3339Nano), identity)) + } + } + return normalized +} + +func normalizeActionRequests(invoker *scheduler.Invoker) []string { + requests := make([]string, 0, len(invoker.GetTerminateWorkflows())+len(invoker.GetCancelWorkflows())) + for _, execution := range invoker.GetTerminateWorkflows() { + requests = append(requests, "terminate:"+execution.GetWorkflowId()+":"+execution.GetRunId()) + } + for _, execution := range invoker.GetCancelWorkflows() { + requests = append(requests, "cancel:"+execution.GetWorkflowId()+":"+execution.GetRunId()) + } + return requests +} + +func normalizeMetrics(snapshot metricstest.CaptureSnapshot) []string { + var normalized []string + for name, recordings := range snapshot { + for _, recording := range recordings { + var tags []string + for key, value := range recording.Tags { + tags = append(tags, key+"="+value) + } + sort.Strings(tags) + normalized = append(normalized, fmt.Sprintf("%s:%v:%s", name, recording.Value, strings.Join(tags, ","))) + } + } + sort.Strings(normalized) + return normalized +} + +func normalizeBufferOutcomes(before, after []*schedulespb.BufferedStart) []string { + afterByRequestID := make(map[string][]*schedulespb.BufferedStart) + for _, start := range after { + afterByRequestID[start.GetRequestId()] = append(afterByRequestID[start.GetRequestId()], start) + } + outcomes := make([]string, 0, len(before)) + for _, start := range before { + matches := afterByRequestID[start.GetRequestId()] + if len(matches) == 0 { + outcomes = append(outcomes, start.GetRequestId()+":discard") + continue + } + current := matches[0] + afterByRequestID[start.GetRequestId()] = matches[1:] + switch { + case start.GetAttempt() == 0 && current.GetAttempt() > 0: + outcomes = append(outcomes, start.GetRequestId()+":execute") + case current.GetAttempt() == -1: + outcomes = append(outcomes, start.GetRequestId()+":defer") + case current.GetCompleted() != nil: + outcomes = append(outcomes, start.GetRequestId()+":completed") + case current.GetRunId() != "": + outcomes = append(outcomes, start.GetRequestId()+":running") + case current.GetAttempt() > 1: + outcomes = append(outcomes, start.GetRequestId()+":retry") + default: + outcomes = append(outcomes, start.GetRequestId()+":retain") + } + } + return outcomes +} + // A buffered start with an overlap policy to cancel other workflows is processed. func TestProcessBufferTask_NeedsCancel(t *testing.T) { env := newTestEnv(t) diff --git a/chasm/lib/scheduler/invoker_tasks.go b/chasm/lib/scheduler/invoker_tasks.go index fae04d93329..ff595df7a06 100644 --- a/chasm/lib/scheduler/invoker_tasks.go +++ b/chasm/lib/scheduler/invoker_tasks.go @@ -460,18 +460,38 @@ func (h *InvokerProcessBufferTaskHandler) Execute( return queueerrors.NewUnprocessableTaskError("schedules must have an Action set") } - // Compute actions to take from the current buffer. - result := h.processBuffer(ctx, invoker, scheduler) + var result processBufferResult + if h.config.EnableBufferPlanner != nil && h.config.EnableBufferPlanner(scheduler.Namespace) { + tweakables := h.config.Tweakables(scheduler.Namespace) + snapshot := newBufferProcessingSnapshot(invoker, scheduler, catchupWindow(scheduler, tweakables)) + plan := schedulerinternal.PlanBufferProcessing(snapshot, ctx.Now(invoker)) + result = applyBufferPlan(ctx, scheduler, invoker, plan).result + } else { + result = h.processBufferLegacy(ctx, invoker, scheduler) + var totalMissedCatchup int64 + for _, count := range result.missedCatchupByActionRunning { + totalMissedCatchup += count + } + scheduler.recordActionResult(&schedulerActionResult{ + overlapSkipped: result.overlapSkipped, + missedCatchupWindow: totalMissedCatchup, + }) + invoker.recordProcessBufferResult(ctx, &result) + } + + h.recordBufferProcessingMetrics(scheduler, result) + return nil +} - // Update Scheduler metadata. - var totalMissedCatchup int64 - for _, count := range result.missedCatchupByActionRunning { - totalMissedCatchup += count +func (h *InvokerProcessBufferTaskHandler) recordBufferProcessingMetrics( + scheduler *Scheduler, + result processBufferResult, +) { + metricsHandler := newTaggedMetricsHandler(h.metricsHandler, scheduler) + for _, reason := range result.bufferedStartDropReasons { + metricsHandler.Counter(metrics.ScheduleBufferedStartDropped.Name()). + Record(1, metrics.ReasonTag(reason)) } - scheduler.recordActionResult(&schedulerActionResult{ - overlapSkipped: result.overlapSkipped, - missedCatchupWindow: totalMissedCatchup, - }) for overlapPolicy, count := range result.overlapSkippedByPolicy { newTaggedMetricsHandler(h.metricsHandler, scheduler).WithTags( metrics.StringTag(metrics.ScheduleOverlapPolicyTag, overlapPolicy.String()), @@ -483,17 +503,12 @@ func (h *InvokerProcessBufferTaskHandler) Execute( metrics.StringTag(metrics.ScheduleActionRunningTag, fmt.Sprintf("%t", actionRunning)), ).Counter(metrics.ScheduleMissedCatchupWindow.Name()).Record(count) } - - // Update internal state and create new tasks. - invoker.recordProcessBufferResult(ctx, &result) - - return nil } // processBuffer resolves the Invoker's buffered starts that haven't yet begun // execution. This is where the decision is made to drive execution to // completion, or skip/drop a start. -func (h *InvokerProcessBufferTaskHandler) processBuffer( +func (h *InvokerProcessBufferTaskHandler) processBufferLegacy( ctx chasm.MutableContext, invoker *Invoker, scheduler *Scheduler, @@ -532,8 +547,6 @@ func (h *InvokerProcessBufferTaskHandler) processBuffer( // Add starting workflows to result, trim others. Catchup-window expiry is // checked before consumeScheduledAction so that a start past its catchup // window doesn't consume a LimitedActions slot. - droppedCounter := newTaggedMetricsHandler(h.metricsHandler, scheduler). - Counter(metrics.ScheduleBufferedStartDropped.Name()) for _, start := range readyStarts { deadline := h.startWorkflowDeadline(ctx, scheduler, start) if ctx.Now(invoker).After(deadline) { @@ -552,7 +565,7 @@ func (h *InvokerProcessBufferTaskHandler) processBuffer( result.missedCatchupByActionRunning[actionRunning]++ } result.discardStarts = append(result.discardStarts, start) - droppedCounter.Record(1, metrics.ReasonTag(bufferedStartDroppedMissedCatchup)) + result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedMissedCatchup) continue } @@ -560,7 +573,7 @@ func (h *InvokerProcessBufferTaskHandler) processBuffer( if !start.Manual && !scheduler.consumeScheduledAction() { // Drop buffered automated actions while paused or out of actions. result.discardStarts = append(result.discardStarts, start) - droppedCounter.Record(1, metrics.ReasonTag(bufferedStartDroppedPausedOrLimited)) + result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedPausedOrLimited) continue } diff --git a/tests/schedule_test.go b/tests/schedule_test.go index 60dab633b99..a94b7d148fc 100644 --- a/tests/schedule_test.go +++ b/tests/schedule_test.go @@ -71,6 +71,9 @@ func scheduleCommonOpts(t *testing.T) []testcore.TestOption { // only v1 needs the worker service opts = append(opts, testcore.WithWorkerService("V1 scheduler")) } + if strings.HasPrefix(t.Name(), "TestScheduleCHASM") { + opts = append(opts, testcore.WithDynamicConfig(chasmscheduler.EnableBufferPlanner, true)) + } return opts } From 388ef9523752dc85f74f0d413f1b3f9559b8a05d Mon Sep 17 00:00:00 2001 From: "alex.stanfield" <13949480+chaptersix@users.noreply.github.com> Date: Wed, 19 Aug 2026 03:57:02 -0500 Subject: [PATCH 2/3] Use CHASM scheduler buffer planner unconditionally --- chasm/lib/scheduler/buffer_planner.go | 40 +- .../scheduler/buffer_processor_legacy_test.go | 97 +++++ chasm/lib/scheduler/config.go | 8 - chasm/lib/scheduler/export_test.go | 32 +- .../lib/scheduler/internal/buffer_planner.go | 177 ++++---- .../scheduler/internal/buffer_planner_test.go | 28 +- .../lib/scheduler/internal/buffered_start.go | 18 +- .../invoker_process_buffer_task_test.go | 380 +++++++++++++++++- chasm/lib/scheduler/invoker_tasks.go | 115 +----- tests/schedule_test.go | 3 - 10 files changed, 659 insertions(+), 239 deletions(-) create mode 100644 chasm/lib/scheduler/buffer_processor_legacy_test.go diff --git a/chasm/lib/scheduler/buffer_planner.go b/chasm/lib/scheduler/buffer_planner.go index 26f1b687fe1..ce0c32bd7f0 100644 --- a/chasm/lib/scheduler/buffer_planner.go +++ b/chasm/lib/scheduler/buffer_planner.go @@ -12,12 +12,13 @@ import ( schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal" ) +// appliedBufferPlan collects the live-state result and the number of decisions invalidated during revalidation. type appliedBufferPlan struct { - result processBufferResult - decisions []schedulerinternal.BufferDecision - staleDecisions int64 + result processBufferResult + invalidatedDecisions int64 } +// newBufferProcessingSnapshot projects live CHASM state into values that the pure planner cannot mutate. func newBufferProcessingSnapshot(invoker *Invoker, scheduler *Scheduler, catchupWindow time.Duration) schedulerinternal.BufferProcessingSnapshot { state := scheduler.Schedule.GetState() snapshot := schedulerinternal.BufferProcessingSnapshot{ @@ -56,6 +57,8 @@ func projectBufferedStart(start *schedulespb.BufferedStart) schedulerinternal.Bu } } +// applyBufferPlan revalidates a plan, resolves its value decisions back to live +// BufferedStart pointers, and applies only decisions whose expected state still matches. func applyBufferPlan( ctx chasm.MutableContext, scheduler *Scheduler, @@ -65,7 +68,7 @@ func applyBufferPlan( applied := newAppliedBufferPlan() currentSnapshot := newBufferProcessingSnapshot(invoker, scheduler, plan.Snapshot.CatchupWindow) if !bufferProcessingSnapshotsEqual(plan.Snapshot, currentSnapshot) { - return staleBufferPlan(plan, applied) + return invalidateBufferPlan(plan, applied) } startsByRequestID := make(map[string][]*schedulespb.BufferedStart) for _, start := range invoker.GetBufferedStarts() { @@ -74,24 +77,22 @@ func applyBufferPlan( for _, decision := range plan.Decisions { if !decision.MutatesState() { - applied.decisions = append(applied.decisions, decision) continue } start, ok := popMatchingBufferedStart(startsByRequestID, decision) if !ok { - applied.recordStaleDecision(decision) + applied.recordInvalidatedDecision() continue } - if decision.Action == schedulerinternal.BufferDecisionExecute && !start.GetManual() && !scheduler.consumeScheduledAction() { - applied.recordStaleDecision(decision) + if decision.ConsumesScheduledAction && !scheduler.consumeScheduledAction() { + applied.recordInvalidatedDecision() continue } applyBufferDecision(&applied.result, decision, start) - applied.decisions = append(applied.decisions, decision) } - if applied.staleDecisions == 0 { + if applied.invalidatedDecisions == 0 { applied.result.overlapSkipped = plan.OverlapSkipped applied.result.overlapSkippedByPolicy = maps.Clone(plan.OverlapSkippedByPolicy) } @@ -127,24 +128,21 @@ func newAppliedBufferPlan() appliedBufferPlan { }} } -func staleBufferPlan(plan schedulerinternal.BufferPlan, applied appliedBufferPlan) appliedBufferPlan { +func invalidateBufferPlan(plan schedulerinternal.BufferPlan, applied appliedBufferPlan) appliedBufferPlan { for _, decision := range plan.Decisions { if decision.MutatesState() { - applied.recordStaleDecision(decision) - } else { - applied.decisions = append(applied.decisions, decision) + applied.recordInvalidatedDecision() } } return applied } -func (a *appliedBufferPlan) recordStaleDecision(decision schedulerinternal.BufferDecision) { - decision.Action = schedulerinternal.BufferDecisionRetain - decision.Reason = schedulerinternal.BufferDecisionReasonStale - a.staleDecisions++ - a.decisions = append(a.decisions, decision) +func (a *appliedBufferPlan) recordInvalidatedDecision() { + a.invalidatedDecisions++ } +// popMatchingBufferedStart resolves one decision to its live protobuf pointer. +// Removing the match ensures duplicate request IDs cannot reuse the same start. func popMatchingBufferedStart( startsByRequestID map[string][]*schedulespb.BufferedStart, decision schedulerinternal.BufferDecision, @@ -166,12 +164,12 @@ func applyBufferDecision(result *processBufferResult, decision schedulerinternal result.startWorkflows = append(result.startWorkflows, start) case schedulerinternal.BufferDecisionDiscard: result.discardStarts = append(result.discardStarts, start) - recordAppliedDiscard(result, decision) default: } + recordAppliedDecisionMetrics(result, decision) } -func recordAppliedDiscard(result *processBufferResult, decision schedulerinternal.BufferDecision) { +func recordAppliedDecisionMetrics(result *processBufferResult, decision schedulerinternal.BufferDecision) { switch decision.Reason { case schedulerinternal.BufferDecisionReasonMissedCatchupWindow: result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedMissedCatchup) diff --git a/chasm/lib/scheduler/buffer_processor_legacy_test.go b/chasm/lib/scheduler/buffer_processor_legacy_test.go new file mode 100644 index 00000000000..621de6f513d --- /dev/null +++ b/chasm/lib/scheduler/buffer_processor_legacy_test.go @@ -0,0 +1,97 @@ +package scheduler + +import ( + enumspb "go.temporal.io/api/enums/v1" + schedulespb "go.temporal.io/server/api/schedule/v1" + "go.temporal.io/server/chasm" + "go.temporal.io/server/common/util" + legacyscheduler "go.temporal.io/server/service/worker/scheduler" +) + +func (h *InvokerProcessBufferTaskHandler) processBufferLegacy( + ctx chasm.MutableContext, + invoker *Invoker, + scheduler *Scheduler, +) (result processBufferResult) { + runningWorkflows := invoker.runningWorkflowExecutions() + isRunning := len(runningWorkflows) > 0 + result.missedCatchupByActionRunning = make(map[bool]int64) + + // Processing ignores starts that are already executing or backing off. An existing + // deferred BUFFER_ONE start still participates so it can reject later starts. + pendingBufferedStarts := util.FilterSlice(invoker.GetBufferedStarts(), func(start *schedulespb.BufferedStart) bool { + return start.Attempt == 0 || + (start.Attempt == -1 && scheduler.resolveOverlapPolicy(start.GetOverlapPolicy()) == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE) + }) + + // Resolve overlap policies and trim BufferedStarts that are skipped by policy. + action := legacyscheduler.ProcessBuffer(pendingBufferedStarts, isRunning, scheduler.resolveOverlapPolicy) + + // ProcessBuffer will drop starts by omitting them from NewBuffer. Start with the + // diff between the input and NewBuffer, and add any executing starts. + keepStarts := make(map[string]struct{}) // request ID -> is present + for _, start := range action.NewBuffer { + keepStarts[start.GetRequestId()] = struct{}{} + } + + // Combine all available starts. + readyStarts := action.OverlappingStarts + if action.NonOverlappingStart != nil { + readyStarts = append(readyStarts, action.NonOverlappingStart) + } + + // Update result metrics. + result.overlapSkipped = action.OverlapSkipped + result.overlapSkippedByPolicy = action.OverlapSkippedByPolicy + + // Add starting workflows to result, trim others. Catchup-window expiry is + // checked before consumeScheduledAction so that a start past its catchup + // window doesn't consume a LimitedActions slot. + for _, start := range readyStarts { + deadline := h.startWorkflowDeadline(ctx, scheduler, start) + if ctx.Now(invoker).After(deadline) { + // Action was buffered in time but expired before execution + // (e.g., due to overlap deferral, retries, or system delay). + // Only emit the metric if the schedule would have run this + // start -- skip paused or action-exhausted schedules. + if start.Manual || scheduler.canTakeScheduledAction() { + // Determine if a running action contributed: either one is still + // running, or the previous action's CloseTime (stored in DesiredTime) + // was already past this start's deadline. + // Note: if no prior action completed, DesiredTime is zero-valued, + // so After(deadline) is false, correctly yielding actionRunning=false. + actionRunning := isRunning || + start.GetDesiredTime().AsTime().After(deadline) + result.missedCatchupByActionRunning[actionRunning]++ + } + result.discardStarts = append(result.discardStarts, start) + result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedMissedCatchup) + continue + } + + // Ensure we can take more actions. Manual actions are always allowed. + if !start.Manual && !scheduler.consumeScheduledAction() { + // Drop buffered automated actions while paused or out of actions. + result.discardStarts = append(result.discardStarts, start) + result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedPausedOrLimited) + continue + } + + keepStarts[start.GetRequestId()] = struct{}{} + result.startWorkflows = append(result.startWorkflows, start) + } + + result.discardStarts = util.FilterSlice(pendingBufferedStarts, func(start *schedulespb.BufferedStart) bool { + _, keep := keepStarts[start.GetRequestId()] + return !keep + }) + + // Terminate overrides cancel if both are requested. + if action.NeedTerminate { + result.terminateWorkflows = runningWorkflows + } else if action.NeedCancel { + result.cancelWorkflows = runningWorkflows + } + + return +} diff --git a/chasm/lib/scheduler/config.go b/chasm/lib/scheduler/config.go index 14bcc895838..f90ce47af4d 100644 --- a/chasm/lib/scheduler/config.go +++ b/chasm/lib/scheduler/config.go @@ -29,7 +29,6 @@ type ( ServiceCallTimeout dynamicconfig.DurationPropertyFn RetryPolicy func() backoff.RetryPolicy EncodeInternalTokenWithEnvelope dynamicconfig.BoolPropertyFnWithNamespaceFilter - EnableBufferPlanner dynamicconfig.BoolPropertyFnWithNamespaceFilter } ) @@ -81,12 +80,6 @@ var ( `The upper bound on how long a service call can take before being timed out.`, ) - EnableBufferPlanner = dynamicconfig.NewNamespaceBoolSetting( - "scheduler.enableBufferPlanner", - false, - `Use the pure planner to process CHASM scheduler buffered starts. The legacy processor remains the fallback while this setting is disabled.`, - ) - // SentinelIdleTime is how long a CHASM sentinel reserves the schedule ID // before auto-closing via the idle task mechanism. Matches the dummy // workflow's duration. @@ -111,7 +104,6 @@ func ConfigProvider(dc *dynamicconfig.Collection) *Config { Tweakables: CurrentTweakables.Get(dc), ServiceCallTimeout: ServiceCallTimeout.Get(dc), EncodeInternalTokenWithEnvelope: callback.EncodeInternalTokenWithEnvelope.Get(dc), - EnableBufferPlanner: EnableBufferPlanner.Get(dc), RetryPolicy: func() backoff.RetryPolicy { return backoff.NewExponentialRetryPolicy( RetryPolicyInitialInterval.Get(dc)(), diff --git a/chasm/lib/scheduler/export_test.go b/chasm/lib/scheduler/export_test.go index 286be6943da..175da50faa8 100644 --- a/chasm/lib/scheduler/export_test.go +++ b/chasm/lib/scheduler/export_test.go @@ -9,6 +9,8 @@ import ( "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1" schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal" "go.temporal.io/server/common/log" + "go.temporal.io/server/common/metrics" + queueerrors "go.temporal.io/server/service/history/queues/errors" legacyscheduler "go.temporal.io/server/service/worker/scheduler" ) @@ -82,10 +84,38 @@ func (h *InvokerProcessBufferTaskHandler) PlanBufferProcessingForTest( snapshot := newBufferProcessingSnapshot(invoker, scheduler, catchupWindow(scheduler, tweakables)) plan := schedulerinternal.PlanBufferProcessing(snapshot, now) return func(ctx chasm.MutableContext, scheduler *Scheduler, invoker *Invoker) int64 { - return applyBufferPlan(ctx, scheduler, invoker, plan).staleDecisions + return applyBufferPlan(ctx, scheduler, invoker, plan).invalidatedDecisions } } +func (h *InvokerProcessBufferTaskHandler) ExecuteProcessBufferLegacyForTest( + ctx chasm.MutableContext, + invoker *Invoker, +) error { + scheduler := invoker.Scheduler.Get(ctx) + newTaggedMetricsHandler(h.metricsHandler, scheduler). + Counter(metrics.ScheduleInvokerProcessBufferTask.Name()). + Record(1, metrics.OutcomeTag(outcomeFired), metrics.ReasonTag(reasonNone)) + + invoker.getOrCreateEventLog(ctx).LogEvent(ctx, "processBufferTask executed") + if scheduler.Schedule.GetAction().GetStartWorkflow() == nil { + return queueerrors.NewUnprocessableTaskError("schedules must have an Action set") + } + + result := h.processBufferLegacy(ctx, invoker, scheduler) + var totalMissedCatchup int64 + for _, count := range result.missedCatchupByActionRunning { + totalMissedCatchup += count + } + scheduler.recordActionResult(&schedulerActionResult{ + overlapSkipped: result.overlapSkipped, + missedCatchupWindow: totalMissedCatchup, + }) + invoker.recordProcessBufferResult(ctx, &result) + h.recordBufferProcessingMetrics(scheduler, result) + return nil +} + func (b *BackfillerTaskHandler) ProcessBackfill( scheduler *Scheduler, backfiller *Backfiller, diff --git a/chasm/lib/scheduler/internal/buffer_planner.go b/chasm/lib/scheduler/internal/buffer_planner.go index 958e6d47a47..ed193b43fef 100644 --- a/chasm/lib/scheduler/internal/buffer_planner.go +++ b/chasm/lib/scheduler/internal/buffer_planner.go @@ -7,6 +7,7 @@ import ( enumspb "go.temporal.io/api/enums/v1" ) +// BufferDecisionAction describes how the apply phase should treat one buffered start. type BufferDecisionAction int const ( @@ -19,6 +20,7 @@ const ( BufferDecisionCompleted ) +// BufferDecisionReason records why the planner selected an action. type BufferDecisionReason int const ( @@ -28,9 +30,10 @@ const ( BufferDecisionReasonCancelOrTerminate BufferDecisionReasonMissedCatchupWindow BufferDecisionReasonPausedOrLimited - BufferDecisionReasonStale ) +// BufferedStartSnapshot is the value-only planner projection of a persisted BufferedStart. +// Keeping protobuf pointers out of the planner prevents planning from mutating live CHASM state. type BufferedStartSnapshot struct { RequestID string WorkflowID string @@ -43,15 +46,13 @@ type BufferedStartSnapshot struct { Completed bool } -func (s BufferedStartSnapshot) GetOverlapPolicy() enumspb.ScheduleOverlapPolicy { - return s.OverlapPolicy -} - +// WorkflowExecutionSnapshot identifies a running workflow without retaining a protobuf pointer. type WorkflowExecutionSnapshot struct { WorkflowID string RunID string } +// BufferProcessingSnapshot contains all state read by one buffer-planning pass. type BufferProcessingSnapshot struct { Starts []BufferedStartSnapshot RunningWorkflows []WorkflowExecutionSnapshot @@ -63,29 +64,33 @@ type BufferProcessingSnapshot struct { RemainingActions int64 } +// BufferDecision describes the planned outcome for one exact BufferedStart snapshot. type BufferDecision struct { - RequestID string - Expected BufferedStartSnapshot - Action BufferDecisionAction - Reason BufferDecisionReason - OverlapPolicy enumspb.ScheduleOverlapPolicy - OverlapSkipped bool + RequestID string + // Expected is matched against current persisted state before any mutation is applied. + Expected BufferedStartSnapshot + Action BufferDecisionAction + Reason BufferDecisionReason + // ConsumesScheduledAction defers the live capacity decrement until after revalidation. + ConsumesScheduledAction bool MissedCatchupMetric bool MissedCatchupActionRunning bool } +// MutatesState reports whether applying the decision changes the persisted buffer. func (d BufferDecision) MutatesState() bool { return d.Action == BufferDecisionExecute || d.Action == BufferDecisionDiscard || d.Action == BufferDecisionDefer } +// BufferPlan is an ordered, value-based description of a buffer-processing pass. +// Snapshot anchors whole-plan validation; each decision also carries its exact expected start. type BufferPlan struct { - Snapshot BufferProcessingSnapshot - Decisions []BufferDecision - CancelWorkflows []WorkflowExecutionSnapshot - TerminateWorkflows []WorkflowExecutionSnapshot - OverlapSkipped int64 - OverlapSkippedByPolicy map[enumspb.ScheduleOverlapPolicy]int64 - MissedCatchupByActionRunning map[bool]int64 + Snapshot BufferProcessingSnapshot + Decisions []BufferDecision + CancelWorkflows []WorkflowExecutionSnapshot + TerminateWorkflows []WorkflowExecutionSnapshot + OverlapSkipped int64 + OverlapSkippedByPolicy map[enumspb.ScheduleOverlapPolicy]int64 } type overlapAction struct { @@ -98,11 +103,11 @@ type overlapAction struct { OverlapSkippedByPolicy map[enumspb.ScheduleOverlapPolicy]int64 } +// PlanBufferProcessing computes buffer outcomes without mutating snapshot or live scheduler state. func PlanBufferProcessing(snapshot BufferProcessingSnapshot, now time.Time) BufferPlan { plan := BufferPlan{ - Snapshot: cloneBufferProcessingSnapshot(snapshot), - OverlapSkippedByPolicy: make(map[enumspb.ScheduleOverlapPolicy]int64), - MissedCatchupByActionRunning: make(map[bool]int64), + Snapshot: cloneBufferProcessingSnapshot(snapshot), + OverlapSkippedByPolicy: make(map[enumspb.ScheduleOverlapPolicy]int64), } resolveOverlapPolicy := func(policy enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy { if policy == enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED { @@ -111,14 +116,7 @@ func PlanBufferProcessing(snapshot BufferProcessingSnapshot, now time.Time) Buff return policy } - pending := make([]BufferedStartSnapshot, 0, len(snapshot.Starts)) - for _, start := range snapshot.Starts { - if start.Attempt == 0 || - (start.Attempt == -1 && resolveOverlapPolicy(start.OverlapPolicy) == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE) { - pending = append(pending, start) - } - } - + pending := pendingBufferedStarts(snapshot.Starts, resolveOverlapPolicy) action := planOverlapActions(pending, len(snapshot.RunningWorkflows) > 0, resolveOverlapPolicy) plan.OverlapSkipped = action.OverlapSkipped plan.OverlapSkippedByPolicy = action.OverlapSkippedByPolicy @@ -128,38 +126,52 @@ func PlanBufferProcessing(snapshot BufferProcessingSnapshot, now time.Time) Buff plan.CancelWorkflows = append(plan.CancelWorkflows, snapshot.RunningWorkflows...) } - deferred := make(map[string]struct{}, len(action.NewBuffer)) + keep := make(map[string]struct{}, len(action.NewBuffer)+len(action.OverlappingStarts)+1) for _, start := range action.NewBuffer { - deferred[start.RequestID] = struct{}{} - } - ready := make(map[string]struct{}, len(action.OverlappingStarts)+1) - for _, start := range action.OverlappingStarts { - ready[start.RequestID] = struct{}{} + keep[start.RequestID] = struct{}{} } + readyStarts := slices.Clone(action.OverlappingStarts) if action.NonOverlappingStart.RequestID != "" { - ready[action.NonOverlappingStart.RequestID] = struct{}{} + readyStarts = append(readyStarts, action.NonOverlappingStart) } + // All ready decisions share this task-wide budget. Passing its address preserves + // legacy ready-start ordering while keeping the input snapshot immutable. remainingActions := snapshot.RemainingActions + ready := make(map[string]struct{}, len(readyStarts)) + readyDecisions := make([]BufferDecision, 0, len(readyStarts)) + readyOccurrences := make(map[BufferedStartSnapshot]int, len(readyStarts)) + for _, start := range readyStarts { + decision := planReadyBufferDecision(start, snapshot, now, &remainingActions) + if decision.Action == BufferDecisionExecute { + keep[start.RequestID] = struct{}{} + ready[start.RequestID] = struct{}{} + } + readyDecisions = append(readyDecisions, decision) + readyOccurrences[start]++ + } + + for _, decision := range readyDecisions { + decision.Action = finalPendingBufferAction(decision.RequestID, keep, ready) + plan.Decisions = append(plan.Decisions, decision) + } for _, start := range snapshot.Starts { - isPending := start.Attempt == 0 || - (start.Attempt == -1 && resolveOverlapPolicy(start.OverlapPolicy) == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE) - if !isPending { + if !isPendingBufferedStart(start, resolveOverlapPolicy) { plan.Decisions = append(plan.Decisions, alreadyProcessedBufferDecision(start)) continue } - decision := planPendingBufferDecision( - start, - snapshot, - now, - deferred, - ready, - action.NeedCancel || action.NeedTerminate, - resolveOverlapPolicy, - &remainingActions, - ) - if decision.MissedCatchupMetric { - plan.MissedCatchupByActionRunning[decision.MissedCatchupActionRunning]++ + if readyOccurrences[start] > 0 { + readyOccurrences[start]-- + continue + } + decision := BufferDecision{ + RequestID: start.RequestID, + Expected: start, + Action: finalPendingBufferAction(start.RequestID, keep, ready), + Reason: BufferDecisionReasonOverlapPolicy, + } + if decision.Action == BufferDecisionDiscard && (action.NeedCancel || action.NeedTerminate) { + decision.Reason = BufferDecisionReasonCancelOrTerminate } plan.Decisions = append(plan.Decisions, decision) } @@ -218,6 +230,28 @@ func planOverlapActions( return action } +func pendingBufferedStarts( + starts []BufferedStartSnapshot, + resolve func(enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy, +) []BufferedStartSnapshot { + pending := make([]BufferedStartSnapshot, 0, len(starts)) + for _, start := range starts { + if isPendingBufferedStart(start, resolve) { + pending = append(pending, start) + } + } + return pending +} + +func isPendingBufferedStart( + start BufferedStartSnapshot, + resolve func(enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy, +) bool { + return start.Attempt == bufferedStartUnprocessedAttempt || + (start.Attempt == bufferedStartDeferredAttempt && + resolve(start.OverlapPolicy) == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE) +} + func alreadyProcessedBufferDecision(start BufferedStartSnapshot) BufferDecision { decision := BufferDecision{ RequestID: start.RequestID, @@ -230,42 +264,20 @@ func alreadyProcessedBufferDecision(start BufferedStartSnapshot) BufferDecision decision.Action = BufferDecisionCompleted case start.RunID != "": decision.Action = BufferDecisionRunning - case start.Attempt > 1: + case start.Attempt > bufferedStartFirstExecutionAttempt: decision.Action = BufferDecisionRetry default: } return decision } -func planPendingBufferDecision( +func planReadyBufferDecision( start BufferedStartSnapshot, snapshot BufferProcessingSnapshot, now time.Time, - deferred map[string]struct{}, - ready map[string]struct{}, - needCancelOrTerminate bool, - resolveOverlapPolicy func(enumspb.ScheduleOverlapPolicy) enumspb.ScheduleOverlapPolicy, remainingActions *int64, ) BufferDecision { decision := BufferDecision{RequestID: start.RequestID, Expected: start} - if _, ok := deferred[start.RequestID]; ok { - decision.Action = BufferDecisionDefer - decision.Reason = BufferDecisionReasonOverlapPolicy - return decision - } - if _, ok := ready[start.RequestID]; !ok { - decision.Action = BufferDecisionDiscard - decision.Reason = BufferDecisionReasonOverlapPolicy - if needCancelOrTerminate { - decision.Reason = BufferDecisionReasonCancelOrTerminate - return decision - } - decision.OverlapPolicy = resolveOverlapPolicy(start.OverlapPolicy) - decision.OverlapSkipped = decision.OverlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_SKIP || - decision.OverlapPolicy == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE - return decision - } - deadline := start.ActualTime.Add(max(snapshot.CatchupWindow, snapshot.MinimumCatchupWindow)) canTakeScheduledAction := !snapshot.Paused && (!snapshot.LimitedActions || *remainingActions > 0) if !start.Manual && now.After(deadline) { @@ -286,9 +298,24 @@ func planPendingBufferDecision( *remainingActions-- } decision.Action = BufferDecisionExecute + decision.ConsumesScheduledAction = !start.Manual return decision } +func finalPendingBufferAction( + requestID string, + keep map[string]struct{}, + ready map[string]struct{}, +) BufferDecisionAction { + if _, ok := ready[requestID]; ok { + return BufferDecisionExecute + } + if _, ok := keep[requestID]; ok { + return BufferDecisionDefer + } + return BufferDecisionDiscard +} + func cloneBufferProcessingSnapshot(snapshot BufferProcessingSnapshot) BufferProcessingSnapshot { snapshot.Starts = slices.Clone(snapshot.Starts) snapshot.RunningWorkflows = slices.Clone(snapshot.RunningWorkflows) diff --git a/chasm/lib/scheduler/internal/buffer_planner_test.go b/chasm/lib/scheduler/internal/buffer_planner_test.go index 3899089cc61..b0290d3254e 100644 --- a/chasm/lib/scheduler/internal/buffer_planner_test.go +++ b/chasm/lib/scheduler/internal/buffer_planner_test.go @@ -80,7 +80,33 @@ func TestPlanBufferProcessing_TimeAndCapacity(t *testing.T) { }, decisionActions(plan.Decisions)) require.Equal(t, BufferDecisionReasonMissedCatchupWindow, plan.Decisions[0].Reason) require.Equal(t, BufferDecisionReasonPausedOrLimited, plan.Decisions[2].Reason) - require.Equal(t, int64(1), plan.MissedCatchupByActionRunning[false]) + require.True(t, plan.Decisions[0].MissedCatchupMetric) + require.False(t, plan.Decisions[0].MissedCatchupActionRunning) +} + +func TestPlanBufferProcessing_PreservesLegacyReadyOrder(t *testing.T) { + now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC) + snapshot := BufferProcessingSnapshot{ + Starts: []BufferedStartSnapshot{ + {RequestID: "non-overlapping", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, ActualTime: now}, + {RequestID: "allow-all", OverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, ActualTime: now}, + }, + DefaultOverlapPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, + CatchupWindow: time.Hour, + MinimumCatchupWindow: 5 * time.Second, + LimitedActions: true, + RemainingActions: 1, + } + + plan := PlanBufferProcessing(snapshot, now) + + require.Len(t, plan.Decisions, 2) + require.Equal(t, "allow-all", plan.Decisions[0].RequestID) + require.Equal(t, BufferDecisionExecute, plan.Decisions[0].Action) + require.True(t, plan.Decisions[0].ConsumesScheduledAction) + require.Equal(t, "non-overlapping", plan.Decisions[1].RequestID) + require.Equal(t, BufferDecisionDiscard, plan.Decisions[1].Action) + require.Equal(t, BufferDecisionReasonPausedOrLimited, plan.Decisions[1].Reason) } func TestPlanBufferProcessing_PausedAndProcessedStates(t *testing.T) { diff --git a/chasm/lib/scheduler/internal/buffered_start.go b/chasm/lib/scheduler/internal/buffered_start.go index 8a316469796..b60a3340ea2 100644 --- a/chasm/lib/scheduler/internal/buffered_start.go +++ b/chasm/lib/scheduler/internal/buffered_start.go @@ -20,12 +20,18 @@ const ( BufferedStartStateCompleted ) +const ( + bufferedStartDeferredAttempt int64 = -1 + bufferedStartUnprocessedAttempt int64 = 0 + bufferedStartFirstExecutionAttempt int64 = 1 +) + // ClassifyBufferedStart returns the lifecycle state encoded by start at retryEvaluationTime. func ClassifyBufferedStart( start *schedulespb.BufferedStart, retryEvaluationTime time.Time, ) BufferedStartState { - if start == nil || start.GetAttempt() < -1 { + if start == nil || start.GetAttempt() < bufferedStartDeferredAttempt { return BufferedStartStateInvalid } @@ -34,12 +40,12 @@ func ClassifyBufferedStart( hasBackoff := start.GetBackoffTime() != nil switch start.GetAttempt() { - case -1: + case bufferedStartDeferredAttempt: if hasRun || hasCompletion || hasBackoff { return BufferedStartStateInvalid } return BufferedStartStateDeferred - case 0: + case bufferedStartUnprocessedAttempt: if hasRun || hasCompletion || hasBackoff { return BufferedStartStateInvalid } @@ -63,7 +69,7 @@ func ClassifyBufferedStart( // MarkStartUnprocessed marks a start for its initial overlap-policy pass. func MarkStartUnprocessed(start *schedulespb.BufferedStart) { - start.Attempt = 0 + start.Attempt = bufferedStartUnprocessedAttempt } // MarkStartMigratedUnprocessed clears V2 retry state from a pending V1 start. @@ -74,12 +80,12 @@ func MarkStartMigratedUnprocessed(start *schedulespb.BufferedStart) { // MarkStartDeferred marks a start as waiting for an overlapping action to complete. func MarkStartDeferred(start *schedulespb.BufferedStart) { - start.Attempt = -1 + start.Attempt = bufferedStartDeferredAttempt } // MarkStartReady marks a start as ready for its first execution attempt. func MarkStartReady(start *schedulespb.BufferedStart) { - start.Attempt = 1 + start.Attempt = bufferedStartFirstExecutionAttempt } // MarkStartRetrying advances a start to its next attempt and backoff deadline. diff --git a/chasm/lib/scheduler/invoker_process_buffer_task_test.go b/chasm/lib/scheduler/invoker_process_buffer_task_test.go index bf3e22c9492..8d8db226125 100644 --- a/chasm/lib/scheduler/invoker_process_buffer_task_test.go +++ b/chasm/lib/scheduler/invoker_process_buffer_task_test.go @@ -843,6 +843,357 @@ func TestProcessBufferTask_LegacyAndPlannerDifferentialCorpus(t *testing.T) { } } +func TestProcessBufferTask_LegacyAndPlannerDifferentialOverlapStateSpace(t *testing.T) { + now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC) + policies := []enumspb.ScheduleOverlapPolicy{ + enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, + enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, + enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, + enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER, + enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER, + enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, + } + capacityCases := []struct { + name string + limited bool + remaining int64 + }{ + {name: "unlimited"}, + {name: "exhausted", limited: true}, + {name: "one", limited: true, remaining: 1}, + {name: "two", limited: true, remaining: 2}, + {name: "three", limited: true, remaining: 3}, + } + + comparisons := 0 + for length := 0; length <= 4; length++ { + forEachOverlapPolicySequence(policies, length, func(sequence []enumspb.ScheduleOverlapPolicy) { + for _, running := range []bool{false, true} { + for _, capacity := range capacityCases { + schedule := defaultSchedule() + schedule.Policies.OverlapPolicy = enumspb.SCHEDULE_OVERLAP_POLICY_SKIP + schedule.State.LimitedActions = capacity.limited + schedule.State.RemainingActions = capacity.remaining + starts := make([]*schedulespb.BufferedStart, 0, len(sequence)+1) + if running { + start := newBufferProcessingComparisonStart(now, "running", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + start.Attempt = 1 + start.RunId = "running-run" + starts = append(starts, start) + } + for index, policy := range sequence { + starts = append(starts, newBufferProcessingComparisonStart(now, fmt.Sprintf("pending-%d", index), policy)) + } + compareBufferProcessing(t, bufferProcessingComparisonInput{ + name: fmt.Sprintf( + "overlap state length=%d policies=%v running=%t capacity=%s", + length, + sequence, + running, + capacity.name, + ), + schedule: schedule, + bufferedStarts: starts, + lastProcessedTime: now, + initialConflictToken: 17, + }) + comparisons++ + } + } + }) + } + require.Equal(t, 15550, comparisons) +} + +func TestProcessBufferTask_LegacyAndPlannerDifferentialDuplicateIdentityStateSpace(t *testing.T) { + now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC) + policies := []enumspb.ScheduleOverlapPolicy{ + enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED, + enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, + enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, + enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, + enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER, + enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER, + enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, + } + requestIDPatterns := []struct { + name string + id func(int) string + }{ + {name: "same", id: func(int) string { return "duplicate" }}, + {name: "alternating", id: func(index int) string { return fmt.Sprintf("duplicate-%d", index%2) }}, + } + comparisons := 0 + for length := 1; length <= 3; length++ { + forEachOverlapPolicySequence(policies, length, func(sequence []enumspb.ScheduleOverlapPolicy) { + for _, requestIDs := range requestIDPatterns { + for _, running := range []bool{false, true} { + for remaining := int64(0); remaining <= 3; remaining++ { + schedule := defaultSchedule() + schedule.Policies.OverlapPolicy = enumspb.SCHEDULE_OVERLAP_POLICY_SKIP + schedule.State.LimitedActions = true + schedule.State.RemainingActions = remaining + starts := make([]*schedulespb.BufferedStart, 0, len(sequence)+1) + if running { + start := newBufferProcessingComparisonStart(now, "running", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + start.Attempt = 1 + start.RunId = "running-run" + starts = append(starts, start) + } + for index, policy := range sequence { + starts = append(starts, newBufferProcessingComparisonStart(now, requestIDs.id(index), policy)) + } + compareBufferProcessing(t, bufferProcessingComparisonInput{ + name: fmt.Sprintf( + "duplicate identities pattern=%s policies=%v running=%t remaining=%d", + requestIDs.name, + sequence, + running, + remaining, + ), + schedule: schedule, + bufferedStarts: starts, + lastProcessedTime: now, + initialConflictToken: 17, + }) + comparisons++ + } + } + } + }) + } + require.Equal(t, 6384, comparisons) +} + +func TestProcessBufferTask_LegacyAndPlannerDifferentialDefaultPolicyStateSpace(t *testing.T) { + now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC) + policies := []enumspb.ScheduleOverlapPolicy{ + enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, + enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, + enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, + enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER, + enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER, + enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, + } + comparisons := 0 + for _, defaultPolicy := range policies { + for unspecifiedIndex := 0; unspecifiedIndex < 3; unspecifiedIndex++ { + forEachOverlapPolicySequence(policies, 2, func(surrounding []enumspb.ScheduleOverlapPolicy) { + sequence := make([]enumspb.ScheduleOverlapPolicy, 3) + sequence[unspecifiedIndex] = enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED + for index, policy := range surrounding { + if index >= unspecifiedIndex { + index++ + } + sequence[index] = policy + } + for _, running := range []bool{false, true} { + for remaining := int64(0); remaining <= 3; remaining++ { + schedule := defaultSchedule() + schedule.Policies.OverlapPolicy = defaultPolicy + schedule.State.LimitedActions = true + schedule.State.RemainingActions = remaining + starts := make([]*schedulespb.BufferedStart, 0, len(sequence)+1) + if running { + start := newBufferProcessingComparisonStart(now, "running", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + start.Attempt = 1 + start.RunId = "running-run" + starts = append(starts, start) + } + for index, policy := range sequence { + starts = append(starts, newBufferProcessingComparisonStart(now, fmt.Sprintf("pending-%d", index), policy)) + } + compareBufferProcessing(t, bufferProcessingComparisonInput{ + name: fmt.Sprintf( + "default policy=%s policies=%v running=%t remaining=%d", + defaultPolicy, + sequence, + running, + remaining, + ), + schedule: schedule, + bufferedStarts: starts, + lastProcessedTime: now, + initialConflictToken: 17, + }) + comparisons++ + } + } + }) + } + } + require.Equal(t, 5184, comparisons) +} + +func TestProcessBufferTask_LegacyAndPlannerDifferentialTimeCapacityStateSpace(t *testing.T) { + now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC) + catchupWindows := []time.Duration{time.Second, 5 * time.Second, time.Hour} + timeRelations := []struct { + name string + offset time.Duration + }{ + {name: "before", offset: time.Nanosecond}, + {name: "equal"}, + {name: "after", offset: -time.Nanosecond}, + } + capacityCases := []struct { + name string + limited bool + remaining int64 + }{ + {name: "unlimited"}, + {name: "exhausted", limited: true}, + {name: "one", limited: true, remaining: 1}, + {name: "two", limited: true, remaining: 2}, + } + + comparisons := 0 + for _, catchupWindow := range catchupWindows { + effectiveWindow := max(catchupWindow, 5*time.Second) + for _, deadlineRelation := range timeRelations { + actualTime := now.Add(-effectiveWindow + deadlineRelation.offset) + deadline := actualTime.Add(effectiveWindow) + for _, desiredRelation := range timeRelations { + desiredTime := deadline.Add(-desiredRelation.offset) + for _, manual := range []bool{false, true} { + for _, paused := range []bool{false, true} { + for _, running := range []bool{false, true} { + for _, capacity := range capacityCases { + schedule := defaultSchedule() + schedule.Policies.CatchupWindow = durationpb.New(catchupWindow) + schedule.State.Paused = paused + schedule.State.LimitedActions = capacity.limited + schedule.State.RemainingActions = capacity.remaining + start := newBufferProcessingComparisonStart(now, "pending", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + start.ActualTime = timestamppb.New(actualTime) + start.DesiredTime = timestamppb.New(desiredTime) + start.Manual = manual + starts := []*schedulespb.BufferedStart{start} + if running { + runningStart := newBufferProcessingComparisonStart(now, "running", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + runningStart.Attempt = 1 + runningStart.RunId = "running-run" + starts = append([]*schedulespb.BufferedStart{runningStart}, starts...) + } + compareBufferProcessing(t, bufferProcessingComparisonInput{ + name: fmt.Sprintf( + "time capacity catchup=%s deadline=%s desired=%s manual=%t paused=%t running=%t capacity=%s", + catchupWindow, + deadlineRelation.name, + desiredRelation.name, + manual, + paused, + running, + capacity.name, + ), + schedule: schedule, + bufferedStarts: starts, + lastProcessedTime: now, + initialConflictToken: 17, + }) + comparisons++ + } + } + } + } + } + } + } + require.Equal(t, 864, comparisons) +} + +func TestProcessBufferTask_LegacyAndPlannerDifferentialLifecycleStateSpace(t *testing.T) { + now := time.Date(2026, 8, 19, 12, 0, 0, 0, time.UTC) + policies := []enumspb.ScheduleOverlapPolicy{ + enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED, + enumspb.SCHEDULE_OVERLAP_POLICY_SKIP, + enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE, + enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ALL, + enumspb.SCHEDULE_OVERLAP_POLICY_CANCEL_OTHER, + enumspb.SCHEDULE_OVERLAP_POLICY_TERMINATE_OTHER, + enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL, + } + defaultPolicies := policies[1:] + lifecycleCases := []struct { + name string + attempt int64 + runID string + completed bool + }{ + {name: "deferred", attempt: -1}, + {name: "unprocessed"}, + {name: "ready", attempt: 1}, + {name: "retry", attempt: 2}, + {name: "running", attempt: 1, runID: "run"}, + {name: "completed", attempt: 1, runID: "run", completed: true}, + {name: "completion before start", attempt: 1, completed: true}, + } + + comparisons := 0 + for _, defaultPolicy := range defaultPolicies { + for _, policy := range policies { + for _, lifecycle := range lifecycleCases { + schedule := defaultSchedule() + schedule.Policies.OverlapPolicy = defaultPolicy + start := newBufferProcessingComparisonStart(now, "start", policy) + start.Attempt = lifecycle.attempt + start.RunId = lifecycle.runID + if lifecycle.completed { + start.Completed = &schedulespb.CompletedResult{} + } + compareBufferProcessing(t, bufferProcessingComparisonInput{ + name: fmt.Sprintf( + "lifecycle=%s policy=%s default=%s", + lifecycle.name, + policy, + defaultPolicy, + ), + schedule: schedule, + bufferedStarts: []*schedulespb.BufferedStart{start}, + lastProcessedTime: now, + }) + comparisons++ + } + } + } + require.Equal(t, 294, comparisons) +} + +func forEachOverlapPolicySequence( + policies []enumspb.ScheduleOverlapPolicy, + length int, + visit func([]enumspb.ScheduleOverlapPolicy), +) { + sequence := make([]enumspb.ScheduleOverlapPolicy, length) + var generate func(int) + generate = func(index int) { + if index == len(sequence) { + visit(sequence) + return + } + for _, policy := range policies { + sequence[index] = policy + generate(index + 1) + } + } + generate(0) +} + +func newBufferProcessingComparisonStart( + now time.Time, + requestID string, + policy enumspb.ScheduleOverlapPolicy, +) *schedulespb.BufferedStart { + return &schedulespb.BufferedStart{ + NominalTime: timestamppb.New(now), + ActualTime: timestamppb.New(now), + DesiredTime: timestamppb.New(now), + RequestId: requestID, + WorkflowId: "workflow-" + requestID, + OverlapPolicy: policy, + } +} + func TestApplyBufferPlan_RevalidatesRequestBeforeMutation(t *testing.T) { schedule := defaultSchedule() schedule.State.LimitedActions = true @@ -867,15 +1218,15 @@ func TestApplyBufferPlan_RevalidatesRequestBeforeMutation(t *testing.T) { return nil })) - var staleDecisions int64 + var invalidatedDecisions int64 require.NoError(t, env.updateScheduler(func(s *scheduler.Scheduler, ctx chasm.MutableContext) error { invoker := s.Invoker.Get(ctx) invoker.BufferedStarts[0].Attempt = 1 - staleDecisions = apply(ctx, s, invoker) + invalidatedDecisions = apply(ctx, s, invoker) return nil })) - require.Equal(t, int64(1), staleDecisions) + require.Equal(t, int64(1), invalidatedDecisions) require.NoError(t, env.readScheduler(func(s *scheduler.Scheduler, ctx chasm.Context) error { require.Equal(t, int64(1), s.Invoker.Get(ctx).BufferedStarts[0].GetAttempt()) require.Equal(t, int64(1), s.Schedule.State.GetRemainingActions()) @@ -888,14 +1239,14 @@ func compareBufferProcessing(t *testing.T, input bufferProcessingComparisonInput t.Helper() legacy := runBufferProcessing(t, input, false) planned := runBufferProcessing(t, input, true) - require.True(t, proto.Equal(legacy.schedulerState, planned.schedulerState), "Scheduler state mismatch\nlegacy: %v\nplanned: %v", legacy.schedulerState, planned.schedulerState) - require.True(t, proto.Equal(legacy.invokerState, planned.invokerState), "Invoker state mismatch\nlegacy: %v\nplanned: %v", legacy.invokerState, planned.invokerState) - require.Equal(t, legacy.tasks, planned.tasks) - require.Equal(t, legacy.actionRequests, planned.actionRequests) - require.Equal(t, legacy.metrics, planned.metrics) - require.Equal(t, legacy.outcomes, planned.outcomes) - require.Equal(t, legacy.remainingDelta, planned.remainingDelta) - require.Equal(t, legacy.conflictTokenDelta, planned.conflictTokenDelta) + require.True(t, proto.Equal(legacy.schedulerState, planned.schedulerState), "%s: Scheduler state mismatch\nlegacy: %v\nplanned: %v", input.name, legacy.schedulerState, planned.schedulerState) + require.True(t, proto.Equal(legacy.invokerState, planned.invokerState), "%s: Invoker state mismatch\nlegacy: %v\nplanned: %v", input.name, legacy.invokerState, planned.invokerState) + require.Equal(t, legacy.tasks, planned.tasks, input.name) + require.Equal(t, legacy.actionRequests, planned.actionRequests, input.name) + require.Equal(t, legacy.metrics, planned.metrics, input.name) + require.Equal(t, legacy.outcomes, planned.outcomes, input.name) + require.Equal(t, legacy.remainingDelta, planned.remainingDelta, input.name) + require.Equal(t, legacy.conflictTokenDelta, planned.conflictTokenDelta, input.name) } func runBufferProcessing(t *testing.T, input bufferProcessingComparisonInput, planner bool) normalizedBufferProcessing { @@ -923,7 +1274,6 @@ func runBufferProcessing(t *testing.T, input bufferProcessingComparisonInput, pl env.NodeBackend.TasksByCategory = nil config := defaultConfig() - config.EnableBufferPlanner = func(string) bool { return planner } recorder := metricstest.NewCaptureHandler() capture := recorder.StartCapture() defer recorder.StopCapture(capture) @@ -931,7 +1281,11 @@ func runBufferProcessing(t *testing.T, input bufferProcessingComparisonInput, pl Config: config, MetricsHandler: recorder, BaseLogger: env.Logger, }) initialRemaining := env.Scheduler.Schedule.State.GetRemainingActions() - require.NoError(t, handler.Execute(ctx, invoker, chasm.TaskAttributes{}, &schedulerpb.InvokerProcessBufferTask{})) + if planner { + require.NoError(t, handler.Execute(ctx, invoker, chasm.TaskAttributes{}, &schedulerpb.InvokerProcessBufferTask{})) + } else { + require.NoError(t, handler.ExecuteProcessBufferLegacyForTest(ctx, invoker)) + } require.NoError(t, env.CloseTransaction()) return normalizedBufferProcessing{ diff --git a/chasm/lib/scheduler/invoker_tasks.go b/chasm/lib/scheduler/invoker_tasks.go index ff595df7a06..a358cf50a94 100644 --- a/chasm/lib/scheduler/invoker_tasks.go +++ b/chasm/lib/scheduler/invoker_tasks.go @@ -22,9 +22,7 @@ import ( "go.temporal.io/server/common/log/tag" "go.temporal.io/server/common/metrics" "go.temporal.io/server/common/resource" - "go.temporal.io/server/common/util" queueerrors "go.temporal.io/server/service/history/queues/errors" - legacyscheduler "go.temporal.io/server/service/worker/scheduler" "go.uber.org/fx" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -460,24 +458,10 @@ func (h *InvokerProcessBufferTaskHandler) Execute( return queueerrors.NewUnprocessableTaskError("schedules must have an Action set") } - var result processBufferResult - if h.config.EnableBufferPlanner != nil && h.config.EnableBufferPlanner(scheduler.Namespace) { - tweakables := h.config.Tweakables(scheduler.Namespace) - snapshot := newBufferProcessingSnapshot(invoker, scheduler, catchupWindow(scheduler, tweakables)) - plan := schedulerinternal.PlanBufferProcessing(snapshot, ctx.Now(invoker)) - result = applyBufferPlan(ctx, scheduler, invoker, plan).result - } else { - result = h.processBufferLegacy(ctx, invoker, scheduler) - var totalMissedCatchup int64 - for _, count := range result.missedCatchupByActionRunning { - totalMissedCatchup += count - } - scheduler.recordActionResult(&schedulerActionResult{ - overlapSkipped: result.overlapSkipped, - missedCatchupWindow: totalMissedCatchup, - }) - invoker.recordProcessBufferResult(ctx, &result) - } + tweakables := h.config.Tweakables(scheduler.Namespace) + snapshot := newBufferProcessingSnapshot(invoker, scheduler, catchupWindow(scheduler, tweakables)) + plan := schedulerinternal.PlanBufferProcessing(snapshot, ctx.Now(invoker)) + result := applyBufferPlan(ctx, scheduler, invoker, plan).result h.recordBufferProcessingMetrics(scheduler, result) return nil @@ -505,97 +489,6 @@ func (h *InvokerProcessBufferTaskHandler) recordBufferProcessingMetrics( } } -// processBuffer resolves the Invoker's buffered starts that haven't yet begun -// execution. This is where the decision is made to drive execution to -// completion, or skip/drop a start. -func (h *InvokerProcessBufferTaskHandler) processBufferLegacy( - ctx chasm.MutableContext, - invoker *Invoker, - scheduler *Scheduler, -) (result processBufferResult) { - runningWorkflows := invoker.runningWorkflowExecutions() - isRunning := len(runningWorkflows) > 0 - result.missedCatchupByActionRunning = make(map[bool]int64) - - // Processing ignores starts that are already executing or backing off. An existing - // deferred BUFFER_ONE start still participates so it can reject later starts. - pendingBufferedStarts := util.FilterSlice(invoker.GetBufferedStarts(), func(start *schedulespb.BufferedStart) bool { - return start.Attempt == 0 || - (start.Attempt == -1 && scheduler.resolveOverlapPolicy(start.GetOverlapPolicy()) == enumspb.SCHEDULE_OVERLAP_POLICY_BUFFER_ONE) - }) - - // Resolve overlap policies and trim BufferedStarts that are skipped by policy. - action := legacyscheduler.ProcessBuffer(pendingBufferedStarts, isRunning, scheduler.resolveOverlapPolicy) - - // ProcessBuffer will drop starts by omitting them from NewBuffer. Start with the - // diff between the input and NewBuffer, and add any executing starts. - keepStarts := make(map[string]struct{}) // request ID -> is present - for _, start := range action.NewBuffer { - keepStarts[start.GetRequestId()] = struct{}{} - } - - // Combine all available starts. - readyStarts := action.OverlappingStarts - if action.NonOverlappingStart != nil { - readyStarts = append(readyStarts, action.NonOverlappingStart) - } - - // Update result metrics. - result.overlapSkipped = action.OverlapSkipped - result.overlapSkippedByPolicy = action.OverlapSkippedByPolicy - - // Add starting workflows to result, trim others. Catchup-window expiry is - // checked before consumeScheduledAction so that a start past its catchup - // window doesn't consume a LimitedActions slot. - for _, start := range readyStarts { - deadline := h.startWorkflowDeadline(ctx, scheduler, start) - if ctx.Now(invoker).After(deadline) { - // Action was buffered in time but expired before execution - // (e.g., due to overlap deferral, retries, or system delay). - // Only emit the metric if the schedule would have run this - // start -- skip paused or action-exhausted schedules. - if start.Manual || scheduler.canTakeScheduledAction() { - // Determine if a running action contributed: either one is still - // running, or the previous action's CloseTime (stored in DesiredTime) - // was already past this start's deadline. - // Note: if no prior action completed, DesiredTime is zero-valued, - // so After(deadline) is false, correctly yielding actionRunning=false. - actionRunning := isRunning || - start.GetDesiredTime().AsTime().After(deadline) - result.missedCatchupByActionRunning[actionRunning]++ - } - result.discardStarts = append(result.discardStarts, start) - result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedMissedCatchup) - continue - } - - // Ensure we can take more actions. Manual actions are always allowed. - if !start.Manual && !scheduler.consumeScheduledAction() { - // Drop buffered automated actions while paused or out of actions. - result.discardStarts = append(result.discardStarts, start) - result.bufferedStartDropReasons = append(result.bufferedStartDropReasons, bufferedStartDroppedPausedOrLimited) - continue - } - - keepStarts[start.GetRequestId()] = struct{}{} - result.startWorkflows = append(result.startWorkflows, start) - } - - result.discardStarts = util.FilterSlice(pendingBufferedStarts, func(start *schedulespb.BufferedStart) bool { - _, keep := keepStarts[start.GetRequestId()] - return !keep - }) - - // Terminate overrides cancel if both are requested. - if action.NeedTerminate { - result.terminateWorkflows = runningWorkflows - } else if action.NeedCancel { - result.cancelWorkflows = runningWorkflows - } - - return -} - // applyBackoff advances start's attempt and BackoffTime based on err and the retry policy. // `now` is the framework clock captured at task start - using time.Now would // diverge from the LastProcessedTime/eligibility comparison in tests. diff --git a/tests/schedule_test.go b/tests/schedule_test.go index a94b7d148fc..60dab633b99 100644 --- a/tests/schedule_test.go +++ b/tests/schedule_test.go @@ -71,9 +71,6 @@ func scheduleCommonOpts(t *testing.T) []testcore.TestOption { // only v1 needs the worker service opts = append(opts, testcore.WithWorkerService("V1 scheduler")) } - if strings.HasPrefix(t.Name(), "TestScheduleCHASM") { - opts = append(opts, testcore.WithDynamicConfig(chasmscheduler.EnableBufferPlanner, true)) - } return opts } From 10de8bf51e36dc332cc3cf00c3d97ecfdde06883 Mon Sep 17 00:00:00 2001 From: "alex.stanfield" <13949480+chaptersix@users.noreply.github.com> Date: Wed, 19 Aug 2026 06:29:18 -0500 Subject: [PATCH 3/3] Complete scheduler buffer planner verification --- .../invoker_process_buffer_task_test.go | 102 ++++++++++-------- 1 file changed, 57 insertions(+), 45 deletions(-) diff --git a/chasm/lib/scheduler/invoker_process_buffer_task_test.go b/chasm/lib/scheduler/invoker_process_buffer_task_test.go index 8d8db226125..8dc6742d640 100644 --- a/chasm/lib/scheduler/invoker_process_buffer_task_test.go +++ b/chasm/lib/scheduler/invoker_process_buffer_task_test.go @@ -977,7 +977,7 @@ func TestProcessBufferTask_LegacyAndPlannerDifferentialDefaultPolicyStateSpace(t } comparisons := 0 for _, defaultPolicy := range policies { - for unspecifiedIndex := 0; unspecifiedIndex < 3; unspecifiedIndex++ { + for unspecifiedIndex := range 3 { forEachOverlapPolicySequence(policies, 2, func(surrounding []enumspb.ScheduleOverlapPolicy) { sequence := make([]enumspb.ScheduleOverlapPolicy, 3) sequence[unspecifiedIndex] = enumspb.SCHEDULE_OVERLAP_POLICY_UNSPECIFIED @@ -1046,52 +1046,64 @@ func TestProcessBufferTask_LegacyAndPlannerDifferentialTimeCapacityStateSpace(t {name: "one", limited: true, remaining: 1}, {name: "two", limited: true, remaining: 2}, } + policyCases := []struct { + name string + policy enumspb.ScheduleOverlapPolicy + defaultPolicy enumspb.ScheduleOverlapPolicy + }{ + {name: "explicit allow all", policy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL}, + {name: "default allow all", defaultPolicy: enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL}, + } comparisons := 0 - for _, catchupWindow := range catchupWindows { - effectiveWindow := max(catchupWindow, 5*time.Second) - for _, deadlineRelation := range timeRelations { - actualTime := now.Add(-effectiveWindow + deadlineRelation.offset) - deadline := actualTime.Add(effectiveWindow) - for _, desiredRelation := range timeRelations { - desiredTime := deadline.Add(-desiredRelation.offset) - for _, manual := range []bool{false, true} { - for _, paused := range []bool{false, true} { - for _, running := range []bool{false, true} { - for _, capacity := range capacityCases { - schedule := defaultSchedule() - schedule.Policies.CatchupWindow = durationpb.New(catchupWindow) - schedule.State.Paused = paused - schedule.State.LimitedActions = capacity.limited - schedule.State.RemainingActions = capacity.remaining - start := newBufferProcessingComparisonStart(now, "pending", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) - start.ActualTime = timestamppb.New(actualTime) - start.DesiredTime = timestamppb.New(desiredTime) - start.Manual = manual - starts := []*schedulespb.BufferedStart{start} - if running { - runningStart := newBufferProcessingComparisonStart(now, "running", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) - runningStart.Attempt = 1 - runningStart.RunId = "running-run" - starts = append([]*schedulespb.BufferedStart{runningStart}, starts...) + for _, policy := range policyCases { + for _, catchupWindow := range catchupWindows { + effectiveWindow := max(catchupWindow, 5*time.Second) + for _, deadlineRelation := range timeRelations { + actualTime := now.Add(-effectiveWindow + deadlineRelation.offset) + deadline := actualTime.Add(effectiveWindow) + for _, desiredRelation := range timeRelations { + desiredTime := deadline.Add(-desiredRelation.offset) + for _, manual := range []bool{false, true} { + for _, paused := range []bool{false, true} { + for _, running := range []bool{false, true} { + for _, capacity := range capacityCases { + schedule := defaultSchedule() + schedule.Policies.OverlapPolicy = policy.defaultPolicy + schedule.Policies.CatchupWindow = durationpb.New(catchupWindow) + schedule.State.Paused = paused + schedule.State.LimitedActions = capacity.limited + schedule.State.RemainingActions = capacity.remaining + start := newBufferProcessingComparisonStart(now, "pending", policy.policy) + start.ActualTime = timestamppb.New(actualTime) + start.DesiredTime = timestamppb.New(desiredTime) + start.Manual = manual + starts := []*schedulespb.BufferedStart{start} + if running { + runningStart := newBufferProcessingComparisonStart(now, "running", enumspb.SCHEDULE_OVERLAP_POLICY_ALLOW_ALL) + runningStart.Attempt = 1 + runningStart.RunId = "running-run" + starts = append([]*schedulespb.BufferedStart{runningStart}, starts...) + } + compareBufferProcessing(t, bufferProcessingComparisonInput{ + name: fmt.Sprintf( + "time capacity policy=%s catchup=%s deadline=%s desired=%s manual=%t paused=%t running=%t capacity=%s", + policy.name, + catchupWindow, + deadlineRelation.name, + desiredRelation.name, + manual, + paused, + running, + capacity.name, + ), + schedule: schedule, + bufferedStarts: starts, + lastProcessedTime: now, + initialConflictToken: 17, + }) + comparisons++ } - compareBufferProcessing(t, bufferProcessingComparisonInput{ - name: fmt.Sprintf( - "time capacity catchup=%s deadline=%s desired=%s manual=%t paused=%t running=%t capacity=%s", - catchupWindow, - deadlineRelation.name, - desiredRelation.name, - manual, - paused, - running, - capacity.name, - ), - schedule: schedule, - bufferedStarts: starts, - lastProcessedTime: now, - initialConflictToken: 17, - }) - comparisons++ } } } @@ -1099,7 +1111,7 @@ func TestProcessBufferTask_LegacyAndPlannerDifferentialTimeCapacityStateSpace(t } } } - require.Equal(t, 864, comparisons) + require.Equal(t, 1728, comparisons) } func TestProcessBufferTask_LegacyAndPlannerDifferentialLifecycleStateSpace(t *testing.T) {