Separate CHASM scheduler action capacity APIs

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.
 //

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

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

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

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)

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