From 5718039d8b11d168f55de0e8b06b8822f159c53d Mon Sep 17 00:00:00 2001 From: sergeyb Date: Tue, 27 Oct 2026 23:28:00 +0000 Subject: [PATCH] fix(stovepipe): key build-slot ownership by request ID Track build slots in in_flight_request_ids, the set of requests holding a slot, instead of the queue's anonymous in_flight_count. The set lives in a new column next to in_flight_count, which is retained unused for schema compatibility. A redelivery that finds its own claim resumes the admit instead of holding behind it, or, when a newer head arrived, releases the slot before being superseded, so a crash between the claim and the processing write no longer wedges the queue. Release removes by ID and is a no-op when absent, so a retried outcome write cannot free another request's slot, and the DLQ releases for any nonterminal request. Co-Authored-By: Claude Fable 5.1 --- doc/rfc/stovepipe/steps/build.md | 4 +- doc/rfc/stovepipe/steps/buildsignal.md | 10 +- doc/rfc/stovepipe/steps/process.md | 52 +++--- doc/rfc/stovepipe/steps/record.md | 4 +- .../controller/build/build_analysis_test.go | 4 +- .../controller/buildsignal/buildsignal.go | 42 ++--- .../buildsignal/buildsignal_test.go | 59 ++++--- stovepipe/controller/dlq/build_test.go | 4 +- stovepipe/controller/dlq/buildsignal.go | 6 +- stovepipe/controller/dlq/buildsignal_test.go | 4 +- stovepipe/controller/dlq/dlq.go | 72 +++----- stovepipe/controller/dlq/dlq_test.go | 28 ++-- stovepipe/controller/ingest_test.go | 4 +- stovepipe/controller/process/process.go | 87 +++++----- stovepipe/controller/process/process_test.go | 155 ++++++++++++------ stovepipe/core/queuestate/queuestate_test.go | 8 +- stovepipe/entity/queue.go | 5 +- stovepipe/entity/queue_config.go | 2 +- stovepipe/extension/storage/README.md | 2 + .../extension/storage/mysql/queue_store.go | 46 +++++- .../storage/mysql/queue_store_test.go | 29 ++-- .../extension/storage/mysql/schema/queue.sql | 5 +- test/e2e/stovepipe/harness_test.go | 11 +- test/e2e/stovepipe/ingest_test.go | 2 +- test/e2e/stovepipe/issuetrack_test.go | 4 +- .../stovepipe/extension/storage/suite.go | 19 ++- 26 files changed, 373 insertions(+), 295 deletions(-) diff --git a/doc/rfc/stovepipe/steps/build.md b/doc/rfc/stovepipe/steps/build.md index 7fb415d74..655c3b45c 100644 --- a/doc/rfc/stovepipe/steps/build.md +++ b/doc/rfc/stovepipe/steps/build.md @@ -90,7 +90,7 @@ Every branch is safe under at-least-once redelivery — with SubmitQueue's postu ## Fail-closed interaction -A build that never reaches step 8 — `Trigger` failing repeatedly, the publish to `buildsignal` never landing, `BuildStore.Create` down — must not wedge its `Request`'s Queue slot forever: `process`'s per-Queue concurrency gate holds `in_flight_count` open until the Request reaches a terminal state (see [process.md](process.md#concurrency-lifecycle)). `build` does not implement the forcing function itself. Per [workflow.md](../workflow.md#fail-closed-on-unprocessable-work), every non-retryable failure in the algorithm rejects to DLQ (see [Error classification](#error-classification)), and a Request stuck past `MaxAttempts` is driven to a conservative terminal `failed` by the DLQ reconciler, which decrements `in_flight_count` and frees the slot. This is the same posture `buildsignal` relies on for its own poll loop (see [buildsignal.md](buildsignal.md#fail-closed-interaction)) — `build` and `buildsignal` are two links in the same fail-closed chain that keeps one bad Request from wedging its Queue. +A build that never reaches step 8 — `Trigger` failing repeatedly, the publish to `buildsignal` never landing, `BuildStore.Create` down — must not wedge its `Request`'s Queue slot forever: `process`'s per-Queue concurrency gate keeps the Request's ID in `in_flight_request_ids` until the Request reaches a terminal state (see [process.md](process.md#concurrency-lifecycle)). `build` does not implement the forcing function itself. Per [workflow.md](../workflow.md#fail-closed-on-unprocessable-work), every non-retryable failure in the algorithm rejects to DLQ (see [Error classification](#error-classification)), and a Request stuck past `MaxAttempts` is driven to a conservative terminal `failed` by the DLQ reconciler, which removes the Request's ID and frees the slot. This is the same posture `buildsignal` relies on for its own poll loop (see [buildsignal.md](buildsignal.md#fail-closed-interaction)) — `build` and `buildsignal` are two links in the same fail-closed chain that keeps one bad Request from wedging its Queue. `service/stovepipe/server` registers `dlq.NewDLQBuildController` on `build_dlq`. The reconciler loads the request named by the dead-lettered `BuildRequest`, writes a non-terminal request to `failed`, and releases the Queue slot when that request is still `processing`. A request that is already terminal is left alone, except a request already `failed`, whose missing failure log is repaired. A poll `build` already published keeps running on its own delivery: `buildsignal` still releases the slot when that build goes terminal, and this reconciler does not overwrite that outcome. @@ -98,7 +98,7 @@ One boundary is worth stating explicitly: this path fires only when `build` (or ## Cancellation: defined, not yet called -`Cancel` is in the `BuildRunner` contract for parity with SubmitQueue, but no controller in this design calls it. SubmitQueue cancels from its `speculate`/cancel path when a batch is preempted mid-validation; stovepipe's `process` explicitly does **not** preempt an admitted `Request` ("a newer head does not preempt an in-flight validation" — [process.md](process.md#concurrency-lifecycle)). So there is currently no trigger for cancelling a running build. `Cancel` stays in the contract because a future change — the lease-based self-healing `in_flight_count`, or raising `max_concurrent` past 1, both floated in `process.md` — could need to abandon a build no longer worth finishing. Until then it is unused surface, not dead weight. +`Cancel` is in the `BuildRunner` contract for parity with SubmitQueue, but no controller in this design calls it. SubmitQueue cancels from its `speculate`/cancel path when a batch is preempted mid-validation; stovepipe's `process` explicitly does **not** preempt an admitted `Request` ("a newer head does not preempt an in-flight validation" — [process.md](process.md#concurrency-lifecycle)). So there is currently no trigger for cancelling a running build. `Cancel` stays in the contract because a future change — lease-based self-healing build slots, or raising `max_concurrent` past 1, both floated in `process.md` — could need to abandon a build no longer worth finishing. Until then it is unused surface, not dead weight. ## Error classification diff --git a/doc/rfc/stovepipe/steps/buildsignal.md b/doc/rfc/stovepipe/steps/buildsignal.md index cf0b7f78b..71c82279e 100644 --- a/doc/rfc/stovepipe/steps/buildsignal.md +++ b/doc/rfc/stovepipe/steps/buildsignal.md @@ -62,7 +62,7 @@ For a delivery carrying build id `B`: 7. If the stored status is terminal: a. If R does not already carry an outcome, release the queue's build slot: - CAS-decrement Queue.in_flight_count, clamped at zero. + CAS-remove R.ID from Queue.in_flight_request_ids; no write if it is absent. - failure here aborts the step: R must not go terminal while still holding a slot. b. If R does not already carry an outcome, CAS R from processing to the outcome the stored status projects onto it and persist B as R's terminal build id: succeeded -> succeeded, failed -> failed, cancelled -> cancelled. First writer wins. @@ -85,7 +85,7 @@ For a delivery carrying build id `B`: exactly one poll chain; no message ids are minted and no new rows are written per tick. ``` -**Why the slot is released before the outcome write, and why a failed release aborts it**: `Queue` and `Request` are separate entities with no cross-entity transaction, so the ordering picks which crash failure mode we accept. Both rules serve one invariant — *the request must not go terminal while still holding a slot* — because a terminal request is skipped by redelivery and by the DLQ reconciler alike, so nothing would ever decrement it. Failing this way leaves the request non-terminal: redelivery re-runs both steps and decrements again, transiently over-admitting by one slot until the zero clamp reconverges. Over-admission is the failure mode this pipeline already prefers, for the same reason and in the same words as the DLQ reconciler (see [process.md](process.md#in_flight_count-integrity)). +**Why the slot is released before the outcome write, and why a failed release aborts it**: `Queue` and `Request` are separate entities with no cross-entity transaction, so the ordering picks which crash failure mode we accept. Both rules serve one invariant — *the request must not go terminal while still holding a slot* — because a terminal request is skipped by redelivery and by the DLQ reconciler alike, so nothing would ever release it. Failing this way leaves the request non-terminal: redelivery re-runs both steps, and the release finds R's ID already gone and writes nothing, so a retry cannot free another Request's slot (see [process.md](process.md#in_flight_request_ids-integrity)). **Baseline timing.** After the winning Request outcome is durable, a successful Request advances `last_green_uri` and `last_green_request_id` in a CAS retry loop before publishing to `record`. Older candidates never regress the bookmark. A terminal redelivery retries the advance without releasing capacity again. Slot release and bookmark advancement are separate updates, so an admission between them can still select the previous baseline. Project resolution, facts, and Git promotion remain in `record`. @@ -136,7 +136,7 @@ Everything else — factory lookup, an `Update` store error other than a CAS con ### What it costs when a backend does not classify `Status` errors -Leaving `Status` to the classifier only works if the backend classifies. A `BuildRunner` whose transport returns plain `fmt.Errorf` values gets the non-retryable default, and here that default is expensive: dead-lettering ends the *only* poll chain for a build that is still running, and the request keeps holding one of the queue's `in_flight_count` build slots until reconciliation gives it back. A single `502` from a proxy in front of the build API then looks exactly like "this build can never be polled". +Leaving `Status` to the classifier only works if the backend classifies. A `BuildRunner` whose transport returns plain `fmt.Errorf` values gets the non-retryable default, and here that default is expensive: dead-lettering ends the *only* poll chain for a build that is still running, and the request keeps holding one of the queue's build slots until reconciliation gives it back. A single `502` from a proxy in front of the build API then looks exactly like "this build can never be polled". Two things keep a blip from stalling a queue, and a backend needs both: @@ -163,9 +163,9 @@ The window to guard is between persisting status (step 6) and ack (steps 7–8); ## Fail-closed interaction -A build that never reaches terminal `Status` — runner outage, a build the runner lost — must not wedge its `Request` forever, since callers gate deployments on greenness reaching a recorded terminal state. `buildsignal` does not implement the forcing function: per [workflow.md](../workflow.md#fail-closed-on-unprocessable-work) and the `in_flight_count` slot lifecycle in [process.md](process.md#concurrency-lifecycle), a `Request` stuck at `buildsignal` past `MaxAttempts` dead-letters, and the DLQ reconciler forces a conservative terminal `failed` and releases the Queue's slot. This is the same posture SubmitQueue's build/buildsignal pair relies on: terminal status is what releases the slot and lets validation progress. +A build that never reaches terminal `Status` — runner outage, a build the runner lost — must not wedge its `Request` forever, since callers gate deployments on greenness reaching a recorded terminal state. `buildsignal` does not implement the forcing function: per [workflow.md](../workflow.md#fail-closed-on-unprocessable-work) and the build-slot lifecycle in [process.md](process.md#concurrency-lifecycle), a `Request` stuck at `buildsignal` past `MaxAttempts` dead-letters, and the DLQ reconciler forces a conservative terminal `failed` and releases the Queue's slot. This is the same posture SubmitQueue's build/buildsignal pair relies on: terminal status is what releases the slot and lets validation progress. -One boundary of that posture is worth stating: the `MaxAttempts` path fires only when polls *fail*. A runner that keeps answering a healthy non-terminal status forever — a hung build on a backend with no timeout of its own — never errors, so the hold chain (whose redeliveries deliberately do not count toward the retry limit) re-polls indefinitely and nothing dead-letters; SubmitQueue's poll loop shares this property. Bounding it requires a poll deadline — a `max_validation_ms` past which `buildsignal` treats the build as failed and lets the normal terminal path run — which pairs naturally with the lease idea [process.md](process.md#per-queue-concurrency-gate) floats for `in_flight_count`. Deferred with it; until then a too-old non-terminal `Build` is an operational alert, not a self-healing path. +One boundary of that posture is worth stating: the `MaxAttempts` path fires only when polls *fail*. A runner that keeps answering a healthy non-terminal status forever — a hung build on a backend with no timeout of its own — never errors, so the hold chain (whose redeliveries deliberately do not count toward the retry limit) re-polls indefinitely and nothing dead-letters; SubmitQueue's poll loop shares this property. Bounding it requires a poll deadline — a `max_validation_ms` past which `buildsignal` treats the build as failed and lets the normal terminal path run — which pairs naturally with the lease idea [process.md](process.md#per-queue-concurrency-gate) floats for `in_flight_request_ids`. Deferred with it; until then a too-old non-terminal `Build` is an operational alert, not a self-healing path. ## Entity, storage, and queue additions diff --git a/doc/rfc/stovepipe/steps/process.md b/doc/rfc/stovepipe/steps/process.md index b1089fbe0..c1ca6c6f4 100644 --- a/doc/rfc/stovepipe/steps/process.md +++ b/doc/rfc/stovepipe/steps/process.md @@ -8,7 +8,7 @@ It handles everything *before* the first build is triggered: it does not run bui Ingest publishes a `ProcessRequest` (request id only — producer and consumer share the store) to the process topic, **partitioned by Queue name**. Two consequences the design relies on: -1. **One partition per Queue**, and `BatchSize = 1` (strict serialization; see [sql-queue-rfc.md](../../sql-queue-rfc.md#strict-serialization-opt-in)) — so at most one `process` invocation runs per Queue at a time. The read-modify-write on the Queue row (gate check, count increment) is therefore race-free **without** a transaction, the same property SubmitQueue's `validate` gets from per-queue partition leasing. +1. **One partition per Queue**, and `BatchSize = 1` (strict serialization; see [sql-queue-rfc.md](../../sql-queue-rfc.md#strict-serialization-opt-in)) — so at most one `process` invocation runs per Queue at a time. The read-modify-write on the Queue row (gate check, slot claim) is therefore race-free **without** a transaction, the same property SubmitQueue's `validate` gets from per-queue partition leasing. 2. **`process` is not the only Queue-row writer.** `ingest` stamps `latest_request_id` (the request id of the newest accepted head, not a URI; see [Backlog coalescing](#backlog-coalescing)) and `buildsignal` releases slots when a build goes terminal, then advances the `last_green_uri` bookmark for a successful Request. They touch different fields under optimistic-locking CAS, so concurrent writes converge instead of clobbering. ## Process algorithm @@ -24,19 +24,19 @@ For a delivery carrying request id `R`: - re-announce validation start and re-publish R to build (either prior publish may have failed), ack, return. 4. R.State is accepted. Load the Queue row Q. 5. Coalesce: if CompareRequestID(R.Queue, R.ID, Q.latest_request_id) < 0: - - a newer head exists -> mark R superseded, ack, return. (No slot consumed.) -6. R is the latest head. Gate: if Q.in_flight_count >= max_concurrent (from queue config; see below): + - a newer head exists -> if R holds a slot, release it first; mark R superseded, ack, return. (No slot consumed.) +6. R is the latest head. Gate: if R.ID is not in Q.in_flight_request_ids and len(Q.in_flight_request_ids) >= max_concurrent (from queue config; see below): - 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. + b. CAS the Queue row: add R.ID to in_flight_request_ids (no write if already present). c. CAS the Request: accepted -> processing, persist build_strategy + base_uri. d. Announce validation start on the hook topic (see "Hooks"). e. Publish R to build. f. ack. ``` -Step 5 runs regardless of the gate: an intermediate head is superseded on sight (even mid-validation), because superseding consumes no slot. +Step 5 runs regardless of the gate: an intermediate head is superseded on sight (even mid-validation), because superseding consumes no slot. A Request superseded while holding a slot from an interrupted admit releases it first, so no slot is spent on a stale head; a failed release fails the delivery for retry instead of stranding the slot. ## Build-strategy decision @@ -60,14 +60,14 @@ Validation is expensive and shares a baseline, so heads arriving while an earlie | Source | Field | Meaning | |---|---|---| | Queue row | `last_green_uri` | Bookmark `buildsignal` advances on a successful build; empty until first green. | -| Queue row | `in_flight_count` | Requests past `process` and not yet terminal. `process` increments on admit; `record` (or DLQ reconciliation) decrements on terminal. | +| Queue row | `in_flight_request_ids` | IDs of Requests holding a build slot; its length is the in-flight count. `process` adds on admit; `buildsignal` (or DLQ reconciliation) removes on terminal. | | Queue config | `max_concurrent` | Cap on concurrent in-flight validations. **Default 1** (global wiring default for MVP; per-queue override when a Stovepipe `queueconfig` extension lands). | -A slot is held from admit until the build goes terminal (`process → build → buildsignal`), not just while `process` runs. It is released when the Request reaches **any** terminal state and `in_flight_count` is decremented — `buildsignal` recording the build's outcome, success *or* failure, or the DLQ reconciler forcing a terminal `failed` (see [integrity](#in_flight_count-integrity)). A build *failure* frees the slot just like a success; only a Request that never terminates keeps its slot. +A slot is held from admit until the build goes terminal (`process → build → buildsignal`), not just while `process` runs. It is released when the Request reaches **any** terminal state and its ID leaves `in_flight_request_ids` — `buildsignal` recording the build's outcome, success *or* failure, or the DLQ reconciler forcing a terminal `failed` (see [integrity](#in_flight_request_ids-integrity)). A build *failure* frees the slot just like a success; only a Request that never terminates keeps its slot. **Liveness — a stuck Request wedges the whole Queue.** The slot is shared per-Queue, so a Request admitted but never driven terminal holds the only slot and stalls the whole Queue. The fail-closed path in [workflow.md](../workflow.md#fail-closed-on-unprocessable-work) prevents this: every admitted Request terminates — `build`/`buildsignal` errors retry to `MaxAttempts`, then dead-letter, and the DLQ reconciler forces a conservative terminal `failed`, freeing the slot. Residual risk is operational: a **poison DLQ message** the reconciler can't process loops forever (it runs always-retryable), wedging that Queue until an operator removes it. The gate makes the blast radius the whole Queue, not one Request — monitor and alert on the DLQ. -**Future alternative — time-bounded leases.** To self-heal instead of relying on operator cleanup, `in_flight_count` could become a set of `{owner, expires_at}` leases: the gate counts only unexpired leases, terminal transitions drop the owner's lease, and a leaked lease is reclaimed on expiry — bounding any stall to a `max_validation_ms`. A *list* of leases would also generalize to `max_concurrent > 1`. Deferred; `in_flight_count` plus the fail-closed path is enough for MVP. +**Future alternative — time-bounded leases.** To self-heal instead of relying on operator cleanup, each owner in `in_flight_request_ids` could carry an `expires_at`: the gate counts only unexpired owners, and a leaked owner is reclaimed on expiry — bounding any stall to a `max_validation_ms`. Deferred; the owner set plus the fail-closed path is enough for MVP. `max_concurrent` is a scalar, and `= 1` is an MVP simplicity choice, **not** a correctness requirement — see [Raising `max_concurrent`](#raising-max_concurrent-speculative-validation). @@ -115,7 +115,7 @@ The gate is **not** tied to `process` returning; a slot taken at admit is held u **Rules** -1. **One slot per in-flight validation** (MVP: one per Queue). `process` increments `in_flight_count` on admit; `record` decrements on terminal. +1. **One slot per in-flight validation** (MVP: one per Queue). `process` adds the Request's ID to `in_flight_request_ids` on admit; the terminal transition removes it. 2. **No skip-ahead while in-flight.** The latest head waits for a slot until the running validation completes; it never preempts. 3. **Intermediates are superseded on sight**, gate open or closed — no slot consumed (step 5). 4. **Coalesce-to-latest on gate open.** When a slot frees, the waiting latest head is admitted. @@ -123,10 +123,10 @@ The gate is **not** tied to `process` returning; a slot taken at admit is held u **Worked example** — Queue `monorepo/main`, `max_concurrent = 1`, poller reporting heads A→F: -1. **A** admitted (`in_flight_count = 1`), published to `build`. +1. **A** admitted (`in_flight_request_ids = [A]`), published to `build`. 2. While A runs, **B**, **C**, **D** are ingested (`latest_request_id = D.id`). 3. B: older than D → **superseded** (acked), though A is still in flight. Same for **C**. D is latest but the gate is closed → **waits for slot** (held). -4. A's build finishes → `record` records A's greenness, `in_flight_count → 0`. +4. A's build finishes → `record` records A's greenness, `in_flight_request_ids → []`. 5. D's re-check → gate open, D still latest → **D admitted**, published to `build`. 6. While D runs, **E**, **F** ingested (`latest_request_id = F.id`). E superseded on sight; F waits for slot. 7. D completes → slot frees → **F admitted**. @@ -159,28 +159,28 @@ 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. On a crash after claiming the slot but before persisting `processing`, redelivery re-reads `accepted`, finds R's ID in `in_flight_request_ids`, and either resumes the admit without claiming again or, behind a newer head, releases the slot and is superseded (see integrity below). - **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). +- **deferred (waiting for slot)** → no state or slot 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 "slot claimed, state not yet `processing`". Admit does the claim and the state transition as two ordered CAS writes; because the slot is keyed by request ID, the redelivery recognizes its own claim (see integrity below). -## `in_flight_count` integrity +## `in_flight_request_ids` 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: +Each build slot is owned by a Request ID, so both writes are idempotent per Request: -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. **Claim is idempotent.** `process` adds R's ID only if it is absent. A crash between the claim and the `accepted → processing` CAS — or a claim that committed but reported an error — is repaired by the redelivery: if R is still the latest head it finds its own ID, passes the gate, and resumes the admit; if a newer head arrived it releases the slot and is superseded. No sweeper is needed. +2. **Release is idempotent.** `buildsignal` and the DLQ reconciler remove R's ID before the terminal write, and removing an absent ID writes nothing. A retry after a failed terminal write therefore cannot free another Request's slot, and the DLQ reconciler can release for any non-terminal Request. -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. +An owner that is never driven terminal still holds its slot; the fail-closed DLQ path is what bounds that (see [Liveness](#per-queue-concurrency-gate)). ## Edge cases - **Re-ingest of a superseded URI.** Ingest dedups on `(Queue, URI)` and returns the existing (now terminal `superseded`) id; `process` acks it as a no-op (step 2). Correct: a URI is only superseded for a *strictly newer* head, so re-validating it is never wanted. - **Gate closed, no newer head.** The single latest head waits for a slot until the in-flight validation completes — the steady state, not an error. - **Head equals last-green.** `IsAncestor(lastGreen, R.URI)` with `R.URI == lastGreen` is degenerate; treat as already-green, or (simpler) run an incremental build with an empty delta. Left to `build`. -- **Queue row missing.** First head for a Queue: ingest get-or-creates the row with defaults (`in_flight_count = 0`, empty `last_green_uri`). `process` treats a missing row as non-retryable — storage's read-after-write guarantee means ingest's write is already visible by the time `process` reads it, so a miss is a storage defect, not lag. +- **Queue row missing.** First head for a Queue: ingest get-or-creates the row with defaults (empty `in_flight_request_ids` and `last_green_uri`). `process` treats a missing row as non-retryable — storage's read-after-write guarantee means ingest's write is already visible by the time `process` reads it, so a miss is a storage defect, not lag. ## Entity model @@ -192,7 +192,7 @@ Runtime coordination only — fields the pipeline writes under CAS: |---|---|---| | `name` | Stable logical id (`monorepo/main`); the string ingest accepts | ingest (create) | | `last_green_uri` | Bookmark; empty until first green | buildsignal | -| `in_flight_count` | Active Phase 1 validations | process (+1), record/DLQ (−1) | +| `in_flight_request_ids` | Requests holding a build slot | process (add), buildsignal/DLQ (remove) | | `latest_request_id` | Request id of the newest head ingest accepted | ingest | | `version` | Optimistic-locking version | all writers | @@ -222,19 +222,19 @@ Transitions use the repo's optimistic-locking pattern: compute `newVersion = old New key/value-shaped operations (single-key reads/writes, no server-side filtering or aggregation): -- **`QueueStore`** (new): `Create(ctx, queue)`, `Get(ctx, name)`, and `Update(ctx, queue, oldVersion, newVersion)` (CAS). Callers orchestrate get-or-create; ingest CASes `latest_request_id`; `process` CASes `in_flight_count`; `record` CASes `last_green_uri` + `in_flight_count`. +- **`QueueStore`** (new): `Create(ctx, queue)`, `Get(ctx, name)`, and `Update(ctx, queue, oldVersion, newVersion)` (CAS). Callers orchestrate get-or-create; ingest CASes `latest_request_id`; `process` CASes `in_flight_request_ids`; `buildsignal` CASes `in_flight_request_ids` and `last_green_uri`. - **`RequestStore`**: no new methods — the added `Request` fields ride the existing `Create`/`Update` CAS. No "list requests by queue/state" query is introduced; coalescing uses the single-row `latest_request_id` pointer instead, keeping the contract satisfiable by a plain KV backend. ## Waiting for a slot -When the gate is closed, `process` must defer the latest head without admitting it (no `in_flight_count` increment, no publish to `build`). The mechanism is the consumer hold primitive ([consumer-hold.md](../../consumer-hold.md)): the controller records a hold for `gate_wait_delay_ms` and returns success, and the framework postpones the delivery — the same message redelivers after the delay, and the redelivery does not count toward `MaxAttempts`. +When the gate is closed, `process` must defer the latest head without admitting it (no slot claim, no publish to `build`). The mechanism is the consumer hold primitive ([consumer-hold.md](../../consumer-hold.md)): the controller records a hold for `gate_wait_delay_ms` and returns success, and the framework postpones the delivery — the same message redelivers after the delay, and the redelivery does not count toward `MaxAttempts`. Every wake-up re-runs the same **coalesce-then-gate** checks (steps 5 → 6): 1. **Stale? (checked first.)** If `CompareRequestID(R.Queue, R.ID, Q.latest_request_id) < 0`, `R` is no longer latest → supersede it (ack). A newer head is admitted by its own delivery when its slot attempt runs. -2. **Slot free?** If `in_flight_count < max_concurrent` (from config) and `R` is still latest → admit (step 7). +2. **Slot free?** If `len(in_flight_request_ids) < max_concurrent` (from config) and `R` is still latest → admit (step 7). Nothing is admitted to `build` until the gate opens. @@ -242,12 +242,12 @@ Nothing is admitted to `build` until the gate opens. **Walkthrough** — Queue `monorepo/main`, `max_concurrent = 1`, heads A→F: -1. **A** admitted, published to `build` (`in_flight_count = 1`). `process` returns (acks); A continues through `build → buildsignal → record`. +1. **A** admitted, published to `build` (`in_flight_request_ids = [A]`). `process` returns (acks); A continues through `build → buildsignal → record`. 2. **B**, **C** ingested. Their deliveries run behind A's in-flight validation (not behind a gate wait yet) → superseded on sight (step 5), acked. 3. **D** ingested (`latest_request_id = D.id`). D's delivery: latest, gate closed → **hold** (postponed for `gate_wait_delay_ms`; the partition waits behind D). 4. While D waits, **E**, **F** ingested (`latest_request_id = F.id`). Their process messages sit behind D's postponed row — they do not run yet. 5. D's hold expires → D redelivers first, re-reads the Queue row → D is older than F → **supersede D**, ack (partition drains). **E**'s delivery runs → superseded. **F**'s delivery runs → latest, gate still closed → **hold**. -6. A completes at `buildsignal` → `in_flight_count → 0`. F's hold expires → gate open, still latest → **admit F**, publish to `build`, ack. +6. A completes at `buildsignal` → `in_flight_request_ids → []`. F's hold expires → gate open, still latest → **admit F**, publish to `build`, ack. **Properties.** This section previously weighed two options — park-and-extend-visibility versus ack-and-`PublishAfter` — and deferred the choice to a future consumer primitive. Hold is that primitive, and it dominates both: diff --git a/doc/rfc/stovepipe/steps/record.md b/doc/rfc/stovepipe/steps/record.md index e420f1ae0..d85fa7b7a 100644 --- a/doc/rfc/stovepipe/steps/record.md +++ b/doc/rfc/stovepipe/steps/record.md @@ -127,7 +127,7 @@ Whole-repository degree depends on the outcome `buildsignal` stamped on the Requ ## Last-green advancement (Queue bookmark) -`buildsignal` advances the bookmark after releasing the build slot and persisting the successful Request outcome, before publishing to `record`. The bookmark only moves forward. `entity.Queue` carries `LastGreenURI` alongside `LastGreenRequestID`, the request id that owns the current URI, plus `InFlightCount` and `LatestRequestID`. The guard compares ids with `entity.CompareRequestID(R.Queue, …)`, the same ingest-order comparison `ingest` and `process` use for coalescing. It errors on an id that does not match the queue's format, which is non-retryable: re-parsing the same ids cannot start succeeding. +`buildsignal` advances the bookmark after releasing the build slot and persisting the successful Request outcome, before publishing to `record`. The bookmark only moves forward. `entity.Queue` carries `LastGreenURI` alongside `LastGreenRequestID`, the request id that owns the current URI, plus `InFlightRequestIDs` and `LatestRequestID`. The guard compares ids with `entity.CompareRequestID(R.Queue, …)`, the same ingest-order comparison `ingest` and `process` use for coalescing. It errors on an id that does not match the queue's format, which is non-retryable: re-parsing the same ids cannot start succeeding. The write goes through `QueueStore.Update(ctx, queue, oldVersion, newVersion)`, so version arithmetic stays in the controller: it computes `newVersion = oldVersion + 1` and the store does a pure conditional write. `ErrVersionMismatch` is absorbed by the loop rather than returned, since a concurrent writer moved the row and reloading converges. @@ -162,7 +162,7 @@ Both are best-effort. Every way the lookup can fail is counted, logged, and swal ## Build slot release -`Queue.in_flight_count` is released by `buildsignal` before it stamps the outcome and publishes to the record step. The DLQ reconciler releases the slot on the fail-closed path for the same reason. `record`'s only Queue write is the bookmark. +The Request's slot in `Queue.in_flight_request_ids` is released by `buildsignal` before it stamps the outcome and publishes to the record step. The DLQ reconciler releases the slot on the fail-closed path for the same reason. `record`'s only Queue write is the bookmark. ## Hooks diff --git a/stovepipe/controller/build/build_analysis_test.go b/stovepipe/controller/build/build_analysis_test.go index 0609168d0..5dcf9fe7b 100644 --- a/stovepipe/controller/build/build_analysis_test.go +++ b/stovepipe/controller/build/build_analysis_test.go @@ -43,7 +43,7 @@ func TestBuildAnalysisPipeline(t *testing.T) { var savedBuild entity.Build request := processingRequest(entity.BuildStrategyFull, "") - queue := entity.Queue{Name: testQueue, InFlightCount: 1, Version: 1} + queue := entity.Queue{Name: testQueue, InFlightRequestIDs: []string{testID}, Version: 1} queueStore := storagemock.NewMockQueueStore(ctrl) issueStore := storagemock.NewMockBuildIssueStore(ctrl) @@ -150,7 +150,7 @@ func TestBuildAnalysisPipeline(t *testing.T) { require.Len(t, published["buildanalysis"], 2) assert.Equal(t, entity.BuildStatusFailed, savedBuild.Status) assert.Equal(t, entity.RequestStateFailed, request.State) - assert.Zero(t, queue.InFlightCount) + assert.Empty(t, queue.InFlightRequestIDs) assert.True(t, persisted[b]) assert.NotEqual(t, published["buildanalysis"][0].ID, published["buildanalysis"][1].ID) diff --git a/stovepipe/controller/buildsignal/buildsignal.go b/stovepipe/controller/buildsignal/buildsignal.go index c6d3e0f30..54e0d33a3 100644 --- a/stovepipe/controller/buildsignal/buildsignal.go +++ b/stovepipe/controller/buildsignal/buildsignal.go @@ -25,6 +25,7 @@ import ( "context" "errors" "fmt" + "slices" "github.com/uber-go/tally" entityqueue "github.com/uber/submitqueue/platform/base/messagequeue" @@ -150,8 +151,8 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) er // Anything else is terminal without a build outcome, which today means only // `superseded` — and process supersedes solely from `accepted` (see // supersedeRequest), so a superseded request can never reach this point. It - // is guarded rather than assumed away because finishRequest would otherwise - // release a build slot the request never claimed: superseding consumes none. + // is guarded rather than assumed away because the terminal path below has no + // build outcome to record for such a request. if request.State != entity.RequestStateProcessing && !request.State.HasBuildOutcome() { return nil } @@ -232,25 +233,14 @@ func (c *Controller) persistBuildFinishedLog(ctx context.Context, store storage. return nil } -// finishRequest releases the queue's build slot and projects the build's -// terminal status and ID onto the request, leaving it terminal. It is a no-op -// when the request already carries an outcome, so a redelivery neither -// double-releases the slot nor rewrites the state. -// -// The slot is released *before* the terminal write, and a failed release aborts -// the write, matching the DLQ reconciler (see stovepipe/controller/dlq/dlq.go). -// Both rules exist for the same reason: the request must not go terminal while -// still holding a slot, because a terminal request is skipped on redelivery and -// by the reconciler, so nothing would ever decrement it. Failing this way leaves -// the request non-terminal, so redelivery re-runs both steps and decrements again -// — transiently over-admitting by one until releaseBuildSlot's zero clamp -// reconverges, which is the failure mode this pipeline prefers. +// finishRequest releases the request's build slot before recording the outcome, since nothing +// releases a terminal request; slots are keyed by request ID, so a retry cannot free another's. func (c *Controller) finishRequest(ctx context.Context, store storage.Storage, request *entity.Request, buildID string, status entity.BuildStatus) error { if request.State.HasBuildOutcome() { return nil } - if err := c.releaseBuildSlot(ctx, store, request.Queue); err != nil { + if err := c.releaseBuildSlot(ctx, store, *request); err != nil { metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) return err } @@ -336,31 +326,29 @@ 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. -func (c *Controller) releaseBuildSlot(ctx context.Context, store storage.Storage, queueName string) error { +// releaseBuildSlot CAS-removes request's ID from the queue's in_flight_request_ids, writing nothing +// when it is absent. A hard failure is returned so the caller does not mark the request terminal. +func (c *Controller) releaseBuildSlot(ctx context.Context, store storage.Storage, request entity.Request) error { queueStore := store.GetQueueStore() for { - queueRow, err := queueStore.Get(ctx, queueName) + queueRow, err := queueStore.Get(ctx, request.Queue) if err != nil { - return fmt.Errorf("failed to load queue %s to release build slot: %w", queueName, err) + return fmt.Errorf("failed to load queue %s to release build slot: %w", request.Queue, err) } - if queueRow.InFlightCount <= 0 { + ownerIndex := slices.Index(queueRow.InFlightRequestIDs, request.ID) + if ownerIndex < 0 { return nil } updated := queueRow - updated.InFlightCount = queueRow.InFlightCount - 1 + updated.InFlightRequestIDs = slices.Delete(slices.Clone(queueRow.InFlightRequestIDs), ownerIndex, ownerIndex+1) newVersion := queueRow.Version + 1 if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil { if errors.Is(err, storage.ErrVersionMismatch) { continue } - return fmt.Errorf("failed to release build slot for queue %s: %w", queueName, err) + return fmt.Errorf("failed to release build slot for request %s in queue %s: %w", request.ID, request.Queue, err) } metrics.NamedCounter(c.metricsScope, _opName, "slot_released", 1, metrics.TagsFromContext(ctx)...) return nil diff --git a/stovepipe/controller/buildsignal/buildsignal_test.go b/stovepipe/controller/buildsignal/buildsignal_test.go index 5731831d7..cb07fee41 100644 --- a/stovepipe/controller/buildsignal/buildsignal_test.go +++ b/stovepipe/controller/buildsignal/buildsignal_test.go @@ -48,6 +48,7 @@ import ( const ( testQueue = "monorepo/main" testID = "7" + testOtherID = "8" testBuildID = "bk-1" ) @@ -175,19 +176,19 @@ func build(status entity.BuildStatus, version int32) entity.Build { } } -// queueRow returns the testQueue's row holding inFlight admitted validations. -func queueRow(inFlight, version int32) entity.Queue { +// queueRow returns the testQueue's row whose build slots are held by inFlightRequestIDs. +func queueRow(inFlightRequestIDs []string, version int32) entity.Queue { return entity.Queue{ - Name: testQueue, - InFlightCount: inFlight, - Version: version, + Name: testQueue, + InFlightRequestIDs: inFlightRequestIDs, + Version: version, } } func expectFinishWrites(m buildsignalMocks, state entity.RequestState) *gomock.Call { eventCall := expectBuildFinished(m) - m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow(1, 4), nil).After(eventCall) - m.queueStore.EXPECT().Update(gomock.Any(), queueRow(0, 4), int32(4), int32(5)).Return(nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow([]string{testID}, 4), nil).After(eventCall) + m.queueStore.EXPECT().Update(gomock.Any(), queueRow([]string{}, 4), int32(4), int32(5)).Return(nil) outcomeCall := m.reqStore.EXPECT().Update(gomock.Any(), requestWithState(state), int32(1), int32(2)).Return(nil) if state == entity.RequestStateSucceeded { return expectAdvanceLastGreen(m, outcomeCall) @@ -196,11 +197,11 @@ func expectFinishWrites(m buildsignalMocks, state entity.RequestState) *gomock.C } func expectAdvanceLastGreen(m buildsignalMocks, after *gomock.Call) *gomock.Call { - getCall := m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow(0, 5), nil) + getCall := m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow([]string{}, 5), nil) if after != nil { getCall.After(after) } - updated := queueRow(0, 5) + updated := queueRow([]string{}, 5) updated.LastGreenURI = requestWithState(entity.RequestStateSucceeded).URI updated.LastGreenRequestID = testID return m.queueStore.EXPECT().Update(gomock.Any(), updated, int32(5), int32(6)).Return(nil).After(getCall) @@ -296,7 +297,7 @@ func TestProcess(t *testing.T) { { // Neither state can reach the poll loop today — a build only exists once // process admitted the request. The guard covers them anyway because - // finishRequest would release a build slot they never claimed. + // the terminal path has no build outcome to record for them. name: "request that never claimed a slot is a no-op", setup: func(m buildsignalMocks) { m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusRunning, 2), nil) @@ -397,7 +398,7 @@ func TestProcess(t *testing.T) { // No build/queue/request write: the status is unchanged and the outcome // and bookmark are already recorded, so the slot must not be released a second time. eventCall := expectBuildFinished(m) - queue := queueRow(1, 6) + queue := queueRow([]string{testOtherID}, 6) queue.LastGreenRequestID = testID queue.LastGreenURI = request.URI getCall := m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queue, nil).After(eventCall) @@ -496,11 +497,31 @@ func TestProcess(t *testing.T) { m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil) eventCall := expectBuildFinished(m) - m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow(1, 4), nil).After(eventCall) - m.queueStore.EXPECT().Update(gomock.Any(), queueRow(0, 4), int32(4), int32(5)).Return(nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow([]string{testID}, 4), nil).After(eventCall) + m.queueStore.EXPECT().Update(gomock.Any(), queueRow([]string{}, 4), int32(4), int32(5)).Return(nil) m.reqStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(errors.New("db down")) }, }, + { + name: "outcome retry after its slot was released leaves another request's slot alone", + setup: func(m buildsignalMocks) { + m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(entity.BuildStatusSucceeded, 3), nil) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) + m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) + m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil) + eventCall := expectBuildFinished(m) + queue := queueRow([]string{testOtherID}, 5) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queue, nil).Times(2).After(eventCall) + outcomeCall := m.reqStore.EXPECT().Update(gomock.Any(), requestWithState(entity.RequestStateSucceeded), int32(1), int32(2)).Return(nil) + bookmarked := queue + bookmarked.LastGreenURI = requestWithState(entity.RequestStateSucceeded).URI + bookmarked.LastGreenRequestID = testID + bookmarkCall := m.queueStore.EXPECT().Update(gomock.Any(), bookmarked, int32(5), int32(6)).Return(nil).After(outcomeCall) + logCall := expectOutcomeLog(m, entity.RequestStateSucceeded, 2).After(bookmarkCall) + m.publisher.EXPECT().Publish(gomock.Any(), "record", gomock.Any()).Return(nil).After(logCall) + m.publisher.EXPECT().Publish(gomock.Any(), "buildanalysis", gomock.Any()).Return(nil).After(logCall) + }, + }, { name: "slot release failure leaves the request non-terminal", wantErr: true, @@ -729,13 +750,13 @@ func TestProcessRetriesBookmarkWithoutReleasingSlotAgain(t *testing.T) { m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil).Times(2) m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil).Times(2) m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(nil).Times(3) - queue := queueRow(1, 5) + queue := queueRow([]string{testOtherID}, 5) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).DoAndReturn(func(context.Context, string) (entity.Queue, error) { return queue, nil }).Times(2) m.queueStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(5), int32(6)).DoAndReturn( func(_ context.Context, updated entity.Queue, _, version int32) error { - assert.Equal(t, int32(1), updated.InFlightCount) + assert.Equal(t, []string{testOtherID}, updated.InFlightRequestIDs) assert.Equal(t, request.ID, updated.LastGreenRequestID) assert.Equal(t, request.URI, updated.LastGreenURI) if committed { @@ -767,8 +788,8 @@ func TestProcessBookmarkPreservesConcurrentQueueUpdates(t *testing.T) { m.runnerFactory.EXPECT().For(buildrunner.Config{QueueName: testQueue}).Return(m.runner, nil) m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(entity.BuildStatusSucceeded, nil, nil) m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(nil).Times(2) - stale := queueRow(0, 5) - fresh := queueRow(1, 6) + stale := queueRow([]string{}, 5) + fresh := queueRow([]string{testOtherID}, 6) fresh.LastGreenRequestID = "9" fresh.LastGreenURI = "newer-commit" gomock.InOrder( @@ -796,8 +817,8 @@ func TestProcessBookmarkUsesWinningRequestOutcome(t *testing.T) { m.runner.EXPECT().Status(gomock.Any(), entity.BuildID{ID: testBuildID}).Return(status, nil, nil) m.materializer.EXPECT().PersistLog(gomock.Any(), m.store, gomock.Any()).Return(nil).Times(2) gomock.InOrder( - m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow(1, 4), nil), - m.queueStore.EXPECT().Update(gomock.Any(), queueRow(0, 4), int32(4), int32(5)).Return(nil), + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queueRow([]string{testID}, 4), nil), + m.queueStore.EXPECT().Update(gomock.Any(), queueRow([]string{}, 4), int32(4), int32(5)).Return(nil), m.reqStore.EXPECT().Update(gomock.Any(), gomock.Any(), int32(1), int32(2)).Return(storage.ErrVersionMismatch), m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(state), nil), ) diff --git a/stovepipe/controller/dlq/build_test.go b/stovepipe/controller/dlq/build_test.go index c97686202..814175ec3 100644 --- a/stovepipe/controller/dlq/build_test.go +++ b/stovepipe/controller/dlq/build_test.go @@ -67,8 +67,8 @@ func TestBuildProcess(t *testing.T) { name: "processing request is failed", setup: func(m dlqMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) - m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, InFlightCount: 1, Version: 5}, nil) - m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{Name: testQueue, Version: 5}, int32(5), int32(6)).Return(nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, InFlightRequestIDs: []string{testID}, Version: 5}, nil) + m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{Name: testQueue, InFlightRequestIDs: []string{}, Version: 5}, int32(5), int32(6)).Return(nil) updated := requestWithState(entity.RequestStateProcessing) updated.State = entity.RequestStateFailed updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) diff --git a/stovepipe/controller/dlq/buildsignal.go b/stovepipe/controller/dlq/buildsignal.go index 69b2b9399..b2961c92c 100644 --- a/stovepipe/controller/dlq/buildsignal.go +++ b/stovepipe/controller/dlq/buildsignal.go @@ -39,7 +39,7 @@ const _buildSignalOpName = "buildsignal_dlq" // request via failRequest. // // This DLQ is the one that matters most. A request only reaches buildsignal -// after process admitted it, so it holds one of the queue's in_flight_count +// after process admitted it, so it holds one of the queue's // build slots, and buildsignal's terminal path is the only thing that gives that // slot back. Once a poll message dead-letters — a Status call that stayed broken // through every retry, an unknown build id, a storage write that kept failing — @@ -149,10 +149,6 @@ func (c *buildSignalController) Process(ctx context.Context, delivery consumer.D return nil } - // Every request reachable from a build row is either still processing, and holding - // the slot failRequest releases, or already terminal, and past releasing it: build - // triggers only once process has written the strategy, which lands in the same CAS - // as accepted→processing, and processing exits only to a terminal outcome. if err := failRequest(ctx, store, c.materializer, c.logger, build.RequestID, entity.RequestOutcomeReasonBuildPollingExhausted); err != nil { metrics.NamedCounter(c.metricsScope, _buildSignalOpName, "reconcile_errors", 1, metrics.TagsFromContext(ctx)...) return err diff --git a/stovepipe/controller/dlq/buildsignal_test.go b/stovepipe/controller/dlq/buildsignal_test.go index 08c0f4aab..7ffda57ae 100644 --- a/stovepipe/controller/dlq/buildsignal_test.go +++ b/stovepipe/controller/dlq/buildsignal_test.go @@ -103,10 +103,10 @@ func TestBuildSignalProcess(t *testing.T) { m.buildStore.EXPECT().Get(gomock.Any(), testBuildID).Return(build(), nil) m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ - Name: testQueue, InFlightCount: 1, Version: 5, + Name: testQueue, InFlightRequestIDs: []string{testID}, Version: 5, }, nil) m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, InFlightCount: 0, Version: 5, + Name: testQueue, InFlightRequestIDs: []string{}, Version: 5, }, int32(5), int32(6)).Return(nil) updated := requestWithState(entity.RequestStateProcessing) updated.State = entity.RequestStateFailed diff --git a/stovepipe/controller/dlq/dlq.go b/stovepipe/controller/dlq/dlq.go index cbde8e207..a2730d8e6 100644 --- a/stovepipe/controller/dlq/dlq.go +++ b/stovepipe/controller/dlq/dlq.go @@ -28,16 +28,19 @@ // transitions it to RequestStateFailed — the conservative unsuccessful outcome, which // whatever records greenness treats as not-green for gating (see entity.RequestState) — // with an idempotent optimistic-locking write so -// concurrent activity (a late successful pipeline transition) wins cleanly. If the request had -// already been admitted (processing) and was holding a concurrency slot, the -// reconciler also releases it by CAS-decrementing the queue's in_flight_count, per -// doc/rfc/stovepipe/steps/process.md#in_flight_count-integrity. +// concurrent activity (a late successful pipeline transition) wins cleanly. Before that write +// the reconciler releases the request's build slot by removing its ID from the queue's +// in_flight_request_ids. Ownership is keyed by request ID, so the release is idempotent and +// runs for any nonterminal request; a crash between process's claim and admit is repaired +// when the redelivery recognizes its own claim. See +// doc/rfc/stovepipe/steps/process.md#in_flight_request_ids-integrity. package dlq import ( "context" "errors" "fmt" + "slices" "github.com/uber/submitqueue/platform/consumer" "github.com/uber/submitqueue/stovepipe/core/requestlog" @@ -59,32 +62,8 @@ func TopicKey(main consumer.TopicKey) consumer.TopicKey { return consumer.TopicKey(string(main) + topicSuffix) } -// failRequest transitions request to RequestStateFailed if it is not already -// in a terminal state. If the request had reached RequestStateProcessing — meaning process's -// admit step already CAS-incremented the queue's in_flight_count for it and no terminal -// outcome has released it yet — the queue's -// 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. -// -// Queue and Request are separate entities with no cross-entity -// transaction, so the two writes cannot be atomic and the ordering picks which crash -// failure mode we accept: a crash between the writes leaves the request non-terminal, -// redelivery re-runs reconciliation, and releaseSlot (which tracks no per-request slot -// ownership) decrements again — transiently over-admitting by one slot until the -// under-count re-converges at releaseSlot's zero clamp. The reverse order would leak -// the slot instead: redelivery skips terminal requests, permanently shrinking the -// queue's capacity toward a wedge. Over-admission is the failure mode we prefer. See -// doc/rfc/stovepipe/steps/process.md#in_flight_count-integrity for the broader -// counter-drift story. +// failRequest marks a nonterminal request failed, releasing its build slot first: a terminal +// request is skipped on redelivery, so a slot left behind would never be released. func failRequest( ctx context.Context, store storage.Storage, @@ -117,10 +96,8 @@ func failRequest( return nil } - if request.State == entity.RequestStateProcessing { - if err := releaseSlot(ctx, store, logger, request.Queue); err != nil { - return fmt.Errorf("failed to release queue slot for request %s: %w", requestID, err) - } + if err := releaseSlot(ctx, store, logger, request); err != nil { + return fmt.Errorf("failed to release queue slot for request %s: %w", requestID, err) } updated := request @@ -154,42 +131,41 @@ func persistFailureLog( return nil } -// releaseSlot CAS-decrements the queue's in_flight_count, retrying on version -// conflicts, mirroring process.Controller's own CAS-retry loop for queue updates. -func releaseSlot(ctx context.Context, store storage.Storage, logger *zap.SugaredLogger, queueName string) error { +// releaseSlot CAS-removes request's ID from the queue's in_flight_request_ids, writing +// nothing when it is absent and retrying on version conflicts. +func releaseSlot(ctx context.Context, store storage.Storage, logger *zap.SugaredLogger, request entity.Request) error { queueStore := store.GetQueueStore() for { - queueRow, err := queueStore.Get(ctx, queueName) + queueRow, err := queueStore.Get(ctx, request.Queue) if err != nil { if errors.Is(err, storage.ErrNotFound) { logger.Warnw("dlq reconcile: queue not found, skipping slot release", - "queue", queueName, + "queue", request.Queue, ) return nil } - return fmt.Errorf("failed to get queue %s: %w", queueName, err) + return fmt.Errorf("failed to get queue %s: %w", request.Queue, err) } - if queueRow.InFlightCount <= 0 { - logger.Warnw("dlq reconcile: queue in_flight_count already at zero, skipping slot release", - "queue", queueName, - ) + ownerIndex := slices.Index(queueRow.InFlightRequestIDs, request.ID) + if ownerIndex < 0 { return nil } updated := queueRow - updated.InFlightCount-- + updated.InFlightRequestIDs = slices.Delete(slices.Clone(queueRow.InFlightRequestIDs), ownerIndex, ownerIndex+1) newVersion := queueRow.Version + 1 if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil { if errors.Is(err, storage.ErrVersionMismatch) { continue } - return fmt.Errorf("failed to release slot for queue %s: %w", queueName, err) + return fmt.Errorf("failed to release slot for request %s in queue %s: %w", request.ID, request.Queue, err) } logger.Infow("dlq reconcile: released queue slot", - "queue", queueName, - "in_flight_count", updated.InFlightCount, + "queue", request.Queue, + "request_id", request.ID, + "in_flight_request_ids", updated.InFlightRequestIDs, ) return nil } diff --git a/stovepipe/controller/dlq/dlq_test.go b/stovepipe/controller/dlq/dlq_test.go index eeb7c838c..288ea8d9a 100644 --- a/stovepipe/controller/dlq/dlq_test.go +++ b/stovepipe/controller/dlq/dlq_test.go @@ -134,11 +134,15 @@ func TestProcess(t *testing.T) { wantMetric string }{ { - // No queue expectations: an accepted request never claimed a slot, - // so releasing one here would over-admit against MaxConcurrent. - name: "accepted request is marked failed without releasing a slot", + name: "accepted request releases a slot claimed before admission", setup: func(m dlqMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateAccepted), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, InFlightRequestIDs: []string{testID}, Version: 5, + }, nil) + m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ + Name: testQueue, InFlightRequestIDs: []string{}, Version: 5, + }, int32(5), int32(6)).Return(nil) updated := requestWithState(entity.RequestStateAccepted) updated.State = entity.RequestStateFailed updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) @@ -150,10 +154,10 @@ func TestProcess(t *testing.T) { setup: func(m dlqMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ - Name: testQueue, InFlightCount: 1, Version: 5, + Name: testQueue, InFlightRequestIDs: []string{testID}, Version: 5, }, nil) m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, InFlightCount: 0, Version: 5, + Name: testQueue, InFlightRequestIDs: []string{}, Version: 5, }, int32(5), int32(6)).Return(nil) updated := requestWithState(entity.RequestStateProcessing) updated.State = entity.RequestStateFailed @@ -184,6 +188,7 @@ func TestProcess(t *testing.T) { name: "request update retries on version mismatch", setup: func(m dlqMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateAccepted), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, Version: 5}, nil) updated := requestWithState(entity.RequestStateAccepted) updated.State = entity.RequestStateFailed m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(storage.ErrVersionMismatch) @@ -195,16 +200,16 @@ func TestProcess(t *testing.T) { setup: func(m dlqMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ - Name: testQueue, InFlightCount: 1, Version: 5, + Name: testQueue, InFlightRequestIDs: []string{testID}, Version: 5, }, nil) m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, InFlightCount: 0, Version: 5, + Name: testQueue, InFlightRequestIDs: []string{}, Version: 5, }, int32(5), int32(6)).Return(storage.ErrVersionMismatch) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ - Name: testQueue, InFlightCount: 1, Version: 6, + Name: testQueue, InFlightRequestIDs: []string{testID}, Version: 6, }, nil) m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, InFlightCount: 0, Version: 6, + Name: testQueue, InFlightRequestIDs: []string{}, Version: 6, }, int32(6), int32(7)).Return(nil) updated := requestWithState(entity.RequestStateProcessing) updated.State = entity.RequestStateFailed @@ -213,11 +218,11 @@ func TestProcess(t *testing.T) { }, }, { - name: "queue already drained is a no-op for slot release", + name: "slot release writes nothing when only another request holds a slot", setup: func(m dlqMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateProcessing), nil) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ - Name: testQueue, InFlightCount: 0, Version: 5, + Name: testQueue, InFlightRequestIDs: []string{"8"}, Version: 5, }, nil) updated := requestWithState(entity.RequestStateProcessing) updated.State = entity.RequestStateFailed @@ -230,6 +235,7 @@ func TestProcess(t *testing.T) { wantErr: true, setup: func(m dlqMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(requestWithState(entity.RequestStateAccepted), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{Name: testQueue, Version: 5}, nil) updated := requestWithState(entity.RequestStateAccepted) updated.State = entity.RequestStateFailed updateCall := m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(2), int32(3)).Return(nil) diff --git a/stovepipe/controller/ingest_test.go b/stovepipe/controller/ingest_test.go index 93edd2623..352ffb95b 100644 --- a/stovepipe/controller/ingest_test.go +++ b/stovepipe/controller/ingest_test.go @@ -594,7 +594,7 @@ func TestIngestInitializesQueueEnablement(t *testing.T) { f := newIngestEnablementFixture(t) f.queue = entity.Queue{} if existing { - f.queue = entity.Queue{Name: testQueue, Version: 5, InFlightCount: 3, LastGreenRequestID: "4"} + f.queue = entity.Queue{Name: testQueue, Version: 5, InFlightRequestIDs: []string{"1", "2", "3"}, LastGreenRequestID: "4"} } f.summary = entity.QueueSummary{} clear(f.logs) @@ -613,7 +613,7 @@ func TestIngestInitializesQueueEnablement(t *testing.T) { require.Len(t, f.logs, int(decision.version)) require.Equal(t, "7", f.queue.LatestRequestID) if existing { - require.Equal(t, int32(3), f.queue.InFlightCount) + require.Equal(t, []string{"1", "2", "3"}, f.queue.InFlightRequestIDs) require.Equal(t, "4", f.queue.LastGreenRequestID) } retry, err := f.controller.Ingest(context.Background(), request) diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index f334f70bd..46ac0ddc7 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -22,6 +22,7 @@ import ( "context" "errors" "fmt" + "slices" "github.com/uber-go/tally" basehook "github.com/uber/submitqueue/api/base/hook" @@ -179,7 +180,7 @@ func (c *Controller) processAccepted(ctx context.Context, store storage.Storage, return nil } - superseded, err := c.coalesce(ctx, store, request, queueRow.LatestRequestID) + superseded, err := c.coalesce(ctx, store, request, queueRow) if err != nil || superseded { return err } @@ -194,17 +195,23 @@ func (c *Controller) processAccepted(ctx context.Context, store storage.Storage, return c.admitLatestHead(ctx, store, delivery, request, queueRow, cfg) } -// coalesce supersedes request when a newer head exists (RFC process step 5), returning -// true so the caller acks. It returns false when request is still the latest head and -// should proceed to the gate. Superseding consumes no build slot. -func (c *Controller) coalesce(ctx context.Context, store storage.Storage, request entity.Request, latestRequestID string) (bool, error) { - cmp, err := entity.CompareRequestID(request.ID, latestRequestID) +// coalesce supersedes request when a newer head exists (RFC process step 5), returning true so +// the caller acks. A request that claimed a build slot before the newer head arrived releases it first. +func (c *Controller) coalesce(ctx context.Context, store storage.Storage, request entity.Request, queueRow entity.Queue) (bool, error) { + cmp, err := entity.CompareRequestID(request.ID, queueRow.LatestRequestID) if err != nil { return false, fmt.Errorf("failed to compare request ids for queue %s: %w", request.Queue, err) } if cmp >= 0 { return false, nil } + // Release before superseding: nothing revisits a superseded request, so its slot would leak. + if slices.Contains(queueRow.InFlightRequestIDs, request.ID) { + if err := c.releaseBuildSlot(ctx, store, request); err != nil { + metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) + return false, err + } + } request, err = c.supersedeRequest(ctx, store, request) if err != nil { metrics.NamedCounter(c.metricsScope, _opName, "storage_errors", 1, metrics.TagsFromContext(ctx)...) @@ -220,7 +227,7 @@ func (c *Controller) coalesce(ctx context.Context, store storage.Storage, reques c.logger.Infow("superseded request for newer head", "request_id", request.ID, "queue", request.Queue, - "latest_request_id", latestRequestID, + "latest_request_id", queueRow.LatestRequestID, ) return true, nil } @@ -236,8 +243,8 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, var err error for { - if queueRow.InFlightCount >= cfg.MaxConcurrent { - return c.holdForBuildSlot(ctx, delivery, request, queueRow.InFlightCount, cfg.GateWaitDelayMs) + if !slices.Contains(queueRow.InFlightRequestIDs, request.ID) && len(queueRow.InFlightRequestIDs) >= int(cfg.MaxConcurrent) { + return c.holdForBuildSlot(ctx, delivery, request, queueRow.InFlightRequestIDs, cfg.GateWaitDelayMs) } if queueRow.LastGreenURI != "" && sc == nil { @@ -255,7 +262,7 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, return err } - err = c.claimBuildSlot(ctx, store, &queueRow) + err = c.claimBuildSlot(ctx, store, &queueRow, request.ID) if err == nil { break } @@ -264,7 +271,7 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, } // 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) + superseded, err := c.coalesce(ctx, store, request, queueRow) if err != nil || superseded { return err } @@ -272,14 +279,14 @@ func (c *Controller) admitLatestHead(ctx context.Context, store storage.Storage, 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) + // The claim stays: the redelivery resumes it, or the DLQ releases it. return err } if !transitioned { - // Lost the admit race: another delivery advanced this request. Release and skip. - c.releaseBuildSlot(ctx, store, request.Queue) + // A duplicate admission owns the slot; a terminal transition owns nothing. + if request.State.IsTerminal() { + return c.releaseBuildSlot(ctx, store, request) + } return nil } @@ -349,13 +356,16 @@ 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-adds requestID to queue.in_flight_request_ids unless it is already there. 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, requestID string) error { + if slices.Contains(queueRow.InFlightRequestIDs, requestID) { + return nil + } queueStore := store.GetQueueStore() updated := *queueRow - updated.InFlightCount = queueRow.InFlightCount + 1 + updated.InFlightRequestIDs = append(slices.Clone(queueRow.InFlightRequestIDs), requestID) newVersion := queueRow.Version + 1 if err := queueStore.Update(ctx, updated, queueRow.Version, newVersion); err != nil { if errors.Is(err, storage.ErrVersionMismatch) { @@ -376,8 +386,7 @@ func (c *Controller) claimBuildSlot(ctx context.Context, store storage.Storage, // 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. +// means a concurrent writer already advanced the request past accepted. 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,40 +417,32 @@ 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) { +// releaseBuildSlot CAS-removes request's ID from queue.in_flight_request_ids, writing nothing +// when it is absent and retrying on version conflicts. +func (c *Controller) releaseBuildSlot(ctx context.Context, store storage.Storage, request entity.Request) error { queueStore := store.GetQueueStore() for { - queueRow, err := queueStore.Get(ctx, queueName) + queueRow, err := queueStore.Get(ctx, request.Queue) if err != nil { - c.logger.Errorw("failed to release claimed build slot", - "queue", queueName, - "error", err, - ) - return + return fmt.Errorf("failed to load queue %s to release build slot: %w", request.Queue, err) } - if queueRow.InFlightCount <= 0 { - return + ownerIndex := slices.Index(queueRow.InFlightRequestIDs, request.ID) + if ownerIndex < 0 { + return nil } updated := queueRow - updated.InFlightCount = queueRow.InFlightCount - 1 + updated.InFlightRequestIDs = slices.Delete(slices.Clone(queueRow.InFlightRequestIDs), ownerIndex, ownerIndex+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 + return fmt.Errorf("failed to release build slot for request %s in queue %s: %w", request.ID, request.Queue, err) } metrics.NamedCounter(c.metricsScope, _opName, "slot_released", 1, metrics.TagsFromContext(ctx)...) - return + return nil } } @@ -477,7 +478,7 @@ func (c *Controller) supersedeRequest(ctx context.Context, store storage.Storage // delayMs and the gate is re-checked, without burning MaxAttempts — the partition // (keyed by queue name) waits with it. delayMs must be positive: a non-positive // hold would redeliver immediately and hot-loop the gate check. -func (c *Controller) holdForBuildSlot(ctx context.Context, delivery consumer.Delivery, request entity.Request, inFlightCount int32, delayMs int64) error { +func (c *Controller) holdForBuildSlot(ctx context.Context, delivery consumer.Delivery, request entity.Request, inFlightRequestIDs []string, delayMs int64) error { if delayMs <= 0 { metrics.NamedCounter(c.metricsScope, _opName, "config_errors", 1, metrics.TagsFromContext(ctx)...) return fmt.Errorf("requires a positive gate wait delay for queue %s, got %dms", request.Queue, delayMs) @@ -488,7 +489,7 @@ func (c *Controller) holdForBuildSlot(ctx context.Context, delivery consumer.Del "request_id", request.ID, "queue", request.Queue, "uri", request.URI, - "in_flight_count", inFlightCount, + "in_flight_request_ids", inFlightRequestIDs, "delay_ms", delayMs, ) return nil diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index cd8ef4c61..d243e7772 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -176,10 +176,10 @@ func expectAdmit(t *testing.T, m processMocks, id string) { t.Helper() updatedQueue := entity.Queue{ - Name: testQueue, - LatestRequestID: id, - InFlightCount: 1, - Version: 1, + Name: testQueue, + LatestRequestID: id, + InFlightRequestIDs: []string{id}, + Version: 1, } m.queueStore.EXPECT().Update(gomock.Any(), updatedQueue, int32(1), int32(2)).Return(nil) @@ -466,9 +466,9 @@ func TestProcessRederivesStrategyAfterQueueReload(t *testing.T) { Version: 2, } initialClaim := initialQueue - initialClaim.InFlightCount = 1 + initialClaim.InFlightRequestIDs = []string{testID} claimedQueue := reloadedQueue - claimedQueue.InFlightCount = 1 + claimedQueue.InFlightRequestIDs = []string{testID} updatedRequest := request updatedRequest.State = entity.RequestStateProcessing @@ -601,6 +601,48 @@ func TestProcess(t *testing.T) { expectAdmit(t, m, testID) }, }, + { + name: "superseded slot holder releases its slot before superseding", + id: testOlderID, + setup: func(m processMocks) { + queue := entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + InFlightRequestIDs: []string{testOlderID}, + Version: 2, + } + m.reqStore.EXPECT().Get(gomock.Any(), testOlderID).Return(acceptedRequest(testOlderID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(queue, nil).Times(2) + released := queue + released.InFlightRequestIDs = []string{} + releaseCall := m.queueStore.EXPECT().Update(gomock.Any(), released, int32(2), int32(3)).Return(nil) + superseded := acceptedRequest(testOlderID) + superseded.State = entity.RequestStateSuperseded + updateCall := m.reqStore.EXPECT().Update(gomock.Any(), superseded, int32(1), int32(2)).Return(nil).After(releaseCall) + loggedRequest := superseded + loggedRequest.Version = 2 + expectStateLog(t, m, loggedRequest, entity.RequestOutcomeReasonSupersededByNewerHead).After(updateCall) + }, + }, + { + name: "latest slot holder resumes admission without a queue write", + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + InFlightRequestIDs: []string{testID}, + Version: 2, + }, nil) + updatedReq := acceptedRequest(testID) + updatedReq.State = entity.RequestStateProcessing + updatedReq.BuildStrategy = entity.BuildStrategyFull + m.reqStore.EXPECT().Update(gomock.Any(), updatedReq, int32(1), int32(2)).Return(nil) + loggedRequest := updatedReq + loggedRequest.Version = 2 + expectStartAnnounceAndBuildPublish(t, m, loggedRequest) + }, + }, { name: "start announce failure leaves the request admitted for a redelivery to re-announce", wantErr: true, @@ -612,10 +654,10 @@ func TestProcess(t *testing.T) { Version: 1, }, nil) updatedQueue := entity.Queue{ - Name: testQueue, - LatestRequestID: testID, - InFlightCount: 1, - Version: 1, + Name: testQueue, + LatestRequestID: testID, + InFlightRequestIDs: []string{testID}, + Version: 1, } m.queueStore.EXPECT().Update(gomock.Any(), updatedQueue, int32(1), int32(2)).Return(nil) updatedRequest := acceptedRequest(testID) @@ -642,10 +684,10 @@ func TestProcess(t *testing.T) { Version: 1, }, nil) updatedQueue := entity.Queue{ - Name: testQueue, - LatestRequestID: testID, - InFlightCount: 1, - Version: 1, + Name: testQueue, + LatestRequestID: testID, + InFlightRequestIDs: []string{testID}, + Version: 1, } m.queueStore.EXPECT().Update(gomock.Any(), updatedQueue, int32(1), int32(2)).Return(nil) updatedRequest := acceptedRequest(testID) @@ -674,10 +716,10 @@ func TestProcess(t *testing.T) { Version: 1, }, nil) updatedQueue := entity.Queue{ - Name: testQueue, - LatestRequestID: testID, - InFlightCount: 1, - Version: 1, + Name: testQueue, + LatestRequestID: testID, + InFlightRequestIDs: []string{testID}, + Version: 1, } m.queueStore.EXPECT().Update(gomock.Any(), updatedQueue, int32(1), int32(2)).Return(nil) updatedRequest := acceptedRequest(testID) @@ -725,11 +767,11 @@ func TestProcess(t *testing.T) { setup: func(m processMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ - Name: testQueue, - LatestRequestID: testID, - InFlightCount: 1, - LastGreenURI: "git://repo/monorepo/main/green", - Version: 1, + Name: testQueue, + LatestRequestID: testID, + InFlightRequestIDs: []string{testOlderID}, + LastGreenURI: "git://repo/monorepo/main/green", + Version: 1, }, nil) }, }, @@ -744,16 +786,16 @@ func TestProcess(t *testing.T) { Version: 1, }, nil) m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, - LatestRequestID: testID, - InFlightCount: 1, - Version: 1, + Name: testQueue, + LatestRequestID: testID, + InFlightRequestIDs: []string{testID}, + Version: 1, }, int32(1), int32(2)).Return(storage.ErrVersionMismatch) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ - Name: testQueue, - LatestRequestID: testID, - InFlightCount: 1, - Version: 2, + Name: testQueue, + LatestRequestID: testID, + InFlightRequestIDs: []string{testOlderID}, + Version: 2, }, nil) }, }, @@ -766,7 +808,7 @@ func TestProcess(t *testing.T) { }, nil) // First claim CAS loses to a concurrent writer. m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 1, + Name: testQueue, LatestRequestID: testID, InFlightRequestIDs: []string{testID}, Version: 1, }, int32(1), int32(2)).Return(storage.ErrVersionMismatch) // Reload: still latest, slot still free (version advanced by an unrelated field). m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ @@ -774,7 +816,7 @@ func TestProcess(t *testing.T) { }, nil) // Retry claim succeeds, then admit. m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 2, + Name: testQueue, LatestRequestID: testID, InFlightRequestIDs: []string{testID}, Version: 2, }, int32(2), int32(3)).Return(nil) updatedReq := acceptedRequest(testID) updatedReq.State = entity.RequestStateProcessing @@ -793,7 +835,7 @@ func TestProcess(t *testing.T) { Name: testQueue, LatestRequestID: testID, Version: 1, }, nil) m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 1, + Name: testQueue, LatestRequestID: testID, InFlightRequestIDs: []string{testID}, Version: 1, }, int32(1), int32(2)).Return(storage.ErrVersionMismatch) // Reload: ingest stamped a newer head — our head is no longer latest. m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ @@ -819,7 +861,7 @@ func TestProcess(t *testing.T) { m.sourceFactory.EXPECT().For(sourcecontrol.Config{QueueName: testQueue}).Return(m.sourceControl, nil) m.sourceControl.EXPECT().IsAncestor(gomock.Any(), lastGreenURI, testURI).Return(true, nil) m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 1, LastGreenURI: lastGreenURI, Version: 1, + Name: testQueue, LatestRequestID: testID, InFlightRequestIDs: []string{testID}, LastGreenURI: lastGreenURI, Version: 1, }, int32(1), int32(2)).Return(nil) firstAttempt := acceptedRequest(testID) @@ -843,17 +885,15 @@ func TestProcess(t *testing.T) { }, }, { - name: "mark processing lost race releases slot and skips admit", + name: "mark processing lost to a duplicate admission leaves the slot to the winner", setup: func(m processMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ Name: testQueue, LatestRequestID: testID, Version: 1, }, nil) - // Claim succeeds. m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 1, + Name: testQueue, LatestRequestID: testID, InFlightRequestIDs: []string{testID}, Version: 1, }, int32(1), int32(2)).Return(nil) - // markProcessing CAS loses, reload shows a concurrent writer already advanced it. updatedReq := acceptedRequest(testID) updatedReq.State = entity.RequestStateProcessing updatedReq.BuildStrategy = entity.BuildStrategyFull @@ -861,17 +901,35 @@ func TestProcess(t *testing.T) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{ ID: testID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 2, }, nil) - // Compensating decrement of the spurious slot. + }, + }, + { + name: "mark processing lost to a terminal transition releases the claim", + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 2, + Name: testQueue, LatestRequestID: testID, Version: 1, + }, nil) + m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ + Name: testQueue, LatestRequestID: testID, InFlightRequestIDs: []string{testID}, Version: 1, + }, int32(1), int32(2)).Return(nil) + updatedReq := acceptedRequest(testID) + updatedReq.State = entity.RequestStateProcessing + updatedReq.BuildStrategy = entity.BuildStrategyFull + m.reqStore.EXPECT().Update(gomock.Any(), updatedReq, int32(1), int32(2)).Return(storage.ErrVersionMismatch) + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{ + ID: testID, Queue: testQueue, State: entity.RequestStateFailed, Version: 2, + }, nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, LatestRequestID: testID, InFlightRequestIDs: []string{testID}, Version: 2, }, nil) m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 0, Version: 2, + Name: testQueue, LatestRequestID: testID, InFlightRequestIDs: []string{}, Version: 2, }, int32(2), int32(3)).Return(nil) }, }, { - name: "mark processing error releases slot and returns error", + name: "mark processing error keeps the claim for the redelivery", wantErr: true, setup: func(m processMocks) { m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(acceptedRequest(testID), nil) @@ -879,19 +937,12 @@ func TestProcess(t *testing.T) { Name: testQueue, LatestRequestID: testID, Version: 1, }, nil) m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 1, + Name: testQueue, LatestRequestID: testID, InFlightRequestIDs: []string{testID}, Version: 1, }, int32(1), int32(2)).Return(nil) updatedReq := acceptedRequest(testID) updatedReq.State = entity.RequestStateProcessing updatedReq.BuildStrategy = entity.BuildStrategyFull m.reqStore.EXPECT().Update(gomock.Any(), updatedReq, int32(1), int32(2)).Return(errors.New("db down")) - // Best-effort compensating decrement before the error propagates. - m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 1, Version: 2, - }, nil) - m.queueStore.EXPECT().Update(gomock.Any(), entity.Queue{ - Name: testQueue, LatestRequestID: testID, InFlightCount: 0, Version: 2, - }, int32(2), int32(3)).Return(nil) }, }, { @@ -1047,7 +1098,7 @@ func TestHoldForBuildSlotRequiresPositiveDelay(t *testing.T) { d := consumermock.NewMockDelivery(ctrl) // No Hold expectation: the guard must reject before recording a hold. - err := c.holdForBuildSlot(queueContext(testQueue), d, acceptedRequest(testID), 1, 0) + err := c.holdForBuildSlot(queueContext(testQueue), d, acceptedRequest(testID), []string{testOlderID}, 0) require.Error(t, err) assert.False(t, errs.IsRetryable(err)) diff --git a/stovepipe/core/queuestate/queuestate_test.go b/stovepipe/core/queuestate/queuestate_test.go index 8c4f99cc9..e222ebf01 100644 --- a/stovepipe/core/queuestate/queuestate_test.go +++ b/stovepipe/core/queuestate/queuestate_test.go @@ -463,7 +463,7 @@ func TestStateChangesPreserveOperationalQueueUpdates(t *testing.T) { mutateQueue := func() { f.queue.Version++ f.queue.LastGreenURI = "green" - f.queue.InFlightCount = 3 + f.queue.InFlightRequestIDs = []string{"6"} f.queue.LatestRequestID = "7" } if duringWrite { @@ -476,7 +476,7 @@ func TestStateChangesPreserveOperationalQueueUpdates(t *testing.T) { require.Equal(t, int64(2), applied.Version) require.Equal(t, int32(3), f.queue.Version) require.Equal(t, "green", f.queue.LastGreenURI) - require.Equal(t, int32(3), f.queue.InFlightCount) + require.Equal(t, []string{"6"}, f.queue.InFlightRequestIDs) require.Equal(t, "7", f.queue.LatestRequestID) require.Equal(t, applied, f.summary.Snapshot()) require.Len(t, f.logs, 2) @@ -486,11 +486,11 @@ func TestStateChangesPreserveOperationalQueueUpdates(t *testing.T) { func TestStateInitializationAdoptsExistingOperationalQueue(t *testing.T) { f := newStateFixture() - f.queue = entity.Queue{Name: "repo/main", Version: 12, InFlightCount: 2, LastGreenURI: "green", LatestRequestID: "9"} + f.queue = entity.Queue{Name: "repo/main", Version: 12, InFlightRequestIDs: []string{"8", "9"}, LastGreenURI: "green", LatestRequestID: "9"} initial := initializeFixture(t, f) require.Equal(t, int64(1), initial.Version) require.Equal(t, int32(13), f.queue.Version) - require.Equal(t, int32(2), f.queue.InFlightCount) + require.Equal(t, []string{"8", "9"}, f.queue.InFlightRequestIDs) require.Equal(t, "green", f.queue.LastGreenURI) require.Equal(t, "9", f.queue.LatestRequestID) require.Equal(t, initial, f.summary.Snapshot()) diff --git a/stovepipe/entity/queue.go b/stovepipe/entity/queue.go index 9ca1c8fcc..84517d894 100644 --- a/stovepipe/entity/queue.go +++ b/stovepipe/entity/queue.go @@ -36,8 +36,9 @@ type Queue struct { // than the candidate's, compared by ingest order via CompareRequestID. LastGreenRequestID string `json:"last_green_request_id"` - // InFlightCount is the number of trunk validations admitted by process but not yet terminal. - InFlightCount int32 `json:"in_flight_count"` + // InFlightRequestIDs holds the IDs of requests admitted to a build slot and not yet released, + // each present at most once. Its length is the queue's in-flight count. + InFlightRequestIDs []string `json:"in_flight_request_ids"` // LatestRequestID is the request id of the newest head ingest accepted for this queue. // Empty until the first request is created. Coalescing compares IDs via CompareRequestID. diff --git a/stovepipe/entity/queue_config.go b/stovepipe/entity/queue_config.go index b6681cd83..ed6d4c90a 100644 --- a/stovepipe/entity/queue_config.go +++ b/stovepipe/entity/queue_config.go @@ -15,7 +15,7 @@ package entity // QueueConfig holds deployment configuration for a Stovepipe validation queue. -// Mutable runtime state (latest head, in-flight count) lives on the Queue row; +// Mutable runtime state (latest head, in-flight requests) lives on the Queue row; // knobs such as max_concurrent are resolved here at gate-check time. Immutable // after load. type QueueConfig struct { diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md index b456ee8b5..ae59615ac 100644 --- a/stovepipe/extension/storage/README.md +++ b/stovepipe/extension/storage/README.md @@ -24,6 +24,8 @@ Same design space as [`submitqueue/extension/storage`](../../../submitqueue/exte `QueueStore` owns internal enablement alongside pipeline coordination. `QueueSummaryStore` persists the current read model projected from immutable `QueueLogStore` occurrences. Enablement sequences, operational row versions and projection versions are distinct. These stores use primary-key operations; version arithmetic, duplicate reconciliation and projection repair belong to business logic. See [queue state](../../../doc/rfc/stovepipe/queue-state.md). +Existing MySQL deployments must add `in_flight_request_ids JSON NOT NULL DEFAULT (JSON_ARRAY())` to `queue` (MySQL 8.0.13+ for the expression default) before deploying the new readers and writers. The column is added alongside `in_flight_count`, which is retained unused for schema compatibility. The default fills existing rows and older writers' inserts, so no backfill is needed. Slots held at upgrade time are not carried into the set; those requests release a no-op and the gate over-admits by at most that many until they finish. + `ProjectFactByDegreeStore` is the ordered lookup mapping for named project facts. Its immutable `(queue, URI, degree DESC, project ASC)` primary key stores the original degree and orders by descending degree and ascending bytewise project ID, so listing is a bounded key-range read. It holds no duplicate validation payload; readers hydrate facts by their original keys and verify the mapping. The record controller ensures mappings from persisted winning facts before recording project completion, including retries after partial writes. Create the table before deploying writers. The table starts empty: only results processed by index-aware writers receive mappings, with no historical population or read-time fallback. ## Build issue observations diff --git a/stovepipe/extension/storage/mysql/queue_store.go b/stovepipe/extension/storage/mysql/queue_store.go index c4fa43ca3..9246723ad 100644 --- a/stovepipe/extension/storage/mysql/queue_store.go +++ b/stovepipe/extension/storage/mysql/queue_store.go @@ -17,6 +17,7 @@ package mysql import ( "context" "database/sql" + "encoding/json" "errors" "fmt" @@ -47,13 +48,17 @@ func (q *queueStore) Create(ctx context.Context, queue entity.Queue) (retErr err if queue.Name != q.queue { return fmt.Errorf("queue row name %q does not match the store's bound queue %q", queue.Name, q.queue) } + inFlightRequestIDs, err := marshalInFlightRequestIDs(queue) + if err != nil { + return err + } - _, err := q.db.ExecContext(ctx, - `INSERT INTO queue (name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms) + _, err = q.db.ExecContext(ctx, + `INSERT INTO queue (name, last_green_uri, in_flight_request_ids, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, queue.Name, queue.LastGreenURI, - queue.InFlightCount, + inFlightRequestIDs, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID, @@ -78,13 +83,14 @@ func (q *queueStore) Get(ctx context.Context, name string) (ret entity.Queue, re } var queue entity.Queue + var inFlightRequestIDs []byte err := q.db.QueryRowContext(ctx, - "SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms FROM queue WHERE name = ?", + "SELECT name, last_green_uri, in_flight_request_ids, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms FROM queue WHERE name = ?", name, ).Scan( &queue.Name, &queue.LastGreenURI, - &queue.InFlightCount, + &inFlightRequestIDs, &queue.LatestRequestID, &queue.Version, &queue.LastGreenRequestID, @@ -97,6 +103,14 @@ func (q *queueStore) Get(ctx context.Context, name string) (ret entity.Queue, re if err != nil { return entity.Queue{}, fmt.Errorf("failed to get queue name=%s from the database: %w", name, err) } + var ids []string + if err := json.Unmarshal(inFlightRequestIDs, &ids); err != nil { + return entity.Queue{}, fmt.Errorf("failed to decode in_flight_request_ids for queue name=%s: %w", name, err) + } + // An empty set reads back as nil so a zero-valued Queue round-trips unchanged. + if len(ids) > 0 { + queue.InFlightRequestIDs = ids + } return queue, nil } @@ -110,13 +124,17 @@ func (q *queueStore) Update(ctx context.Context, queue entity.Queue, oldVersion, if queue.Name != q.queue { return fmt.Errorf("queue row name %q does not match the store's bound queue %q", queue.Name, q.queue) } + inFlightRequestIDs, err := marshalInFlightRequestIDs(queue) + if err != nil { + return err + } result, err := q.db.ExecContext(ctx, `UPDATE queue - SET last_green_uri = ?, in_flight_count = ?, latest_request_id = ?, version = ?, last_green_request_id = ?, enablement = ?, enablement_version = ?, effective_from_commit_uri = ?, enablement_changed_at_ms = ? + SET last_green_uri = ?, in_flight_request_ids = ?, latest_request_id = ?, version = ?, last_green_request_id = ?, enablement = ?, enablement_version = ?, effective_from_commit_uri = ?, enablement_changed_at_ms = ? WHERE name = ? AND version = ?`, queue.LastGreenURI, - queue.InFlightCount, + inFlightRequestIDs, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, @@ -148,3 +166,17 @@ func (q *queueStore) Update(ctx context.Context, queue entity.Queue, oldVersion, return nil } + +// marshalInFlightRequestIDs encodes the queue's in-flight request IDs as a JSON array, +// writing [] rather than null for an empty set. +func marshalInFlightRequestIDs(queue entity.Queue) ([]byte, error) { + ids := queue.InFlightRequestIDs + if ids == nil { + ids = []string{} + } + encoded, err := json.Marshal(ids) + if err != nil { + return nil, fmt.Errorf("failed to encode in_flight_request_ids for queue name=%s: %w", queue.Name, err) + } + return encoded, nil +} diff --git a/stovepipe/extension/storage/mysql/queue_store_test.go b/stovepipe/extension/storage/mysql/queue_store_test.go index e96365daf..c1ec557bd 100644 --- a/stovepipe/extension/storage/mysql/queue_store_test.go +++ b/stovepipe/extension/storage/mysql/queue_store_test.go @@ -45,7 +45,6 @@ func TestQueueStore_Create(t *testing.T) { Name: "monorepo/main", LastGreenURI: "git://remote/monorepo/main/green", LastGreenRequestID: "1", - InFlightCount: 0, LatestRequestID: "1", Version: 1, } @@ -60,7 +59,7 @@ func TestQueueStore_Create(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO queue"). - WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs). + WithArgs(queue.Name, queue.LastGreenURI, []byte("[]"), queue.LatestRequestID, queue.Version, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -68,7 +67,7 @@ func TestQueueStore_Create(t *testing.T) { name: "duplicate name returns ErrAlreadyExists", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO queue"). - WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs). + WithArgs(queue.Name, queue.LastGreenURI, []byte("[]"), queue.LatestRequestID, queue.Version, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs). WillReturnError(&mysql.MySQLError{Number: mysqlErrDuplicateEntry}) }, wantErr: true, @@ -78,7 +77,7 @@ func TestQueueStore_Create(t *testing.T) { name: "other exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("INSERT INTO queue"). - WithArgs(queue.Name, queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, queue.Version, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs). + WithArgs(queue.Name, queue.LastGreenURI, []byte("[]"), queue.LatestRequestID, queue.Version, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -111,7 +110,7 @@ func TestQueueStore_Get(t *testing.T) { Name: "monorepo/main", LastGreenURI: "git://remote/monorepo/main/green", LastGreenRequestID: "2", - InFlightCount: 2, + InFlightRequestIDs: []string{"2", "3"}, LatestRequestID: "3", Version: 3, } @@ -128,9 +127,9 @@ func TestQueueStore_Get(t *testing.T) { name: "found", queueName: want.Name, setup: func(mock sqlmock.Sqlmock) { - rows := sqlmock.NewRows([]string{"name", "last_green_uri", "in_flight_count", "latest_request_id", "version", "last_green_request_id", "enablement", "enablement_version", "effective_from_commit_uri", "enablement_changed_at_ms"}). - AddRow(want.Name, want.LastGreenURI, want.InFlightCount, want.LatestRequestID, want.Version, want.LastGreenRequestID, want.Enablement, want.EnablementVersion, want.EffectiveFromCommitURI, want.EnablementChangedAtMs) - mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms FROM queue"). + rows := sqlmock.NewRows([]string{"name", "last_green_uri", "in_flight_request_ids", "latest_request_id", "version", "last_green_request_id", "enablement", "enablement_version", "effective_from_commit_uri", "enablement_changed_at_ms"}). + AddRow(want.Name, want.LastGreenURI, []byte(`["2","3"]`), want.LatestRequestID, want.Version, want.LastGreenRequestID, want.Enablement, want.EnablementVersion, want.EffectiveFromCommitURI, want.EnablementChangedAtMs) + mock.ExpectQuery("SELECT name, last_green_uri, in_flight_request_ids, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms FROM queue"). WithArgs(want.Name). WillReturnRows(rows) }, @@ -140,7 +139,7 @@ func TestQueueStore_Get(t *testing.T) { name: "not found", queueName: want.Name, setup: func(mock sqlmock.Sqlmock) { - mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms FROM queue"). + mock.ExpectQuery("SELECT name, last_green_uri, in_flight_request_ids, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms FROM queue"). WithArgs(want.Name). WillReturnError(sql.ErrNoRows) }, @@ -151,7 +150,7 @@ func TestQueueStore_Get(t *testing.T) { name: "query error", queueName: want.Name, setup: func(mock sqlmock.Sqlmock) { - mock.ExpectQuery("SELECT name, last_green_uri, in_flight_count, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms FROM queue"). + mock.ExpectQuery("SELECT name, last_green_uri, in_flight_request_ids, latest_request_id, version, last_green_request_id, enablement, enablement_version, effective_from_commit_uri, enablement_changed_at_ms FROM queue"). WithArgs(want.Name). WillReturnError(fmt.Errorf("connection reset")) }, @@ -186,7 +185,7 @@ func TestQueueStore_Update(t *testing.T) { Name: "monorepo/main", LastGreenURI: "git://remote/monorepo/main/green", LastGreenRequestID: "1", - InFlightCount: 1, + InFlightRequestIDs: []string{"2"}, LatestRequestID: "2", } const oldVersion, newVersion = int32(1), int32(2) @@ -201,7 +200,7 @@ func TestQueueStore_Update(t *testing.T) { name: "success", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE queue"). - WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion). + WithArgs(queue.LastGreenURI, []byte(`["2"]`), queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion). WillReturnResult(sqlmock.NewResult(0, 1)) }, }, @@ -209,7 +208,7 @@ func TestQueueStore_Update(t *testing.T) { name: "version mismatch", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE queue"). - WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion). + WithArgs(queue.LastGreenURI, []byte(`["2"]`), queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion). WillReturnResult(sqlmock.NewResult(0, 0)) }, wantErr: true, @@ -219,7 +218,7 @@ func TestQueueStore_Update(t *testing.T) { name: "exec error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE queue"). - WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion). + WithArgs(queue.LastGreenURI, []byte(`["2"]`), queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion). WillReturnError(fmt.Errorf("connection reset")) }, wantErr: true, @@ -228,7 +227,7 @@ func TestQueueStore_Update(t *testing.T) { name: "rows affected error", setup: func(mock sqlmock.Sqlmock) { mock.ExpectExec("UPDATE queue"). - WithArgs(queue.LastGreenURI, queue.InFlightCount, queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion). + WithArgs(queue.LastGreenURI, []byte(`["2"]`), queue.LatestRequestID, newVersion, queue.LastGreenRequestID, queue.Enablement, queue.EnablementVersion, queue.EffectiveFromCommitURI, queue.EnablementChangedAtMs, queue.Name, oldVersion). WillReturnResult(sqlmock.NewErrorResult(fmt.Errorf("driver error"))) }, wantErr: true, diff --git a/stovepipe/extension/storage/mysql/schema/queue.sql b/stovepipe/extension/storage/mysql/schema/queue.sql index 05acbc350..d6d6dbfe3 100644 --- a/stovepipe/extension/storage/mysql/schema/queue.sql +++ b/stovepipe/extension/storage/mysql/schema/queue.sql @@ -1,9 +1,10 @@ -- queue holds per-queue coordination state for the validation pipeline: the last-green --- bookmark, in-flight gate count, and latest-request id pointer. +-- bookmark, the request ids holding build slots, and the latest-request id pointer. CREATE TABLE IF NOT EXISTS queue ( name VARCHAR(255) NOT NULL, last_green_uri VARCHAR(255) NOT NULL DEFAULT '', - in_flight_count INT NOT NULL DEFAULT 0, + in_flight_count INT NOT NULL DEFAULT 0, -- Retained for schema compatibility with existing deployments; active slot ownership uses in_flight_request_ids. + in_flight_request_ids JSON NOT NULL DEFAULT (JSON_ARRAY()), latest_request_id VARCHAR(255) NOT NULL DEFAULT '', version INT NOT NULL, last_green_request_id VARCHAR(255) NOT NULL DEFAULT '', diff --git a/test/e2e/stovepipe/harness_test.go b/test/e2e/stovepipe/harness_test.go index 656cdf79c..d8f49f4e8 100644 --- a/test/e2e/stovepipe/harness_test.go +++ b/test/e2e/stovepipe/harness_test.go @@ -15,6 +15,7 @@ package e2e_test import ( + "encoding/json" "time" "github.com/stretchr/testify/assert" @@ -95,11 +96,13 @@ func (s *StovepipeE2ESuite) awaitRequestState(queue, id, want string) { }) } -func (s *StovepipeE2ESuite) inFlightCount(queue string) int32 { +func (s *StovepipeE2ESuite) inFlightRequestIDs(queue string) []string { s.T().Helper() - var count int32 - require.NoError(s.T(), s.db.QueryRowContext(s.ctx, "SELECT in_flight_count FROM queue WHERE name = ?", queue).Scan(&count)) - return count + var encoded []byte + require.NoError(s.T(), s.db.QueryRowContext(s.ctx, "SELECT in_flight_request_ids FROM queue WHERE name = ?", queue).Scan(&encoded)) + var ids []string + require.NoError(s.T(), json.Unmarshal(encoded, &ids)) + return ids } func (s *StovepipeE2ESuite) awaitBuildStatus(queue, requestID, want string) { diff --git a/test/e2e/stovepipe/ingest_test.go b/test/e2e/stovepipe/ingest_test.go index dce88e080..c9c962be1 100644 --- a/test/e2e/stovepipe/ingest_test.go +++ b/test/e2e/stovepipe/ingest_test.go @@ -89,6 +89,6 @@ func (s *StovepipeE2ESuite) TestIngest_SlowBuild_PollsToCompletion() { require.NotEmpty(s.T(), history.Events) assert.Equal(s.T(), "accepted", history.Events[0].GetRequestState()) assert.Equal(s.T(), history.Events[0].TimestampMs, summary.AcceptedAtMs) - assert.Equal(s.T(), int32(0), s.inFlightCount(queue), + assert.Empty(s.T(), s.inFlightRequestIDs(queue), "a terminal build should release the queue's build slot") } diff --git a/test/e2e/stovepipe/issuetrack_test.go b/test/e2e/stovepipe/issuetrack_test.go index 780e23b02..7e7603833 100644 --- a/test/e2e/stovepipe/issuetrack_test.go +++ b/test/e2e/stovepipe/issuetrack_test.go @@ -80,7 +80,7 @@ func (s *StovepipeE2ESuite) TestIssueTracking_DLQAbandonsPrematureFinalNotificat build := entity.Build{ID: "running-build", RequestID: request.ID, Status: entity.BuildStatusRunning, Version: 1} require.NoError(t, store.GetRequestStore().Create(s.ctx, request)) require.NoError(t, store.GetBuildStore().Create(s.ctx, build)) - require.NoError(t, store.GetQueueStore().Create(s.ctx, entity.Queue{Name: queue, InFlightCount: 1, Version: 1})) + require.NoError(t, store.GetQueueStore().Create(s.ctx, entity.Queue{Name: queue, InFlightRequestIDs: []string{request.ID}, Version: 1})) mq, err := queuemysql.NewQueue(queuemysql.Params{ DB: s.queueDB, Logger: zap.NewNop(), MetricsScope: tally.NoopScope, Tenants: []string{queue}, @@ -112,7 +112,7 @@ func (s *StovepipeE2ESuite) TestIssueTracking_DLQAbandonsPrematureFinalNotificat retainedBuild, err := store.GetBuildStore().Get(s.ctx, build.ID) require.NoError(t, err) require.Equal(t, build, retainedBuild) - require.Equal(t, int32(1), s.inFlightCount(queue)) + require.Equal(t, []string{request.ID}, s.inFlightRequestIDs(queue)) // The DLQ watermark survives message garbage collection and proves recovery acknowledged. pollUntil(convergencePollInterval, func() bool { diff --git a/test/integration/stovepipe/extension/storage/suite.go b/test/integration/stovepipe/extension/storage/suite.go index fec3a3d8b..93dcd2d2e 100644 --- a/test/integration/stovepipe/extension/storage/suite.go +++ b/test/integration/stovepipe/extension/storage/suite.go @@ -90,6 +90,7 @@ func (s *QueueStoreContractSuite) TestQueueStore_CreateWithFields() { Name: name, LastGreenURI: "git://remote/monorepo/main/green-bbbb", LastGreenRequestID: "98", + InFlightRequestIDs: []string{"97", "99"}, LatestRequestID: "99", Version: 1, } @@ -141,7 +142,7 @@ func (s *QueueStoreContractSuite) TestQueueStore_UpdateCAS() { updated.LastGreenURI = "git://remote/monorepo/main/green-cccc" updated.LastGreenRequestID = "41" updated.LatestRequestID = "42" - updated.InFlightCount = 1 + updated.InFlightRequestIDs = []string{"42"} require.NoError(t, s.storeFor(name).Update(s.ctx, updated, 1, 2)) got, err := s.storeFor(name).Get(s.ctx, name) @@ -149,7 +150,7 @@ func (s *QueueStoreContractSuite) TestQueueStore_UpdateCAS() { assert.Equal(t, updated.LastGreenURI, got.LastGreenURI) assert.Equal(t, "41", got.LastGreenRequestID) assert.Equal(t, "42", got.LatestRequestID) - assert.Equal(t, int32(1), got.InFlightCount) + assert.Equal(t, []string{"42"}, got.InFlightRequestIDs) assert.Equal(t, int32(2), got.Version) err = s.storeFor(name).Update(s.ctx, updated, 1, 2) @@ -174,13 +175,13 @@ func (s *QueueStoreContractSuite) TestQueueStore_UpdateSequentialCAS() { v2 := entity.Queue{Name: name, LatestRequestID: "10", Version: 1} require.NoError(t, s.storeFor(name).Update(s.ctx, v2, 1, 2)) - v3 := entity.Queue{Name: name, LatestRequestID: "10", InFlightCount: 1, Version: 2} + v3 := entity.Queue{Name: name, LatestRequestID: "10", InFlightRequestIDs: []string{"10"}, Version: 2} require.NoError(t, s.storeFor(name).Update(s.ctx, v3, 2, 3)) got, err := s.storeFor(name).Get(s.ctx, name) require.NoError(t, err) assert.Equal(t, "10", got.LatestRequestID) - assert.Equal(t, int32(1), got.InFlightCount) + assert.Equal(t, []string{"10"}, got.InFlightRequestIDs) assert.Equal(t, int32(3), got.Version) } @@ -511,11 +512,11 @@ func (s *QueueStoreContractSuite) TestQueueStore_QueueIsolation() { require.NoError(t, err) updatedA := entity.Queue{ - Name: nameA, - LastGreenURI: "git://remote/monorepo/a/green", - LatestRequestID: "7", - InFlightCount: 2, - Version: 1, + Name: nameA, + LastGreenURI: "git://remote/monorepo/a/green", + LatestRequestID: "7", + InFlightRequestIDs: []string{"6", "7"}, + Version: 1, } require.NoError(t, storeA.Update(s.ctx, updatedA, 1, 2))