Size the worker fleet against connections that can actually be obtained (#722)

Two defects in the connection-budget guard, both deferred from the #719
review because neither is reachable at the shipped defaults.

The guard admitted budgets its own error message rejected. It refused
below connsPerWorker (2) while telling the operator to raise the pool to
4, so 2 and 3 passed a check whose own advice called them insufficient,
and the max(..., 1) floor then manufactured a worker the pool could not
serve: it pins the only connection for GET_LOCK and waits forever for a
claim connection. That is the stall the guard exists to prevent.

And affordability ignored the leader loops. Three RunIfLeader loops each
pin a pooled connection for the whole process lifetime, so counting them
as available sizes the fleet against connections that are never coming
back.

Connections pinned for the process lifetime are now subtracted before
anything is sized. The count is injected rather than assumed, because
this package should not encode how many sweeps its caller starts; the
number lives next to the lock names it counts, as LeaderGatedLoops, with
a test pinning the two together so a fourth sweep cannot drift from it.

The threshold and the message now quote one number instead of two that
disagreed, and it includes the reservation, so the figure an operator is
told to reach is the figure enforced. With an honest threshold the
max(..., 1) floor stops being a fiction and is gone: the refusal already
guarantees one affordable worker, so a floor could only ever manufacture
one the pool cannot serve.

Verified rather than assumed that the shipped defaults do not change,
since a sizing fix that quietly shrank production would be worse than the
bug: 25 budget, 3 reserved, 4 configured still yields 4 workers, and
there is a case that fails if that stops being true.

One existing test case had to move from the bounds table to the refusal
table: budget 3 was asserted to clamp to one worker, which is defect one
written down as expected behaviour.

Author: getvictor

openspec/changes/reserve-pinned-conns-before-sizing/proposal.md

@@ -0,0 +1,25 @@
+# Size the worker fleet against connections that are actually available
+
+## Why
+
+Two defects in the processor's connection-budget guard, both found by Copilot in the #719 review and deliberately deferred there because neither is reachable at the shipped defaults (issue #722).
+
+**The guard admitted budgets its own error message rejected.** It refused below `connsPerWorker` (2) while telling the operator to raise the pool to `connsPerWorker * workerPoolShareDivisor` (4). Budgets of 2 and 3 passed a check whose own advice called them insufficient, and the `max(..., 1)` floor then manufactured a worker the pool could not serve: it pins the only connection for `GET_LOCK` and waits forever for a claim connection. That is precisely the stall the guard exists to prevent.
+
+**Affordability ignored the leader loops.** In full mode three `RunIfLeader` loops each pin a pooled connection for the whole process lifetime. Confirmed empirically rather than from source: a dev server logs `acquired leadership` for exactly `edr_process_ttl`, `edr_queue_prune` and `edr_retention`. The worker sizing counted those three connections as available, so a small pool could be consumed entirely by the leader locks plus one worker's `GET_LOCK`, with the claim waiting on a connection nobody will release.
+
+Neither is reachable today because `dbMaxOpenConns` is a fixed 25: three leader connections plus four workers at two each is 11 of 25. Both become reachable the moment the pool size becomes configurable or concurrency rises far enough.
+
+## What changes
+
+Connections that are pinned for the process lifetime are subtracted before anything is sized. The count is injected (`ProcessorOptions.ReservedConns`) rather than assumed, because the processor should not encode how many background sweeps its caller starts; the bootstrap that wires those loops is the only thing that knows. The number itself lives next to the lock names it counts, as `pipeline.LeaderGatedLoops`, with a test pinning the two together.
+
+The refusal threshold and the error message now quote one number instead of two that disagreed, and the message includes the reservation so the figure an operator is told to raise the pool to is the figure actually enforced. With an honest threshold the `max(..., 1)` floor stops being a fiction and is gone: the refusal already guarantees at least one worker is affordable, so a floor could only ever manufacture one the pool cannot serve.
+
+## Not a behaviour change at the shipped defaults
+
+Deliberately verified rather than assumed, because a sizing fix that quietly shrank the production fleet would be a worse bug than the one being fixed: 25 budget, 3 reserved, 4 configured still yields 4 workers, and there is a test case that fails if that ever stops being true.
+
+## Why a new requirement
+
+The natural home is `The processor scales across replicas via SKIP LOCKED`, which owns the existing sizing scenarios. It already carries an in-flight MODIFIED delta from `per-host-claim-affinity`, and two in-flight deltas against one requirement is a known archive hazard here: the second to archive silently drops the first's scenarios. The existing scenarios also remain true as written; what is new is the reservation, which is a separable property.

openspec/changes/reserve-pinned-conns-before-sizing/specs/server-availability/spec.md

@@ -0,0 +1,34 @@
+# server-availability delta: worker sizing subtracts connections that are already pinned
+
+## ADDED Requirements
+
+### Requirement: Worker sizing counts only connections that can actually be obtained
+
+When the system sizes its worker fleet against the database connection pool, it SHALL first subtract every connection that is held for the lifetime of the process and therefore never returned to the pool. Today those are the leader-gated periodic sweeps, each of which pins one connection for its advisory lock from boot to shutdown.
+
+Counting a permanently pinned connection as available is not a rounding error, it is the same stall the sizing check exists to prevent: the fleet is sized against connections that are never coming back, so workers pin what is left for their locks and then wait forever for a claim connection that no one will release. The result is a pipeline that boots cleanly and processes nothing.
+
+The reserved count SHALL be supplied by whatever wires those long-lived holders, not assumed by the component doing the sizing, so that adding or removing a sweep cannot leave the sizing silently wrong.
+
+Where the remaining budget cannot serve even one worker, the system SHALL refuse to start rather than reduce the fleet, and the threshold it enforces SHALL be the threshold its error reports, including the reservation. A guard that admits a value its own message calls insufficient is worse than no guard: it produces a running deployment whose configuration was already diagnosed as unusable. Once the refusal is honest, the system SHALL NOT floor the computed worker count to a minimum of one, because the refusal already guarantees one is affordable and a floor could only ever manufacture a worker the pool cannot serve.
+
+#### Scenario: A pool with room for the sweeps but not for a worker is refused
+
+- **GIVEN** a connection pool large enough for the leader-gated sweeps but with too little left for one worker
+- **WHEN** the processor is constructed with a coordinator
+- **THEN** construction fails rather than starting a worker that would stall on its first claim
+- **AND** the error names the pool size the deployment needs, counting the reservation
+
+#### Scenario: A budget the guard's own advice rejects is refused rather than reduced
+
+- **GIVEN** a connection budget smaller than the threshold the refusal message tells operators to reach
+- **WHEN** the processor is constructed with a coordinator
+- **THEN** construction fails
+- **AND** the fleet is not silently reduced to a worker that cannot make progress
+
+#### Scenario: Reserving the long-lived holders does not shrink a healthy deployment
+
+- **GIVEN** the shipped connection pool, the shipped worker count, and the leader-gated sweeps reserved
+- **WHEN** the processor is constructed
+- **THEN** the configured worker count is honored in full
+- **AND** no reduction is reported

openspec/changes/reserve-pinned-conns-before-sizing/tasks.md

@@ -0,0 +1,8 @@
+# Tasks
+
+- [x] Add `ProcessorOptions.ReservedConns` and subtract it before both the refusal check and the affordability calculation.
+- [x] Make the refusal threshold and its error message quote the same number, reservation included.
+- [x] Drop the `max(..., 1)` floor, now that the refusal guarantees one affordable worker.
+- [x] Export `pipeline.LeaderGatedLoops` next to the lock names, and pin it to that list with a test.
+- [x] Set the reservation from the detection bootstrap, which is what wires the loops, and only when a coordinator gates them.
+- [x] Cover: budgets of 2 and 3 refused rather than clamped, a pool consumed by the leader loops refused, reserved connections excluded from sizing, and the shipped default unchanged.

server/detection/bootstrap/bootstrap.go

@@ -219,6 +219,11 @@ func (d *Detection) wireFullMode(deps Deps, store *mysql.Store, intakeH *intake.
 			// deadlocking on it. Passing the real cap lets the processor size itself, and refuse a pool too small for one worker
 			// rather than boot a pipeline that would stall on its first claim.
 			ConnBudget: connBudget(deps.DB),
+			// The leader-gated sweeps wired into the Runner below each pin a pooled connection for the whole process lifetime, so
+			// they are not available to workers however large the pool looks. Reserve them here, where we know they are being
+			// started, rather than letting the processor guess (issue #722). No coordinator means no leader loops and nothing to
+			// reserve.
+			ReservedConns: reservedLeaderConns(deps.Coordinator),
 		},
 	)
 	if err != nil {
@@ -394,6 +399,16 @@ func (d *Detection) RegisterAuthedRoutes(mux httpserver.Router) {
 
 // connBudget reports the MySQL pool's MaxOpenConns for the processor's concurrency clamp, or 0 when there is no handle to ask (the
 // intake-only modes and tests that wire no DB), which skips the clamp.
+// reservedLeaderConns is how many pooled connections the leader-gated sweeps hold for the lifetime of the process. Setup wires all
+// of them unconditionally, so the only question is whether a coordinator exists to gate them: without one they run ungated on this
+// replica and take no lock connection.
+func reservedLeaderConns(coord leader.Coordinator) int {
+	if coord == nil {
+		return 0
+	}
+	return pipeline.LeaderGatedLoops
+}
+
 func connBudget(db *sqlx.DB) int {
 	if db == nil {
 		return 0

server/detection/internal/pipeline/processor.go

@@ -93,6 +93,13 @@ type ProcessorOptions struct {
 	// refuses, because clamping to one worker there would produce exactly the stall the sizing exists to prevent. Zero means unknown
 	// and skips both the sizing and the refusal.
 	ConnBudget int
+	// ReservedConns is how many pooled connections are already spoken for by something else that holds them for its whole
+	// lifetime, and so are not available to workers however large the pool looks. Today that is the coordinator's leader-gated
+	// loops (LeaderGatedLoops), each of which pins one connection for its lock from boot to shutdown.
+	//
+	// It is injected rather than assumed because this package should not encode how many background sweeps its caller starts:
+	// the bootstrap that wires those loops is the only thing that knows. Zero means nothing is reserved (issue #722).
+	ReservedConns int
 }
 
 // connsPerWorker is how many pooled connections one worker occupies inside its critical section: the GET_LOCK connection the
@@ -106,6 +113,15 @@ const connsPerWorker = 2
 // affords 6 workers against a configured 4); it is what keeps a future concurrency increase from quietly consuming the pool.
 const workerPoolShareDivisor = 2
 
+// minConnsForOneWorker is the smallest usable budget: below it there is no worker count that can run, so the configuration is
+// refused rather than clamped. It is connsPerWorker scaled by the share divisor, which is the same arithmetic the affordability
+// calculation uses, so the guard and its error message quote one number instead of two that disagree.
+//
+// They used to disagree: the guard refused below connsPerWorker (2) while its message told the operator to raise the pool to 4, so
+// budgets of 2 and 3 passed a check whose own advice rejected them and were then clamped to a worker that could not make progress
+// (issue #722).
+const minConnsForOneWorker = connsPerWorker * workerPoolShareDivisor
+
 // concurrencyClamp records that the effective worker count came out below the configured one, and why. The constructor decides it but
 // cannot log it: it has no context, and fabricating a background one there is what contextcheck (rightly) rejects. Run emits it once
 // at startup instead, so an operator whose configured concurrency was not honored sees the reason in the logs.
@@ -151,18 +167,25 @@ func NewProcessor(
 	// forever for a claim connection. That is the exact stall the clamp exists to avoid, so refuse the configuration instead. A
 	// deployment that will not boot states its problem; one that boots and silently never processes an event does not.
 	if opts.Coordinator != nil && opts.ConnBudget > 0 {
-		if opts.ConnBudget < connsPerWorker {
+		// Subtract what is already pinned before sizing anything. Connections held by the leader-gated loops are never returned
+		// while the process runs, so counting them as available sizes the fleet against connections that do not exist.
+		available := opts.ConnBudget - opts.ReservedConns
+		if available < minConnsForOneWorker {
 			return nil, fmt.Errorf(
-				"detection processor: MySQL pool of %d connection(s) cannot serve one worker; the per-host claim lock needs %d "+
-					"connections per worker (one pinned by GET_LOCK, one for the claim and flush), so raise the pool to at least %d",
-				opts.ConnBudget, connsPerWorker, connsPerWorker*workerPoolShareDivisor)
+				"detection processor: MySQL pool of %d connection(s) leaves %d after %d reserved for the leader-gated loops, which "+
+					"cannot serve one worker; the per-host claim lock needs %d connections per worker (one pinned by GET_LOCK, one "+
+					"for the claim and flush) and workers take at most half the pool, so raise the pool to at least %d",
+				opts.ConnBudget, available, opts.ReservedConns, connsPerWorker, opts.ReservedConns+minConnsForOneWorker)
 		}
-		if affordable := max(opts.ConnBudget/(connsPerWorker*workerPoolShareDivisor), 1); concurrency > affordable {
+		// No max(..., 1) floor: the refusal above guarantees at least one worker is affordable, so a floor here could only ever
+		// manufacture a worker the pool cannot serve, which is the stall this whole check exists to prevent.
+		if affordable := available / minConnsForOneWorker; concurrency > affordable {
 			clamp = &concurrencyClamp{
 				reason: "detection processor concurrency clamped to the connection budget",
 				attrs: []any{
 					"requested_concurrency", concurrency, "effective_concurrency", affordable,
-					"max_open_conns", opts.ConnBudget, "conns_per_worker", connsPerWorker,
+					"max_open_conns", opts.ConnBudget, "reserved_conns", opts.ReservedConns,
+					"available_conns", available, "conns_per_worker", connsPerWorker,
 					"pool_share_divisor", workerPoolShareDivisor,
 				},
 			}

server/detection/internal/pipeline/processor_hostclaim_test.go

@@ -3,6 +3,7 @@ package pipeline
 import (
 	"context"
 	"errors"
+	"fmt"
 	"strings"
 	"testing"
 
@@ -89,31 +90,38 @@ func TestNewProcessor_ConcurrencyBounds(t *testing.T) {
 	t.Parallel()
 
 	cases := []struct {
-		name        string
-		concurrency int
-		connBudget  int
-		coordinator leader.Coordinator
-		want        int
+		name          string
+		concurrency   int
+		connBudget    int
+		reservedConns int
+		coordinator   leader.Coordinator
+		want          int
 	}{
-		{"non-positive concurrency floors to one worker", 0, 25, stubCoordinator{}, 1},
-		{"the shipped default pool leaves the default worker count alone", 4, 25, stubCoordinator{}, 4},
-		{"budget exactly the worker share leaves it alone", 4, 16, stubCoordinator{}, 4},
-		{"tight budget clamps to the share the pool can serve", 8, 16, stubCoordinator{}, 4},
-		{"a budget serving one worker's share clamps to one", 8, 4, stubCoordinator{}, 1},
-		{"a budget below a full share still leaves one worker", 8, 3, stubCoordinator{}, 1},
-		{"unknown budget skips the sizing", 8, 0, stubCoordinator{}, 8},
-		{"no coordinator forces a single worker", 8, 25, nil, 1},
-		{"a lockless single worker needs no pool share", 8, 2, nil, 1},
+		{"non-positive concurrency floors to one worker", 0, 25, 0, stubCoordinator{}, 1},
+		{"the shipped default pool leaves the default worker count alone", 4, 25, 0, stubCoordinator{}, 4},
+		// The production shape, and the case this must never regress: reserving the leader loops out of the real pool still has
+		// to leave the configured fleet intact, or the fix for issue #722 would quietly shrink every deployment.
+		{"spec:server-availability/worker-sizing-counts-only-connections-that-can-actually-be-obtained/reserving-the-long-lived-holders-does-not-shrink-a-healthy-deployment", 4, 25, LeaderGatedLoops, stubCoordinator{}, 4},
+		{"budget exactly the worker share leaves it alone", 4, 16, 0, stubCoordinator{}, 4},
+		{"tight budget clamps to the share the pool can serve", 8, 16, 0, stubCoordinator{}, 4},
+		{"a budget serving one worker's share clamps to one", 8, 4, 0, stubCoordinator{}, 1},
+		// The reservation is subtracted BEFORE sizing: 16 looks like 4 workers, but 8 of it is already pinned for the whole
+		// process lifetime and is never coming back.
+		{"reserved connections are not available to workers", 8, 16, 8, stubCoordinator{}, 2},
+		{"unknown budget skips the sizing", 8, 0, 0, stubCoordinator{}, 8},
+		{"no coordinator forces a single worker", 8, 25, 0, nil, 1},
+		{"a lockless single worker needs no pool share", 8, 2, 0, nil, 1},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 			proc, err := NewProcessor(&scriptedEventLog{}, nil, nil, ProcessorOptions{
-				Logger:      discardLogger(),
-				Batch:       10,
-				Concurrency: tc.concurrency,
-				Coordinator: tc.coordinator,
-				ConnBudget:  tc.connBudget,
+				Logger:        discardLogger(),
+				Batch:         10,
+				Concurrency:   tc.concurrency,
+				Coordinator:   tc.coordinator,
+				ConnBudget:    tc.connBudget,
+				ReservedConns: tc.reservedConns,
 			})
 			require.NoError(t, err)
 			assert.Equal(t, tc.want, proc.concurrency)
@@ -138,31 +146,55 @@ func TestNewProcessor_RefusesPoolBelowOneWorker(t *testing.T) {
 	t.Parallel()
 
 	cases := []struct {
-		name       string
-		connBudget int
+		name          string
+		connBudget    int
+		reservedConns int
 	}{
-		{"a single-connection pool cannot hold a lock and claim at once", 1},
-		{"a pool one short of a worker's needs is still a stall", connsPerWorker - 1},
+		{"a single-connection pool cannot hold a lock and claim at once", 1, 0},
+		{"a pool one short of a worker's needs is still a stall", connsPerWorker - 1, 0},
+		// These two used to be ADMITTED by a guard whose own error message called them insufficient: it refused below
+		// connsPerWorker (2) while telling the operator to raise the pool to 4, then clamped 2 and 3 to a worker that could not
+		// make progress. A guard must not accept a value its message rejects (issue #722).
+		{"spec:server-availability/worker-sizing-counts-only-connections-that-can-actually-be-obtained/a-budget-the-guard-s-own-advice-rejects-is-refused-rather-than-reduced", connsPerWorker, 0},
+		{"one short of a full worker share is refused too", minConnsForOneWorker - 1, 0},
+		// Room for the leader loops but not for a worker afterwards. Before the reservation existed this looked like a healthy
+		// pool and produced a fleet that would stall on its first claim.
+		{"spec:server-availability/worker-sizing-counts-only-connections-that-can-actually-be-obtained/a-pool-with-room-for-the-sweeps-but-not-for-a-worker-is-refused", LeaderGatedLoops + 1, LeaderGatedLoops},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
 			proc, err := NewProcessor(&scriptedEventLog{}, nil, nil, ProcessorOptions{
-				Logger:      discardLogger(),
-				Batch:       10,
-				Concurrency: 4,
-				Coordinator: stubCoordinator{},
-				ConnBudget:  tc.connBudget,
+				Logger:        discardLogger(),
+				Batch:         10,
+				Concurrency:   4,
+				Coordinator:   stubCoordinator{},
+				ConnBudget:    tc.connBudget,
+				ReservedConns: tc.reservedConns,
 			})
 			require.Error(t, err, "a pool that cannot serve one worker must not produce a processor")
 			assert.Nil(t, proc)
-			// The operator's next action is to raise the pool, so the error has to carry the number to raise it to.
+			// The operator's next action is to raise the pool, so the error has to carry the number to raise it to, and that
+			// number has to be the one the guard actually enforces. The two disagreeing is what let 2 and 3 through.
 			assert.Contains(t, err.Error(), "raise the pool to at least",
 				"the refusal must tell the operator what to change, not just that something is wrong")
+			assert.Contains(t, err.Error(), fmt.Sprintf("at least %d", tc.reservedConns+minConnsForOneWorker),
+				"the advertised threshold must be the one enforced, reservation included")
 		})
 	}
 }
 
+// TestLeaderGatedLoopsMatchesLockNames pins the exported connection reservation to the loops it counts. The processor subtracts
+// LeaderGatedLoops from the pool before sizing its fleet, so if a fourth leader-gated sweep were added without bumping it, the
+// processor would size itself against a connection that is permanently pinned and could stall on its first claim (issue #722). The
+// count is in a different file from the Run that starts the loops, which is exactly the drift this catches.
+func TestLeaderGatedLoopsMatchesLockNames(t *testing.T) {
+	t.Parallel()
+	locks := []string{lockProcessTTL, lockRetention, lockQueuePrune}
+	assert.Len(t, locks, LeaderGatedLoops,
+		"every leader-gated lock pins a pooled connection, so the reservation must count all of them")
+}
+
 // A non-positive batch would spin the drain loop, since an empty claim returns 0 and 0 >= 0 never breaks.
 func TestNewProcessor_BatchFloorsToOne(t *testing.T) {
 	t.Parallel()