Separate CHASM scheduler buffer planning

chasm/lib/scheduler/buffer_planner.go

@@ -0,0 +1,207 @@
+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"
+)
+
+// appliedBufferPlan collects the live-state result and the number of decisions invalidated during revalidation.
+type appliedBufferPlan struct {
+	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{
+		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,
+	}
+}
+
+// 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,
+	invoker *Invoker,
+	plan schedulerinternal.BufferPlan,
+) appliedBufferPlan {
+	applied := newAppliedBufferPlan()
+	currentSnapshot := newBufferProcessingSnapshot(invoker, scheduler, plan.Snapshot.CatchupWindow)
+	if !bufferProcessingSnapshotsEqual(plan.Snapshot, currentSnapshot) {
+		return invalidateBufferPlan(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() {
+			continue
+		}
+		start, ok := popMatchingBufferedStart(startsByRequestID, decision)
+		if !ok {
+			applied.recordInvalidatedDecision()
+			continue
+		}
+
+		if decision.ConsumesScheduledAction && !scheduler.consumeScheduledAction() {
+			applied.recordInvalidatedDecision()
+			continue
+		}
+
+		applyBufferDecision(&applied.result, decision, start)
+	}
+	if applied.invalidatedDecisions == 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 invalidateBufferPlan(plan schedulerinternal.BufferPlan, applied appliedBufferPlan) appliedBufferPlan {
+	for _, decision := range plan.Decisions {
+		if decision.MutatesState() {
+			applied.recordInvalidatedDecision()
+		}
+	}
+	return applied
+}
+
+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,
+) (*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)
+	default:
+	}
+	recordAppliedDecisionMetrics(result, decision)
+}
+
+func recordAppliedDecisionMetrics(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}
+}

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

chasm/lib/scheduler/export_test.go

@@ -7,7 +7,10 @@ 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"
+	"go.temporal.io/server/common/metrics"
+	queueerrors "go.temporal.io/server/service/history/queues/errors"
 	legacyscheduler "go.temporal.io/server/service/worker/scheduler"
 )
 
@@ -72,6 +75,47 @@ 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).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,