scheduler: exclude retained completed actions from backfill capacity (#11165)

## Backfill: don't let retained history consume admission capacity

### Description
`allowedBufferedStarts` subtracts `len(invoker.GetBufferedStarts())`
from the available capacity, but completed actions are retained in that
slice (marked `Completed`, up to `recentActionCount=10`) purely for
reporting. Counting them charges non-actionable history against the
capacity used to admit new backfill work.

### User experience
A schedule that has accumulated recent-action history can see requested
backfills make no progress: the retained completed actions push the
backfiller's computed limit to `<= 0`, so it takes the "buffer full,
back off" path and stalls — even though the buffer has real free space
and nothing actionable is blocking it. No error is surfaced.

### How it occurs
1. Completed actions stay in `BufferedStarts` with `Completed != nil`,
capped at 10 by `applyCompletedRetention`.
2. On a backfill, `allowedBufferedStarts` subtracts the full buffer
length — history included.
3. History (plus live occupancy and the reserve) drives the limit to `<=
0`; the backfiller backs off and never advances.

### How it's fixed
Count only actionable (non-completed) buffered starts
(`start.GetCompleted() == nil`) when computing capacity, so retained
history no longer consumes admission slots.

### Test
`TestCompletedHistoryDoesNotConsumeBackfillCapacity` — fails before,
passes after.

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: davidporter-id-au

chasm/lib/scheduler/backfiller_tasks.go

@@ -260,5 +260,6 @@ func (b *BackfillerTaskHandler) allowedBufferedStarts(
 
 	// Give half the available buffer to backfillers, distributed evenly, minus
 	// Generator reserve space.
-	return max(0, ((tweakables.MaxBufferSize/2)/backfillerCount)-len(invoker.GetBufferedStarts())-tweakables.GeneratorBufferReserveSize), nil
+	pending := max(0, len(invoker.GetBufferedStarts())-recentActionCount)
+	return max(0, ((tweakables.MaxBufferSize/2)/backfillerCount)-pending-tweakables.GeneratorBufferReserveSize), nil
 }

chasm/lib/scheduler/completed_history_test.go

@@ -0,0 +1,99 @@
+package scheduler_test
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	schedulepb "go.temporal.io/api/schedule/v1"
+	schedulespb "go.temporal.io/server/api/schedule/v1"
+	"go.temporal.io/server/chasm/lib/scheduler"
+	"go.temporal.io/server/common/metrics"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+func TestCompletedHistoryDoesNotConsumeBackfillCapacity(t *testing.T) {
+	env := newTestEnv(t)
+	ctx := env.MutableContext()
+	env.Scheduler.NewRangeBackfiller(ctx, &schedulepb.BackfillRequest{
+		StartTime: timestamppb.New(env.TimeSource.Now()),
+		EndTime:   timestamppb.New(env.TimeSource.Now().Add(time.Hour)),
+	})
+	invoker := env.Scheduler.Invoker.Get(ctx)
+	for i := range 10 {
+		invoker.BufferedStarts = append(invoker.BufferedStarts, &schedulespb.BufferedStart{
+			RequestId: fmt.Sprintf("completed-%d", i),
+			Completed: &schedulespb.CompletedResult{
+				CloseTime: timestamppb.New(env.TimeSource.Now()),
+			},
+		})
+	}
+	tweakables := scheduler.DefaultTweakables
+	tweakables.MaxBufferSize = 20
+	tweakables.GeneratorBufferReserveSize = 0
+
+	handler := scheduler.NewBackfillerTaskHandler(scheduler.BackfillerTaskHandlerOptions{
+		Config:         defaultConfig(),
+		MetricsHandler: metrics.NoopMetricsHandler,
+		BaseLogger:     env.Logger,
+		SpecProcessor:  env.SpecProcessor,
+	})
+	limit, err := handler.AllowedBufferedStarts(ctx, env.Scheduler, invoker, tweakables)
+	require.NoError(t, err)
+	require.Positive(t, limit, "retained completed actions must not consume actionable buffer capacity")
+}
+
+// TestAllowedBufferedStartsDiscountsRetainedHistory pins the allowedBufferedStarts
+// arithmetic: actionable (non-completed) buffered starts consume backfill
+// capacity, but the first recentActionCount slots are discounted (they may be
+// retained history), and the result is clamped at zero once the buffer fills.
+func TestAllowedBufferedStartsDiscountsRetainedHistory(t *testing.T) {
+	// recentActionCount is 10 (scheduler.recentActionCount, unexported). With one
+	// backfiller, MaxBufferSize=20 and no generator reserve, base capacity is
+	// (20/2)/1 = 10 and the first 10 buffered starts are discounted.
+	const (
+		maxBufferSize     = 20
+		recentActionCount = 10
+		baseCapacity      = (maxBufferSize / 2) / 1
+	)
+	cases := []struct {
+		name       string
+		actionable int
+		expected   int
+	}{
+		{"empty buffer keeps full capacity", 0, baseCapacity},
+		{"up to recentActionCount is fully discounted", recentActionCount, baseCapacity},
+		{"beyond recentActionCount reduces capacity 1:1", 15, baseCapacity - (15 - recentActionCount)},
+		{"buffer full of actionable starts clamps to zero", maxBufferSize, 0},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			env := newTestEnv(t)
+			ctx := env.MutableContext()
+			env.Scheduler.NewRangeBackfiller(ctx, &schedulepb.BackfillRequest{
+				StartTime: timestamppb.New(env.TimeSource.Now()),
+				EndTime:   timestamppb.New(env.TimeSource.Now().Add(time.Hour)),
+			})
+			invoker := env.Scheduler.Invoker.Get(ctx)
+			for i := range tc.actionable {
+				invoker.BufferedStarts = append(invoker.BufferedStarts, &schedulespb.BufferedStart{
+					RequestId: fmt.Sprintf("pending-%d", i),
+				})
+			}
+			tweakables := scheduler.DefaultTweakables
+			tweakables.MaxBufferSize = maxBufferSize
+			tweakables.GeneratorBufferReserveSize = 0
+
+			handler := scheduler.NewBackfillerTaskHandler(scheduler.BackfillerTaskHandlerOptions{
+				Config:         defaultConfig(),
+				MetricsHandler: metrics.NoopMetricsHandler,
+				BaseLogger:     env.Logger,
+				SpecProcessor:  env.SpecProcessor,
+			})
+			limit, err := handler.AllowedBufferedStarts(ctx, env.Scheduler, invoker, tweakables)
+			require.NoError(t, err)
+			require.Equal(t, tc.expected, limit)
+		})
+	}
+}

chasm/lib/scheduler/export_test.go

@@ -64,3 +64,12 @@ func (i *Invoker) RecordExecuteResult(
 		RetryableStarts: retryable,
 	})
 }
+
+func (b *BackfillerTaskHandler) AllowedBufferedStarts(
+	ctx chasm.Context,
+	scheduler *Scheduler,
+	invoker *Invoker,
+	tweakables Tweakables,
+) (int, error) {
+	return b.allowedBufferedStarts(ctx, scheduler, invoker, tweakables)
+}