From 409830cd23c27cfcab44091cbad5f9c865bedad5 Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 10 Jul 2026 14:45:53 +0000 Subject: [PATCH 1/2] feat(stovepipe): coalesce backlog in process controller Compare accepted requests against queue.latest_request_id and mark older heads superseded. Latest heads remain accepted until admit lands in a follow-up PR. --- stovepipe/controller/process/BUILD.bazel | 1 + stovepipe/controller/process/process.go | 130 ++++++++++-- stovepipe/controller/process/process_test.go | 202 +++++++++++++++---- 3 files changed, 276 insertions(+), 57 deletions(-) diff --git a/stovepipe/controller/process/BUILD.bazel b/stovepipe/controller/process/BUILD.bazel index ab92f47d..0299783f 100644 --- a/stovepipe/controller/process/BUILD.bazel +++ b/stovepipe/controller/process/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "//platform/errs:go_default_library", "//platform/metrics:go_default_library", "//stovepipe/core/messagequeue:go_default_library", + "//stovepipe/entity:go_default_library", "//stovepipe/extension/storage:go_default_library", "@com_github_uber_go_tally//:go_default_library", "@org_uber_go_zap//:go_default_library", diff --git a/stovepipe/controller/process/process.go b/stovepipe/controller/process/process.go index 71da8a54..f66b788d 100644 --- a/stovepipe/controller/process/process.go +++ b/stovepipe/controller/process/process.go @@ -12,11 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Package process holds the process-stage queue controller. It consumes the -// request ids ingest publishes, reloads the Request from storage, and (in a -// future change) decides the build strategy by asking SourceControl how the new -// head relates to the queue's last-green URI. For now it is a thin consumer that -// reloads and logs the request, establishing the stage and its wiring. +// Package process holds the process-stage queue controller. It consumes request +// ids from ingest, reloads the Request from storage, coalesces older heads, and +// (in later changes) gates concurrency, decides build strategy, and admits +// winners to build. package process import ( @@ -29,12 +28,13 @@ import ( "github.com/uber/submitqueue/platform/errs" "github.com/uber/submitqueue/platform/metrics" stovepipemq "github.com/uber/submitqueue/stovepipe/core/messagequeue" + "github.com/uber/submitqueue/stovepipe/entity" "github.com/uber/submitqueue/stovepipe/extension/storage" "go.uber.org/zap" ) // Controller consumes ProcessRequest messages from the process stage, reloads the -// referenced Request from storage, and logs it. Implements consumer.Controller. +// referenced Request from storage, and coalesces older heads. Implements consumer.Controller. type Controller struct { logger *zap.SugaredLogger metricsScope tally.Scope @@ -63,10 +63,8 @@ func NewController( } } -// Process reloads the request referenced by the delivery and logs it. Returns nil -// to ack (success) or an error to nack (retry). A not-yet-visible request is -// retryable: ingest persists and publishes, but a stale read may not see the row -// yet, so redelivery converges. +// Process reloads the request referenced by the delivery and coalesces older heads. +// Returns nil to ack (success) or an error to nack (retry). func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (retErr error) { const opName = "process" @@ -82,29 +80,119 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r return fmt.Errorf("failed to deserialize process request: %w", err) } - request, err := c.store.GetRequestStore().Get(ctx, pr.Id) + request, err := c.loadRequest(ctx, pr.Id) if err != nil { metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1) - if errors.Is(err, storage.ErrNotFound) { - // Retryable: the request row may not be visible yet; redelivery converges. - return errs.NewRetryableError(fmt.Errorf("request %s not found yet: %w", pr.Id, err)) + return err + } + + switch request.State { + case entity.RequestStateSuperseded, entity.RequestStateProcessing: + return nil + case entity.RequestStateAccepted: + return c.processAccepted(ctx, request) + default: + c.logger.Infow("ignored request in unexpected state", + "request_id", request.ID, + "queue", request.Queue, + "state", string(request.State), + ) + return nil + } +} + +// processAccepted coalesces older heads against queue.latest_request_id. The latest +// head is left in accepted until admit and the concurrency gate land in later PRs. +func (c *Controller) processAccepted(ctx context.Context, request entity.Request) error { + queueRow, err := c.loadQueue(ctx, request.Queue) + if err != nil { + return err + } + + if queueRow.LatestRequestID == "" { + c.logger.Infow("latest head awaiting admit", + "request_id", request.ID, + "queue", request.Queue, + "uri", request.URI, + ) + return nil + } + + cmp, err := entity.CompareRequestID(request.Queue, request.ID, queueRow.LatestRequestID) + if err != nil { + return fmt.Errorf("ProcessController failed to compare request ids for queue %s: %w", request.Queue, err) + } + if cmp < 0 { + if err := c.supersedeRequest(ctx, request); err != nil { + return err } - return fmt.Errorf("failed to load request %s: %w", pr.Id, err) + c.logger.Infow("superseded request for newer head", + "request_id", request.ID, + "queue", request.Queue, + "latest_request_id", queueRow.LatestRequestID, + ) + return nil } - c.logger.Infow("processing request", + c.logger.Infow("latest head awaiting admit", "request_id", request.ID, "queue", request.Queue, "uri", request.URI, - "state", string(request.State), - "version", request.Version, - "attempt", delivery.Attempt(), - "partition_key", msg.PartitionKey, ) - return nil } +// supersedeRequest CAS-marks request accepted→superseded, retrying on version conflicts. +func (c *Controller) supersedeRequest(ctx context.Context, request entity.Request) error { + reqStore := c.store.GetRequestStore() + + for { + if request.State != entity.RequestStateAccepted { + return nil + } + + updated := request + updated.State = entity.RequestStateSuperseded + newVersion := request.Version + 1 + if err := reqStore.Update(ctx, updated, request.Version, newVersion); err != nil { + if errors.Is(err, storage.ErrVersionMismatch) { + got, getErr := reqStore.Get(ctx, request.ID) + if getErr != nil { + return fmt.Errorf("ProcessController failed to reload request %s after version mismatch: %w", request.ID, getErr) + } + request = got + continue + } + return fmt.Errorf("ProcessController failed to supersede request %s: %w", request.ID, err) + } + return nil + } +} + +// loadRequest returns the request for id. A not-yet-visible row is retryable. +func (c *Controller) loadRequest(ctx context.Context, id string) (entity.Request, error) { + got, err := c.store.GetRequestStore().Get(ctx, id) + if err == nil { + return got, nil + } + if errors.Is(err, storage.ErrNotFound) { + return entity.Request{}, errs.NewRetryableError(fmt.Errorf("request %s not found yet: %w", id, err)) + } + return entity.Request{}, fmt.Errorf("ProcessController failed to load request %s: %w", id, err) +} + +// loadQueue returns the queue row for name. A not-yet-visible row is retryable. +func (c *Controller) loadQueue(ctx context.Context, name string) (entity.Queue, error) { + got, err := c.store.GetQueueStore().Get(ctx, name) + if err == nil { + return got, nil + } + if errors.Is(err, storage.ErrNotFound) { + return entity.Queue{}, errs.NewRetryableError(fmt.Errorf("queue %s not found yet: %w", name, err)) + } + return entity.Queue{}, fmt.Errorf("ProcessController failed to load queue %s: %w", name, err) +} + // Name returns the controller name for logging and metrics. func (c *Controller) Name() string { return "process" diff --git a/stovepipe/controller/process/process_test.go b/stovepipe/controller/process/process_test.go index e9083bbc..77caaa2b 100644 --- a/stovepipe/controller/process/process_test.go +++ b/stovepipe/controller/process/process_test.go @@ -34,22 +34,37 @@ import ( "go.uber.org/zap" ) -const testID = "request/monorepo/main/7" +const ( + testQueue = "monorepo/main" + testID = "request/monorepo/main/7" + testURI = "git://repo/monorepo/main/abc123" +) + +type processMocks struct { + reqStore *storagemock.MockRequestStore + queueStore *storagemock.MockQueueStore +} -func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, *storagemock.MockRequestStore) { +func newController(t *testing.T, ctrl *gomock.Controller) (*Controller, processMocks) { t.Helper() - reqStore := storagemock.NewMockRequestStore(ctrl) + + m := processMocks{ + reqStore: storagemock.NewMockRequestStore(ctrl), + queueStore: storagemock.NewMockQueueStore(ctrl), + } + store := storagemock.NewMockStorage(ctrl) - store.EXPECT().GetRequestStore().Return(reqStore).AnyTimes() + store.EXPECT().GetRequestStore().Return(m.reqStore).AnyTimes() + store.EXPECT().GetQueueStore().Return(m.queueStore).AnyTimes() + c := NewController(zap.NewNop().Sugar(), tally.NewTestScope("test", nil), store, stovepipemq.TopicKeyProcess, "stovepipe-process") - return c, reqStore + return c, m } -// delivery wraps raw payload bytes in a MockDelivery (which satisfies consumer.Delivery). func delivery(t *testing.T, ctrl *gomock.Controller, payload []byte) consumer.Delivery { t.Helper() d := queuemock.NewMockDelivery(ctrl) - d.EXPECT().Message().Return(entityqueue.NewMessage(testID, payload, "monorepo/main", nil)).AnyTimes() + d.EXPECT().Message().Return(entityqueue.NewMessage(testID, payload, testQueue, nil)).AnyTimes() d.EXPECT().Attempt().Return(1).AnyTimes() return d } @@ -61,39 +76,154 @@ func processPayload(t *testing.T, id string) []byte { return b } -func TestProcess_Success(t *testing.T) { - ctrl := gomock.NewController(t) - c, reqStore := newController(t, ctrl) - reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{ID: testID, Queue: "monorepo/main", URI: "git://x", State: entity.RequestStateAccepted, Version: 1}, nil) - - require.NoError(t, c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID)))) +func acceptedRequest(id string) entity.Request { + return entity.Request{ + ID: id, + Queue: testQueue, + URI: testURI, + State: entity.RequestStateAccepted, + Version: 1, + } } -func TestProcess_NotFoundIsRetryable(t *testing.T) { - ctrl := gomock.NewController(t) - c, reqStore := newController(t, ctrl) - reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, storage.ErrNotFound) +func TestProcess(t *testing.T) { + tests := []struct { + name string + id string + setup func(m processMocks) + wantErr bool + wantRetry bool + }{ + { + name: "superseded is no-op", + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{ + ID: testID, Queue: testQueue, State: entity.RequestStateSuperseded, Version: 2, + }, nil) + }, + }, + { + name: "processing is no-op until republish lands", + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{ + ID: testID, Queue: testQueue, State: entity.RequestStateProcessing, Version: 2, + }, nil) + }, + }, + { + name: "latest accepted head awaits admit", + 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) + }, + }, + { + name: "accepted with empty latest pointer awaits admit", + 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, + Version: 1, + }, nil) + }, + }, + { + name: "older accepted head is superseded", + id: "request/monorepo/main/3", + setup: func(m processMocks) { + olderID := "request/monorepo/main/3" + m.reqStore.EXPECT().Get(gomock.Any(), olderID).Return(acceptedRequest(olderID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + Version: 1, + }, nil) + updated := acceptedRequest(olderID) + updated.State = entity.RequestStateSuperseded + m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(nil) + }, + }, + { + name: "supersede retries on version mismatch", + id: "request/monorepo/main/3", + setup: func(m processMocks) { + olderID := "request/monorepo/main/3" + m.reqStore.EXPECT().Get(gomock.Any(), olderID).Return(acceptedRequest(olderID), nil) + m.queueStore.EXPECT().Get(gomock.Any(), testQueue).Return(entity.Queue{ + Name: testQueue, + LatestRequestID: testID, + Version: 1, + }, nil) + updated := acceptedRequest(olderID) + updated.State = entity.RequestStateSuperseded + m.reqStore.EXPECT().Update(gomock.Any(), updated, int32(1), int32(2)).Return(storage.ErrVersionMismatch) + m.reqStore.EXPECT().Get(gomock.Any(), olderID).Return(entity.Request{ + ID: olderID, Queue: testQueue, State: entity.RequestStateSuperseded, Version: 2, + }, nil) + }, + }, + { + name: "request not found is retryable", + wantErr: true, + wantRetry: true, + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, storage.ErrNotFound) + }, + }, + { + name: "queue not found is retryable", + wantErr: true, + wantRetry: true, + 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{}, storage.ErrNotFound) + }, + }, + { + name: "request storage error is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m processMocks) { + m.reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, errors.New("db down")) + }, + }, + { + name: "malformed payload is not retryable", + wantErr: true, + wantRetry: false, + setup: func(m processMocks) {}, + }, + } - err := c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID))) - require.Error(t, err) - assert.True(t, errs.IsRetryable(err), "a not-yet-visible request must be retryable") -} + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctrl := gomock.NewController(t) + c, m := newController(t, ctrl) + if tt.setup != nil { + tt.setup(m) + } -func TestProcess_StorageErrorNotRetryable(t *testing.T) { - ctrl := gomock.NewController(t) - c, reqStore := newController(t, ctrl) - reqStore.EXPECT().Get(gomock.Any(), testID).Return(entity.Request{}, errors.New("db down")) - - err := c.Process(context.Background(), delivery(t, ctrl, processPayload(t, testID))) - require.Error(t, err) - assert.False(t, errs.IsRetryable(err)) -} + id := testID + if tt.id != "" { + id = tt.id + } + payload := processPayload(t, id) + if tt.name == "malformed payload is not retryable" { + payload = []byte("not-json") + } -func TestProcess_MalformedPayload(t *testing.T) { - ctrl := gomock.NewController(t) - c, _ := newController(t, ctrl) + err := c.Process(context.Background(), delivery(t, ctrl, payload)) - err := c.Process(context.Background(), delivery(t, ctrl, []byte("not-json"))) - require.Error(t, err) - assert.False(t, errs.IsRetryable(err)) + if tt.wantErr { + require.Error(t, err) + assert.Equal(t, tt.wantRetry, errs.IsRetryable(err)) + return + } + require.NoError(t, err) + }) + } } From 700303522898d590e6dbed8bd913b97d31fc190c Mon Sep 17 00:00:00 2001 From: mnoah1 Date: Fri, 10 Jul 2026 15:01:10 +0000 Subject: [PATCH 2/2] docs(stovepipe): align process RFC with latest_request_id Update process.md to describe coalescing via CompareRequestID and queue.latest_request_id instead of the dropped Sequence field and latest_request_seq pointer. Also align BuildStrategy naming with code. --- doc/rfc/stovepipe/steps/process.md | 66 +++++++++++++++--------------- 1 file changed, 33 insertions(+), 33 deletions(-) diff --git a/doc/rfc/stovepipe/steps/process.md b/doc/rfc/stovepipe/steps/process.md index 315e8350..e45b97c0 100644 --- a/doc/rfc/stovepipe/steps/process.md +++ b/doc/rfc/stovepipe/steps/process.md @@ -9,7 +9,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. -2. **`process` is not the only Queue-row writer.** `ingest` stamps `latest_request_seq` (an integer pointer, not a URI; see [Backlog coalescing](#backlog-coalescing)) and `record` advances the `last_green_uri` bookmark and releases slots. They touch different fields under optimistic-locking CAS, so concurrent writes converge instead of clobbering. +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 `record` advances the `last_green_uri` bookmark and releases slots. They touch different fields under optimistic-locking CAS, so concurrent writes converge instead of clobbering. ## Process algorithm @@ -22,8 +22,8 @@ For a delivery carrying request id `R`: - ack and return (idempotent no-op). 3. If R.State is processing (strategy already recorded): - re-publish R to build (the prior publish may have failed), ack, return. -4. R.State is accepted. Load (or create) the Queue row Q. -5. Coalesce: if R.Sequence < Q.latest_request_seq: +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): - defer (Option 1 or Option 2 below) -> re-check until the slot frees (admit) or a newer head supersedes it. See [Waiting for a slot](#waiting-for-a-slot). @@ -41,11 +41,11 @@ Step 5 runs regardless of the gate: an intermediate head is superseded on sight To admit a Request (step 7a), `process` reads the Queue's **last-green URI** and asks the same `SourceControl` implementation ingest uses: -1. **No** last-green URI (cold start) → **full-monorepo**, no baseline. +1. **No** last-green URI (cold start) → **full**, no baseline. 2. Last-green URI present → call `SourceControl.IsAncestor(lastGreen, R.URI)`: - **true** — fast-forward: **incremental-since-green**, baseline = last-green URI; `build` validates only the delta. - - **false** — history rewrite (force-push, rebase): **full-monorepo**, no baseline. The stale bookmark must not be a build baseline. - - **`ErrNotFound`** — a URI isn't on the ref (ancestry uncertainty). A full build is always a valid superset, so fall back to **full-monorepo** and log a warning; not a retryable error. + - **false** — history rewrite (force-push, rebase): **full**, no baseline. The stale bookmark must not be a build baseline. + - **`ErrNotFound`** — a URI isn't on the ref (ancestry uncertainty). A full build is always a valid superset, so fall back to **full** and log a warning; not a retryable error. - **any other error** (connection/timeout) — return it raw; the consumer's classifier decides retryability (see [errs](../../../../platform/errs/README.md)). Strategy and baseline are persisted on the Request and are **immutable**: a redelivery sees `processing` at step 3 and never re-derives them. `SourceControl` owns ancestry; `process` never parses a URI. @@ -79,30 +79,30 @@ The scheme: each admit pins its baseline to last-green *at admit time*; a green Correctness rests on four rules, all with MVP primitives already in place: 1. **Pin to last-green at admit** — a known-green baseline, which `process` already reads. -2. **Advance last-green by sequence, monotonically** — `record` adopts the highest-`Sequence` green and never regresses. (Free at N=1, since verdicts land in order; explicit at N > 1.) +2. **Advance last-green by ingest order, monotonically** — `record` adopts the highest-counter green (per `CompareRequestID`) and never regresses. (Free at N=1, since verdicts land in order; explicit at N > 1.) 3. **Coalesce to the latest N** instead of the latest 1 — otherwise intermediates are superseded on sight and nothing overlaps. 4. **Fall back to serial on a rewrite** — the linear/superset property is what makes "adopt highest green" sound; rewrites already force a full build with no baseline. The only cost is speculation: a build on an older baseline re-tests deltas a concurrent build already greened past — correct but wasteful, growing with how far the baseline lags. Bounding that lag ("drain before adopting a new baseline") is a **cost governor, not a safety gate**. It inherits, but doesn't worsen, the incremental-build soundness assumption already used at N=1. -So per-baseline concurrency isn't an unsolved semantics problem — it's speculative validation with a lag-bounded baseline. It's deferred only for the per-lineage bookkeeping (rules 2–4, derivable from `Request.Sequence` + `Request.BaseURI`) and a coalesce-latest-N policy, neither of which the MVP forecloses. +So per-baseline concurrency isn't an unsolved semantics problem — it's speculative validation with a lag-bounded baseline. It's deferred only for the per-lineage bookkeeping (rules 2–4, derivable from request-id ingest order + `Request.BaseURI`) and a coalesce-latest-N policy, neither of which the MVP forecloses. ## Backlog coalescing While a validation runs, ingest keeps admitting newer heads (distinct Requests, deduped on `(Queue, URI)`). Only the **latest** is worth validating; intermediates are skipped. -"Latest" is the **monotonic request sequence**, not VCS history: +"Latest" is the **monotonic ingest order**, not VCS history: -- Ingest mints ids from a per-Queue counter (`request//`) and stores `n` as `Request.Sequence` (higher = later ingest = newer head). -- Ingest also stamps `Queue.latest_request_seq = max(current, n)` on accept — the **out-of-band latest pointer** (an integer, distinct from `last_green_uri`) that makes coalescing a single-row integer comparison. +- Ingest mints ids from a per-Queue counter (`request//`). The counter suffix is the order key (higher = later ingest = newer head); there is no separate `Sequence` field on `Request`. +- Ingest also CASes `Queue.latest_request_id` to the accepted request's id when it is newer (via `CompareRequestID`) — the **out-of-band latest pointer** (a request id, distinct from `last_green_uri`) that makes coalescing a single-row comparison. Why not `SourceControl.History`: a history walk is expensive, and after a rewrite the superseded URIs may be off-ref entirely — exactly where history order is meaningless. Ingest order authoritatively says which head we learned of last. -**Why not `SourceControl.Latest`** (rejected): comparing `R.URI` to the live ref head leaks work outside the ingested set. The ref moves whenever a commit lands — independent of what the poller reported or ingest minted — so whenever it is ahead of the newest Request, *every* Request differs from `Latest` and is superseded, admitting nothing and chasing a commit `process` was never handed. Invariant: **`process` only validates a commit ingest identified.** The sequence is defined over exactly that set; `Latest` isn't. Nor can `counter.Peek` stand in: ingest spends a counter value on a dedup race-loss, so the highest *minted* value can exceed the highest *real* Request's — an `== latest` test would never match. Stamping `latest_request_seq` only after a successful create keeps it aligned. +**Why not `SourceControl.Latest`** (rejected): comparing `R.URI` to the live ref head leaks work outside the ingested set. The ref moves whenever a commit lands — independent of what the poller reported or ingest minted — so whenever it is ahead of the newest Request, *every* Request differs from `Latest` and is superseded, admitting nothing and chasing a commit `process` was never handed. Invariant: **`process` only validates a commit ingest identified.** Ingest order is defined over exactly that set; `Latest` isn't. Nor can `counter.Peek` stand in: ingest spends a counter value on a dedup race-loss, so the highest *minted* counter can exceed the highest *real* Request's — an equality test against the pointer would never match. Stamping `latest_request_id` only after a successful request create keeps it aligned. Ordering caveat: `counter.Next` doesn't guarantee assignment order, so under concurrent same-Queue ingest (rare — one serial poller) "highest sequence" may not equal "most recently reported". They agree in practice, and a rare inversion self-corrects next poll. Acceptable for MVP. -**The pointer prevents deadlock.** Under `BatchSize = 1`, Option 1 blocks its partition while waiting, so `process` can't learn of newer heads *from the stream* during that wait. Option 2 unblocks the partition, so newer ingest deliveries can arrive immediately. Both options read `latest_request_seq` from the Queue row on every wake-up (step 5), so a stale waiter — blocked or delayed — still supersedes correctly. Ingest stamps the pointer independently of the partition (see [Backlog coalescing](#backlog-coalescing)). +**The pointer prevents deadlock.** Under `BatchSize = 1`, Option 1 blocks its partition while waiting, so `process` can't learn of newer heads *from the stream* during that wait. Option 2 unblocks the partition, so newer ingest deliveries can arrive immediately. Both options read `latest_request_id` from the Queue row on every wake-up (step 5), so a stale waiter — blocked or delayed — still supersedes correctly. Ingest stamps the pointer independently of the partition (see [Backlog coalescing](#backlog-coalescing)). **Progress (no starvation).** Superseding is always forward motion toward the newest head, and the newest head is never superseded (nothing is newer). So as long as `process` supersedes faster than ingest adds heads — it does, since superseding is a CAS + ack with no build, far cheaper than the poll cadence — a build always starts; a high commit rate just coalesces more intermediates away. @@ -123,11 +123,11 @@ 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`. -2. While A runs, **B**, **C**, **D** are ingested (`latest_request_seq = D.seq`). -3. B: `B.seq < D.seq` → **superseded** (acked), though A is still in flight. Same for **C**. D is latest but the gate is closed → **waits for slot** (Option 1 or 2). +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** (Option 1 or 2). 4. A's build finishes → `record` records A's greenness, `in_flight_count → 0`. 5. D's re-check → gate open, D still latest → **D admitted**, published to `build`. -6. While D runs, **E**, **F** ingested (`latest_request_seq = F.seq`). E superseded on sight; F waits for slot. +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**. A, D, F each get a full cycle; B, C, E end `superseded`. No intermediate is validated individually — intentional for MVP. @@ -176,7 +176,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 | record | | `in_flight_count` | Active Phase 1 validations | process (+1), record/DLQ (−1) | -| `latest_request_seq` | Highest request sequence ingested | ingest | +| `latest_request_id` | Request id of the newest head ingest accepted | ingest | | `version` | Optimistic-locking version | all writers | Per-queue knobs such as `max_concurrent` live outside this row — see [Per-Queue concurrency gate](#per-queue-concurrency-gate). @@ -185,8 +185,8 @@ Per-queue knobs such as `max_concurrent` live outside this row — see [Per-Queu | Field | Role | |---|---| -| `Sequence` | The per-Queue counter value `n` behind the id; the coalescing order key | -| `BuildStrategy` | `incremental_since_green` \| `full_monorepo`; immutable once set by `process` | +| `ID` | Globally unique id (`request//`); the counter suffix is the ingest-order key for coalescing | +| `BuildStrategy` | `incremental_since_green` \| `full`; immutable once set by `process` | | `BaseURI` | Last-green URI used as the incremental base; empty for full builds | **States** (extending today's `accepted`-only machine): @@ -205,10 +205,10 @@ 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_seq`; `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_count`; `record` CASes `last_green_uri` + `in_flight_count`. - **`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_seq` pointer instead, keeping the contract satisfiable by a plain KV backend. +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 @@ -219,7 +219,7 @@ When the gate is closed, `process` must defer the latest head without admitting Both re-run the same **coalesce-then-gate** checks on every wake-up (steps 5 → 6): -1. **Stale? (checked first.)** If `R.Sequence < latest_request_seq`, `R` is no longer latest → supersede it (ack). A newer head is admitted by its own delivery when its slot attempt runs. +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). Neither option admits to `build` until the gate opens. @@ -228,19 +228,19 @@ Neither option admits to `build` until the gate opens. **Mechanism.** Keep the in-flight delivery alive. Loop: call `ExtendVisibilityTimeout` on an interval (renews the lease **without** incrementing `retry_count`), reload the Queue row, run coalesce-then-gate. Never ack or nack while waiting. Honor context cancellation — on shutdown return promptly; the head resumes on redelivery. -**Partition behavior.** Under `BatchSize = 1`, the delivery stays in-flight and **blocks the partition** until it admits or supersedes. Newer ingest messages queue behind it in the log; coalescing for the waiting head relies on `latest_request_seq` from the Queue row, not on newer deliveries arriving while blocked. +**Partition behavior.** Under `BatchSize = 1`, the delivery stays in-flight and **blocks the partition** until it admits or supersedes. Newer ingest messages queue behind it in the log; coalescing for the waiting head relies on `latest_request_id` from the Queue row, not on newer deliveries arriving while blocked. **Walkthrough** — Queue `monorepo/main`, `max_concurrent = 1`, heads A→F, Option 1 chosen: 1. **A** admitted, published to `build` (`in_flight_count = 1`). `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_seq = D.seq`). D's delivery: latest, gate closed → **park** (extend loop begins; partition blocked). -4. While D waits, **E**, **F** ingested (`latest_request_seq = F.seq`). Their process messages sit in the log behind D's blocked delivery — they do not run yet. -5. D's renew loop re-reads the Queue row → `D.seq < F.seq` → **supersede D**, ack (partition unblocks). +3. **D** ingested (`latest_request_id = D.id`). D's delivery: latest, gate closed → **park** (extend loop begins; partition blocked). +4. While D waits, **E**, **F** ingested (`latest_request_id = F.id`). Their process messages sit in the log behind D's blocked delivery — they do not run yet. +5. D's renew loop re-reads the Queue row → D is older than F → **supersede D**, ack (partition unblocks). 6. **E**'s delivery runs → superseded. **F**'s delivery runs → latest, gate still closed → **park**. 7. A completes at `record` → `in_flight_count → 0`. F's renew loop sees gate open → **admit F**, publish to `build`, ack. -**Supersede reasoning.** Simple while blocked: the waiter periodically re-checks `latest_request_seq` and supersedes itself when a newer head appears. Intermediates (B, C, E) only run once the partition unblocks enough to reach their offsets. +**Supersede reasoning.** Simple while blocked: the waiter periodically re-reads `latest_request_id` and supersedes itself when a newer head appears. Intermediates (B, C, E) only run once the partition unblocks enough to reach their offsets. **Tradeoffs.** @@ -257,15 +257,15 @@ Neither option admits to `build` until the gate opens. **Mechanism.** On gate closed and still latest: **ack** the current delivery, then **`PublishAfter`** the same `ProcessRequest` to the process topic (same partition key = queue name) after a short delay. Each wake-up is a **fresh** message (`retry_count` starts at 0). Run coalesce-then-gate at the top of every wake-up; if still latest and gate still closed, ack and `PublishAfter` again. Only reschedule when both conditions hold — otherwise supersede or admit immediately. -**Partition behavior.** Acks free the partition. Newer ingest deliveries (E, F, …) are processed and superseded while the latest head waits on a timer. Deferred rows use `visible_after` and are skipped until due — the same non-blocking property as a nacked message — so a reschedule at offset 11 can run *after* a newer ingest message at offset 12. Delivery order reshuffles relative to strict log order; coalescing keys off `latest_request_seq`, not offset. +**Partition behavior.** Acks free the partition. Newer ingest deliveries (E, F, …) are processed and superseded while the latest head waits on a timer. Deferred rows use `visible_after` and are skipped until due — the same non-blocking property as a nacked message — so a reschedule at offset 11 can run *after* a newer ingest message at offset 12. Delivery order reshuffles relative to strict log order; coalescing keys off `latest_request_id`, not offset. **Walkthrough** — same scenario, Option 2 chosen: 1. **A** admitted, published to `build` (`in_flight_count = 1`). 2. **B**, **C** ingested → delivered and **superseded** on sight. -3. **D** ingested (`latest_request_seq = D.seq`). D's delivery: latest, gate closed → **ack + `PublishAfter(D, delay)`**. Partition free. -4. **E**, **F** ingested (`latest_request_seq = F.seq`). **E** delivered → superseded. **F** delivered → latest, gate closed → **ack + `PublishAfter(F, delay)`**. D's pending reschedule is now redundant. -5. D's timer fires → `D.seq < F.seq` → **supersede D**, ack (cheap no-op). +3. **D** ingested (`latest_request_id = D.id`). D's delivery: latest, gate closed → **ack + `PublishAfter(D, delay)`**. Partition free. +4. **E**, **F** ingested (`latest_request_id = F.id`). **E** delivered → superseded. **F** delivered → latest, gate closed → **ack + `PublishAfter(F, delay)`**. D's pending reschedule is now redundant. +5. D's timer fires → D is older than F → **supersede D**, ack (cheap no-op). 6. A completes at `record` → `in_flight_count → 0`. F's timer fires → gate open, still latest → **admit F**, publish to `build`, ack. **Supersede reasoning.** Straightforward: every wake-up (immediate or delayed) runs step 5 first. Stale delayed messages (D) supersede on sight. Only the current latest (F) should schedule the next wait. Older delayed rows are expected no-ops, not errors. @@ -285,7 +285,7 @@ Neither option admits to `build` until the gate opens. |---|---|---| | **`MaxAttempts`** | Safe when renewals keep up; lease lapses increment `retry_count` | Safe — waiting never increments `retry_count` | | **Partition (`BatchSize = 1`)** | Blocks until admit or supersede | Unblocks; newer heads process immediately | -| **Supersede** | Waiter polls `latest_request_seq` in loop | Immediate deliveries + stale timers supersede on wake-up | +| **Supersede** | Waiter polls `latest_request_id` in loop | Immediate deliveries + stale timers supersede on wake-up | | **Churn** | One delivery, periodic extends | One new log row per wait cycle | | **Wiring** | `ExtendVisibilityTimeout` on `Delivery` | Publisher + topic registry in controller | | **Worker** | Goroutine in renew loop | Returns; timer brings work back | @@ -294,6 +294,6 @@ Implementation picks one option and wires step 6 to it. A future `consumer.ErrHo ## Batch consume -Coalescing uses the sequence pointer one delivery at a time. Intermediates are each delivered once and superseded. The latest head adds no extra rows under Option 1 (it waits in place); under Option 2 it adds one reschedule row per wait cycle while the gate is closed. +Coalescing uses the latest-request pointer one delivery at a time. Intermediates are each delivered once and superseded. The latest head adds no extra rows under Option 1 (it waits in place); under Option 2 it adds one reschedule row per wait cycle while the gate is closed. If that churn matters at scale, an optional `BatchController` (receiving `[]Delivery` per poll) would let `process` supersede all intermediates in a single tick — an optimization over the single-delivery path that can land later without changing the state machine or storage contract.