From d0d4e474e5715cb10639a9c3bcd3abdc80fd47e9 Mon Sep 17 00:00:00 2001 From: "alex.stanfield" <13949480+chaptersix@users.noreply.github.com> Date: Tue, 18 Aug 2026 21:21:03 -0500 Subject: [PATCH] Separate CHASM scheduler action capacity APIs --- .../invoker_process_buffer_task_test.go | 2 +- chasm/lib/scheduler/invoker_tasks.go | 6 +- chasm/lib/scheduler/scheduler.go | 47 ++++---- ...scheduler_action_capacity_internal_test.go | 101 ++++++++++++++++++ chasm/lib/scheduler/spec_processor.go | 2 +- tests/schedule_test.go | 2 +- 6 files changed, 126 insertions(+), 34 deletions(-) create mode 100644 chasm/lib/scheduler/scheduler_action_capacity_internal_test.go diff --git a/chasm/lib/scheduler/invoker_process_buffer_task_test.go b/chasm/lib/scheduler/invoker_process_buffer_task_test.go index e4eef043dce..87735003005 100644 --- a/chasm/lib/scheduler/invoker_process_buffer_task_test.go +++ b/chasm/lib/scheduler/invoker_process_buffer_task_test.go @@ -598,7 +598,7 @@ func TestProcessBufferTask_NeedsTerminate(t *testing.T) { } // Past-catchup automated starts must drop WITHOUT consuming a LimitedActions -// slot. Regression for the order-of-checks bug where useScheduledAction(true) +// slot. Regression for the order-of-checks bug where action capacity consumption // fired before the catchup-window check, decrementing RemainingActions for // starts that never ran. // diff --git a/chasm/lib/scheduler/invoker_tasks.go b/chasm/lib/scheduler/invoker_tasks.go index 6da78f9d735..e4a82aa9f0e 100644 --- a/chasm/lib/scheduler/invoker_tasks.go +++ b/chasm/lib/scheduler/invoker_tasks.go @@ -529,7 +529,7 @@ func (h *InvokerProcessBufferTaskHandler) processBuffer( result.overlapSkippedByPolicy = action.OverlapSkippedByPolicy // Add starting workflows to result, trim others. Catchup-window expiry is - // checked before useScheduledAction so that a start past its catchup + // 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()) @@ -540,7 +540,7 @@ func (h *InvokerProcessBufferTaskHandler) processBuffer( // (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.useScheduledAction(false) { + 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. @@ -556,7 +556,7 @@ func (h *InvokerProcessBufferTaskHandler) processBuffer( } // Ensure we can take more actions. Manual actions are always allowed. - if !start.Manual && !scheduler.useScheduledAction(true) { + 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)) diff --git a/chasm/lib/scheduler/scheduler.go b/chasm/lib/scheduler/scheduler.go index 017b7cef668..8ecdd5e86e5 100644 --- a/chasm/lib/scheduler/scheduler.go +++ b/chasm/lib/scheduler/scheduler.go @@ -409,40 +409,31 @@ func (s *Scheduler) NewImmediateBackfiller( return backfiller } -// useScheduledAction returns true when the Scheduler should allow scheduled -// actions to be taken. -// -// When decrement is true, the schedule's state's `RemainingActions` counter is -// decremented when an action can be taken. When decrement is false, no state -// is mutated. -func (s *Scheduler) useScheduledAction(decrement bool) bool { +// canTakeScheduledAction returns true when the Scheduler should allow a +// scheduled action to be taken without mutating state. +func (s *Scheduler) canTakeScheduledAction() bool { scheduleState := s.Schedule.GetState() - // If paused, don't do anything. - if scheduleState.Paused { - return false - } + return !scheduleState.Paused && + (!scheduleState.LimitedActions || scheduleState.RemainingActions > 0) +} - // If unlimited actions, allow. - if !scheduleState.LimitedActions { - return true +// consumeScheduledAction takes one scheduled action when allowed. +func (s *Scheduler) consumeScheduledAction() bool { + if !s.canTakeScheduledAction() { + return false } - // Otherwise check and decrement limit. - if scheduleState.RemainingActions > 0 { - if decrement { - scheduleState.RemainingActions-- + scheduleState := s.Schedule.GetState() + if scheduleState.LimitedActions { + scheduleState.RemainingActions-- - // The conflict token is updated because a client might be in the process of - // preparing an update request that increments their schedule's RemainingActions - // field. - s.updateConflictToken() - } - return true + // The conflict token is updated because a client might be in the process of + // preparing an update request that increments their schedule's RemainingActions + // field. + s.updateConflictToken() } - - // No actions left - return false + return true } func (s *Scheduler) getCompiledSpec(specBuilder *scheduler.SpecBuilder) (*scheduler.CompiledSpec, error) { @@ -564,7 +555,7 @@ func (s *Scheduler) getIdleExpiration( ) (time.Time, bool) { if idleTime == 0 || s.isHeldOpen() || - (!nextWakeup.IsZero() && s.useScheduledAction(false)) { + (!nextWakeup.IsZero() && s.canTakeScheduledAction()) { return time.Time{}, false } return s.idleDeadline(ctx, idleTime), true diff --git a/chasm/lib/scheduler/scheduler_action_capacity_internal_test.go b/chasm/lib/scheduler/scheduler_action_capacity_internal_test.go new file mode 100644 index 00000000000..5b3a91a4448 --- /dev/null +++ b/chasm/lib/scheduler/scheduler_action_capacity_internal_test.go @@ -0,0 +1,101 @@ +package scheduler + +import ( + "testing" + + "github.com/stretchr/testify/require" + schedulepb "go.temporal.io/api/schedule/v1" + schedulerpb "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1" +) + +func TestSchedulerCanTakeScheduledActionDoesNotMutateState(t *testing.T) { + tests := []struct { + name string + paused bool + limitedActions bool + remainingActions int64 + want bool + }{ + { + name: "paused", + paused: true, + }, + { + name: "unlimited", + remainingActions: 0, + want: true, + }, + { + name: "limited with no actions", + limitedActions: true, + }, + { + name: "limited with remaining actions", + limitedActions: true, + remainingActions: 2, + want: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + scheduler := newActionCapacityTestScheduler(tt.paused, tt.limitedActions, tt.remainingActions) + initialToken := scheduler.ConflictToken + + require.Equal(t, tt.want, scheduler.canTakeScheduledAction()) + require.Equal(t, tt.remainingActions, scheduler.Schedule.State.RemainingActions) + require.Equal(t, initialToken, scheduler.ConflictToken) + }) + } +} + +func TestSchedulerConsumeScheduledActionCounterAndTokenDeltas(t *testing.T) { + tests := []struct { + name string + consumptions int + }{ + {name: "zero"}, + {name: "one", consumptions: 1}, + {name: "multiple", consumptions: 3}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + initialActions := int64(tt.consumptions) + scheduler := newActionCapacityTestScheduler(false, true, initialActions) + initialToken := scheduler.ConflictToken + + for range tt.consumptions { + require.True(t, scheduler.consumeScheduledAction()) + } + require.False(t, scheduler.consumeScheduledAction()) + + require.Zero(t, scheduler.Schedule.State.RemainingActions) + require.Equal(t, initialToken+int64(tt.consumptions), scheduler.ConflictToken) + }) + } +} + +func TestSchedulerConsumeScheduledActionUnlimitedDoesNotMutateState(t *testing.T) { + scheduler := newActionCapacityTestScheduler(false, false, 0) + initialToken := scheduler.ConflictToken + + require.True(t, scheduler.consumeScheduledAction()) + require.Zero(t, scheduler.Schedule.State.RemainingActions) + require.Equal(t, initialToken, scheduler.ConflictToken) +} + +func newActionCapacityTestScheduler(paused, limitedActions bool, remainingActions int64) *Scheduler { + return &Scheduler{ + SchedulerState: &schedulerpb.SchedulerState{ + Schedule: &schedulepb.Schedule{ + State: &schedulepb.ScheduleState{ + Paused: paused, + LimitedActions: limitedActions, + RemainingActions: remainingActions, + }, + }, + ConflictToken: 41, + }, + } +} diff --git a/chasm/lib/scheduler/spec_processor.go b/chasm/lib/scheduler/spec_processor.go index 6908d56a03d..383f1113640 100644 --- a/chasm/lib/scheduler/spec_processor.go +++ b/chasm/lib/scheduler/spec_processor.go @@ -98,7 +98,7 @@ func (s *SpecProcessorImpl) ProcessTimeRange( // Skip over entire time range if paused or no actions can be taken. // // Manual (backfill/patch) runs are always buffered here. - if !scheduler.useScheduledAction(false) && !manual { + if !scheduler.canTakeScheduledAction() && !manual { // Use end as last action time so that we don't reprocess time spent paused. next, err := s.NextTime(scheduler, end) wakeup, err := s.checkNextScheduleResult(scheduler, metricsHandler, next, err) diff --git a/tests/schedule_test.go b/tests/schedule_test.go index ea254106ef4..60dab633b99 100644 --- a/tests/schedule_test.go +++ b/tests/schedule_test.go @@ -4366,7 +4366,7 @@ func testTriggerImmediatelyOnActiveSchedule(t *testing.T, newContext contextFact // testTriggerImmediatelyOnPausedSchedule verifies that TriggerImmediately fires // an action even when the schedule is paused. Manual starts bypass the paused -// gate via useScheduledAction's Manual carve-out in processBuffer. +// gate because processBuffer does not consume action capacity for manual starts. func testTriggerImmediatelyOnPausedSchedule(t *testing.T, newContext contextFactory) { s := newScheduleEnv(t, scheduleCommonOpts(t)...)