fix(stovepipe): admit before counting the build slot

doc/rfc/stovepipe/steps/process.md

@@ -29,8 +29,9 @@ For a delivery carrying request id `R`:
    - defer (hold the delivery) -> re-check on redelivery until the slot frees (admit) or a newer head supersedes it. See [Waiting for a slot](#waiting-for-a-slot).
 7. Admit R:
    a. Derive build strategy + baseline (see "Build-strategy decision").
-   b. CAS the Queue row: in_flight_count += 1.
-   c. CAS the Request: accepted -> processing, persist build_strategy + base_uri.
+   b. CAS the Request: accepted -> processing, persist build_strategy + base_uri.
+      - lost to another writer (no longer accepted) -> ack; that writer owns the outcome and nothing is counted.
+   c. CAS the Queue row: in_flight_count += 1, reloading and retrying on a version conflict without re-gating.
    d. Announce validation start on the hook topic (see "Hooks").
    e. Publish R to build.
    f. ack.
@@ -159,21 +160,24 @@ Only an admit announces. A Request that coalescing supersedes never reaches step
 
 Every branch is safe under redelivery:
 
-- **accepted, no strategy** → full admit path. On a crash after incrementing `in_flight_count` but before persisting `processing`, redelivery re-reads `accepted` and re-runs; the increment re-applies only if the count CAS hasn't already moved (see integrity below).
+- **accepted, no strategy** → full admit path. Nothing has been written yet: a crash before the request CAS leaves no trace, and redelivery re-runs the gate and the admit from scratch.
 - **processing** → re-announce the start event, re-publish to `build`, ack. The `build` consumer is keyed on the request id and idempotent, so a duplicate publish is harmless, and the start event's id is derived from the transition rather than the clock, so a re-announce carries the id the first attempt would have and consumers dedupe on it. Re-announcing here is what makes the event at-least-once rather than at-most-once: this is the only branch a redelivery takes once `processing` is durable, so an admit that failed after the state write would otherwise lose the event for good.
 - **terminal** (superseded / recorded) → ack, no-op.
 - **deferred (waiting for slot)** → no state or count change; pure deferral (re-enters when the held delivery comes due).
 
-The window to handle is "count incremented, state not yet `processing`". Admit does the increment and the state transition as two ordered CAS writes, and the decrement is tied to the state transition, not a side counter (see integrity below).
+The window to handle is "state `processing`, count not yet incremented". Admit does the state transition and the increment as two ordered CAS writes, and a crash between them leaves the count low by one rather than high (see integrity below).
 
 ## `in_flight_count` integrity
 
-`in_flight_count` is a cache; the source of truth is **the set of non-terminal Request rows for the Queue**. Two rules keep it from drifting:
+`in_flight_count` is a cache; the source of truth is **the set of non-terminal Request rows for the Queue**. `Queue` and `Request` are separate entities with no cross-entity transaction, so the admit's two writes cannot be atomic, and their order picks the direction the count can drift. Admit writes the request first: the `accepted → processing` CAS *is* the admit, and the increment follows it.
 
-1. **Decrement is bound to the terminal transition.** The single CAS that moves a Request non-terminal → terminal (in `record` or the DLQ reconciler) also decrements. Being CAS-guarded, it fires exactly once per Request even under redelivery.
-2. **Increment is bound to the admit transition.** `process` increments only on the `accepted → processing` CAS; a redelivery of an already-`processing` Request takes step 3 and does not increment again.
+1. **The count is a lower bound on in-flight validations, with one stalled-delivery exception.** A crash or hard failure after the request CAS and before the increment leaves an admitted Request uncounted. Its redelivery reads `processing`, takes step 3, and never increments, so the Queue over-admits by one until the drift clears. No slot is counted for a Request that is still `accepted`, so a redelivery never finds the gate closed by its own claim. The one way the count goes high is a delivery that stalls between the request CAS and the increment for longer than the visibility timeout: its redelivery takes step 3, the build runs to terminal, the zero clamp absorbs the release, and the stalled increment then lands and counts a slot nothing will release. That needs the stall to outlast the redelivery plus a whole validation. A terminal re-check of the Request right before the increment would narrow it to the gap between two reads, at the cost of one extra read per admit; the owner-keyed set (PR #966) closes it.
+2. **A low count reconverges at the zero clamp.** Every Request that reaches `processing` is released when it goes terminal (by `buildsignal` or the DLQ reconciler; see [buildsignal.md](buildsignal.md)), whether or not its increment landed. Both release paths clamp at zero, so the release of an uncounted Request is absorbed by the time the Queue drains, and the count returns to the true value.
+3. **The gate is best-effort.** It is a read of the count, not part of the increment's CAS, so two deliveries for different heads that both read an open gate before either increments are both admitted, over-admitting by one transiently. The per-Queue partition normally serializes `process` deliveries (see [the single-writer property](#input-partitioning-and-the-single-writer-property)), so the overlap this needs is rare.
 
-On a crash between admit and `record`, the Request stays non-terminal; visibility-timeout redelivery drives it forward, and the fail-closed DLQ path eventually forces it terminal, decrementing as it does. The count can drift high only transiently and self-heals as stuck Requests terminate. A reconciler that recomputes the count from non-terminal rows can be added later if drift proves real, but isn't required for MVP.
+Over-admission by one is the failure mode this pipeline prefers to a wedge, here and in the release paths.
+
+An exact count needs per-request slot ownership on the Queue row: an owner-keyed set of request ids (PR #966) that the gate counts and each terminal transition removes from, at the cost of a schema change. A count leaked high before this design (a slot claimed by a Request that never reached `processing`) cannot be told apart from a live one, so a Queue already wedged that way needs a one-time operator reset of `in_flight_count`.
 
 ## Edge cases
 

stovepipe/controller/buildsignal/buildsignal.go

@@ -339,8 +339,8 @@ func (c *Controller) markOutcome(ctx context.Context, store storage.Storage, req
 // releaseBuildSlot CAS-decrements the queue's in_flight_count, reopening the process
 // concurrency gate now that this request's build is over. It decrements relatively
 // (preserving concurrent updates), clamps at zero, and retries on version conflicts.
-// Unlike process's unwind-path release this is not best-effort: the caller must not
-// mark the request terminal if the slot was not freed, so a hard failure is returned.
+// It is not best-effort: the caller must not mark the request terminal if the slot
+// was not freed, so a hard failure is returned.
 func (c *Controller) releaseBuildSlot(ctx context.Context, store storage.Storage, queueName string) error {
 	queueStore := store.GetQueueStore()
 

stovepipe/controller/dlq/dlq.go

@@ -66,14 +66,10 @@ func TopicKey(main consumer.TopicKey) consumer.TopicKey {
 // slot is released first.
 //
 // Processing is the only non-terminal state that can own a slot, so the condition is not a
-// narrowing of some broader set: process claims the slot and CAS-marks accepted→processing,
-// releasing its own claim if that CAS never lands, and the exits from processing are the
-// terminal outcomes, which release the slot themselves. Widening the release to accepted
-// would decrement for the far more common request that never claimed a slot, over-admitting
-// against MaxConcurrent. The one case that escapes both this reconciler and process's
-// compensation is a hard crash between the two admit writes, which leaves an accepted
-// request holding a slot that nothing here can tell apart from a request that never
-// claimed one; distinguishing them needs per-request slot ownership on the row.
+// narrowing of some broader set: process CAS-marks accepted→processing before it counts the
+// slot, and the exits from processing are the terminal outcomes, which release the slot
+// themselves. Widening the release to accepted would decrement for a request that never
+// counted a slot, over-admitting against MaxConcurrent.
 //
 // Queue and Request are separate entities with no cross-entity
 // transaction, so the two writes cannot be atomic and the ordering picks which crash

stovepipe/controller/process/process.go

@@ -225,62 +225,29 @@ func (c *Controller) coalesce(ctx context.Context, store storage.Storage, reques
 	return true, nil
 }
 
-// admitLatestHead runs the gate-then-admit workflow for the latest head: claim a build
-// slot, mark the request processing, and publish it to build. Every queue-row reload
-// re-runs coalesce-then-gate, so a slot is never spent on a now-stale head; a closed gate
-// defers by holding the delivery (redeliver after the gate wait delay) rather than failing.
+// admitLatestHead holds the delivery while the build-slot gate is closed; otherwise it marks the
+// request processing, counts its build slot, and publishes it to build.
 func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, delivery consumer.Delivery, request entity.Request, queueRow entity.Queue, cfg entity.QueueConfig) error {
-	var sc sourcecontrol.SourceControl
-	var strategy entity.BuildStrategy
-	var baseURI string
-	var err error
-
-	for {
-		if queueRow.InFlightCount >= cfg.MaxConcurrent {
-			return c.holdForBuildSlot(ctx, delivery, request, queueRow.InFlightCount, cfg.GateWaitDelayMs)
-		}
-
-		if queueRow.LastGreenURI != "" && sc == nil {
-			sc, err = c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue})
-			if err != nil {
-				metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1,
-					metrics.TagsFromContext(ctx, metrics.NewTag("stage", "resolve"))...,
-				)
-				return fmt.Errorf("failed to resolve source control for queue %s: %w", request.Queue, err)
-			}
-		}
-
-		strategy, baseURI, err = c.deriveBuildStrategy(ctx, sc, queueRow, request)
-		if err != nil {
-			return err
-		}
+	if queueRow.InFlightCount >= cfg.MaxConcurrent {
+		return c.holdForBuildSlot(ctx, delivery, request, queueRow.InFlightCount, cfg.GateWaitDelayMs)
+	}
 
-		err = c.claimBuildSlot(ctx, store, &queueRow)
-		if err == nil {
-			break
-		}
-		if !errors.Is(err, storage.ErrVersionMismatch) {
-			return err
-		}
-		// claimBuildSlot reloaded queueRow. Re-coalesce: supersede if a newer head arrived,
-		// otherwise loop to re-check the gate.
-		superseded, err := c.coalesce(ctx, store, request, queueRow.LatestRequestID)
-		if err != nil || superseded {
-			return err
-		}
+	strategy, baseURI, err := c.deriveBuildStrategy(ctx, queueRow, request)
+	if err != nil {
+		return err
 	}
 
+	// The request transition is the admit and the count follows it, so a failure between the two
+	// writes leaves the count low (over-admission by one) rather than a slot nobody owns (a wedge). The
+	// gate is therefore best-effort: two heads that read it open in the same window both pass, and
+	// the release paths' zero clamp reconverges a low count. See doc/rfc/stovepipe/steps/process.md
+	// (in_flight_count integrity); PR #966's owner-keyed slot set is the exact alternative.
 	transitioned, err := c.markProcessing(ctx, store, &request, strategy, baseURI)
-	if err != nil {
-		// Slot claimed but never admitted: release best-effort so the slot isn't leaked
-		// (a redelivery would find the gate closed by its own claim and nothing decrements it).
-		c.releaseBuildSlot(ctx, store, request.Queue)
+	if err != nil || !transitioned {
 		return err
 	}
-	if !transitioned {
-		// Lost the admit race: another delivery advanced this request. Release and skip.
-		c.releaseBuildSlot(ctx, store, request.Queue)
-		return nil
+	if err := c.claimBuildSlot(ctx, store, queueRow); err != nil {
+		return err
 	}
 
 	if err := c.persistStateLog(ctx, store, request, entity.RequestOutcomeReasonUnknown); err != nil {
@@ -318,12 +285,19 @@ func (c *Controller) persistStateLog(ctx context.Context, store storage.Storage,
 }
 
 // deriveBuildStrategy chooses the validation scope and baseline from the queue's last-known-good commit.
-// The caller resolves source control once and persists the returned values only after successfully claiming a build slot.
-func (c *Controller) deriveBuildStrategy(ctx context.Context, sc sourcecontrol.SourceControl, queueRow entity.Queue, request entity.Request) (strategy entity.BuildStrategy, baseURI string, err error) {
+func (c *Controller) deriveBuildStrategy(ctx context.Context, queueRow entity.Queue, request entity.Request) (strategy entity.BuildStrategy, baseURI string, err error) {
 	if queueRow.LastGreenURI == "" {
 		return entity.BuildStrategyFull, "", nil
 	}
 
+	sc, err := c.sourceControl.For(sourcecontrol.Config{QueueName: request.Queue})
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, _opName, "source_control_errors", 1,
+			metrics.TagsFromContext(ctx, metrics.NewTag("stage", "resolve"))...,
+		)
+		return entity.BuildStrategyUnknown, "", fmt.Errorf("failed to resolve source control for queue %s: %w", request.Queue, err)
+	}
+
 	isAncestor, err := sc.IsAncestor(ctx, queueRow.LastGreenURI, request.URI)
 	if err != nil {
 		if sourcecontrol.IsNotFound(err) {
@@ -349,35 +323,32 @@ func (c *Controller) deriveBuildStrategy(ctx context.Context, sc sourcecontrol.S
 	return entity.BuildStrategyFull, "", nil
 }
 
-// claimBuildSlot CAS-increments queue.in_flight_count by one. On version mismatch it
-// reloads queueRow and returns ErrVersionMismatch so the caller can retry.
-func (c *Controller) claimBuildSlot(ctx context.Context, store storage.Storage, queueRow *entity.Queue) error {
+// claimBuildSlot CAS-increments queue.in_flight_count, retrying version conflicts without re-gating
+// because the admit is already durable. A hard failure leaves the count low, never high.
+func (c *Controller) claimBuildSlot(ctx context.Context, store storage.Storage, queueRow entity.Queue) error {
 	queueStore := store.GetQueueStore()
 
-	updated := *queueRow
-	updated.InFlightCount = queueRow.InFlightCount + 1
-	newVersion := queueRow.Version + 1
-	if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil {
-		if errors.Is(err, storage.ErrVersionMismatch) {
-			got, getErr := queueStore.Get(ctx, queueRow.Name)
-			if getErr != nil {
-				return fmt.Errorf("failed to reload queue %s after version mismatch: %w", queueRow.Name, getErr)
-			}
-			*queueRow = got
-			return storage.ErrVersionMismatch
+	for {
+		updated := queueRow
+		updated.InFlightCount = queueRow.InFlightCount + 1
+		newVersion := queueRow.Version + 1
+		err := queueStore.Update(ctx, updated, queueRow.Version, newVersion)
+		if err == nil {
+			return nil
+		}
+		if !errors.Is(err, storage.ErrVersionMismatch) {
+			return fmt.Errorf("failed to claim build slot for queue %s: %w", queueRow.Name, err)
 		}
-		return fmt.Errorf("failed to claim build slot for queue %s: %w", queueRow.Name, err)
+		reloaded, getErr := queueStore.Get(ctx, queueRow.Name)
+		if getErr != nil {
+			return fmt.Errorf("failed to reload queue %s after version mismatch: %w", queueRow.Name, getErr)
+		}
+		queueRow = reloaded
 	}
-	updated.Version = newVersion
-	*queueRow = updated
-	return nil
 }
 
-// markProcessing CAS-marks request accepted→processing and persists the strategy chosen after the
-// build slot was claimed. It reapplies the values after a request reload so an accepted concurrent
-// update cannot discard them. transitioned is true only when this call performed the CAS; false
-// means a concurrent writer already advanced the request past accepted, so the caller must release
-// its claimed slot.
+// markProcessing CAS-marks request accepted→processing with strategy and baseURI, reapplied after a reload.
+// transitioned is false when another writer already moved the request past accepted; the caller then counts no slot.
 func (c *Controller) markProcessing(ctx context.Context, store storage.Storage, request *entity.Request, strategy entity.BuildStrategy, baseURI string) (transitioned bool, err error) {
 	reqStore := store.GetRequestStore()
 
@@ -408,43 +379,6 @@ func (c *Controller) markProcessing(ctx context.Context, store storage.Storage,
 	}
 }
 
-// releaseBuildSlot CAS-decrements queue.in_flight_count to compensate a slot claimed but never
-// admitted. It decrements relatively (preserving a concurrent record decrement) and retries on
-// version conflicts. Best-effort: it only logs on a hard failure, since the caller is unwinding.
-func (c *Controller) releaseBuildSlot(ctx context.Context, store storage.Storage, queueName string) {
-	queueStore := store.GetQueueStore()
-
-	for {
-		queueRow, err := queueStore.Get(ctx, queueName)
-		if err != nil {
-			c.logger.Errorw("failed to release claimed build slot",
-				"queue", queueName,
-				"error", err,
-			)
-			return
-		}
-		if queueRow.InFlightCount <= 0 {
-			return
-		}
-
-		updated := queueRow
-		updated.InFlightCount = queueRow.InFlightCount - 1
-		newVersion := queueRow.Version + 1
-		if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil {
-			if errors.Is(err, storage.ErrVersionMismatch) {
-				continue
-			}
-			c.logger.Errorw("failed to release claimed build slot",
-				"queue", queueName,
-				"error", err,
-			)
-			return
-		}
-		metrics.NamedCounter(c.metricsScope, _opName, "slot_released", 1, metrics.TagsFromContext(ctx)...)
-		return
-	}
-}
-
 // supersedeRequest returns the canonical row so the log uses the version that won any CAS race.
 func (c *Controller) supersedeRequest(ctx context.Context, store storage.Storage, request entity.Request) (entity.Request, error) {
 	reqStore := store.GetRequestStore()