Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions doc/rfc/consumer-gate.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ The pipeline is a set of queue controllers spread across three services, and det
- **E2e scenario control.** A test that must interleave two in-flight messages ("the batch controller must not consume its message before the cancel controller has finished") needs to halt exactly one controller while its siblings keep running. Stopping the whole service is too coarse: it kills every controller in the process, re-assigns ephemeral ports, and turns a scenario step into a container lifecycle event.
- **Single-host development.** A developer reproducing an ordering-sensitive scenario locally needs the same stop/observe/start control without restarting the whole stack.

Nothing in the system expresses this today. The queue can be manipulated from outside (e.g. starving a consumer by occupying its partition leases), but data-plane tricks of that kind are structurally limited: they only work if arranged before the controller ever touches the partition, they couple the caller to one backend's internals, and they are invisible to the service. This RFC makes stop/observe/start a small, coherent mechanism for tests and local development instead. Fleet-wide production pause is explicitly out of scope for the file implementation.
Nothing in the system expresses this today. The queue can be manipulated from outside (e.g. starving a consumer by occupying its partition leases), but data-plane tricks of that kind are structurally limited: they only work if arranged before the controller ever touches the partition, they couple the caller to one backend's internals, and they are invisible to the service. This RFC makes stop/observe/start a small, coherent mechanism for tests and local development instead. The file implementation is not a fleet-wide production control plane. Pausing a queue in production is a separate mechanism that wraps the gate with pause state read from its own source; see [Pausing and Unpausing a Queue](submitqueue/queue-pause-and-unpause.md).

## Decisions

Expand All @@ -27,13 +27,13 @@ The re-check loop has a bounded cost: while a gate is closed, each blocked parti

### Gate identity: consumer group, optionally narrowed to a partition

Every controller subscribes with a unique consumer group (`orchestrator-batch`, `runway-merge`, …), so the consumer group *is* the controller's stable runtime name — the natural key for "stop this controller". A gate may optionally carry a partition key; absent one, it gates every partition. Partition-scoped gates keep unrelated traffic flowing through the same controller (e.g. parking one test queue's partition while other queues proceed), which matters if e2e scenarios ever run concurrently.
A controller's consumer group is its stable runtime name — the natural key for "stop this controller". Groups are not unique per controller everywhere: Runway's controllers each use their own (for example `runway-mergeconflictcheck`), but the orchestrator's pipeline stages share one group, `orchestrator`, so a gate on that group covers every stage. The key carries no topic, so a gate cannot single out one orchestrator stage. A gate may optionally carry a partition key; absent one, it gates every partition. Partition-scoped gates keep unrelated traffic flowing through the same controller (e.g. parking one test queue's partition while other queues proceed), which matters if e2e scenarios ever run concurrently.

### Gate state is a separate extension

The consumer gate is a shared extension in its own right, not a feature of any queue backend. The contract lives at `platform/extension/consumergate/`: the behavioral interface the consumer reads and the write surface tests and tooling use. `Park` accepts a caller-owned `DeliveryDescriptor` containing only message data; the implementation combines it with the gate identity captured by `Enter` and its own timestamp to create the observable `Parked` record, so callers cannot supply or overwrite gate-owned fields, and `Unpark` removes the record on the admit path. The consumer package takes the read-side interface as a dependency; wiring passes the file implementation only when `CONSUMER_GATE_DIR` is explicitly configured and otherwise passes the no-op implementation.

Keeping the contract separate from any backend is what lets the storage medium be chosen per deployment: a filesystem directory first (below), a database- or config-service-backed implementation later if fleet-wide coordination demands it — with the middleware, the wiring shape, and every test written against the contract unchanged.
Keeping the contract separate from any backend is what lets the storage medium be chosen per deployment: a filesystem directory first (below), another backend later if a deployment needs one — with the middleware, the wiring shape, and every test written against the contract unchanged.

### E2E and local implementation: files in a shared directory

Expand Down Expand Up @@ -87,7 +87,7 @@ The two-controller interleaving scenario is the same shape: close controller X's

- **Starving a controller through the queue's data plane** (e.g. occupying its partition leases from outside the service). Needs zero service changes, which is why it was the harness's first candidate, but it is pre-hold-only (an actively consuming controller cannot be stopped), coupled to one backend's scheduling internals, and invisible to the service — no observation, no metrics, nothing reusable by an operator. Once service modifications are on the table, the middleware dominates it.
- **Parking deliveries in memory** (block the partition goroutine on a blocked delivery, extending its visibility until the gate opens — this design's original mechanism). It holds a goroutine and a lease per blocked partition, and it only babysits the delivery it parks: messages already fetched into the per-partition buffer behind it (up to the batch size) have their visibility lapse, redeliver as duplicates that stuff the buffer until the topic's routing loop stalls, and burn a retry attempt per lapse until they are spuriously dead-lettered without ever failing. Postponing each blocked delivery back to the queue removes the held state entirely and inherits hold's retry exemption; the costs are a ~1s release quantization and the released message being a fresh attempt rather than the same one.
- **A database-backed store in this E2E-focused change.** A shared store is the appropriate direction if fleet-wide production pause is required, but it needs its own availability, caching, and administration design. It should be added as another `consumergate` implementation rather than coupled to a queue backend.
- **A database-backed store in this E2E-focused change.** Production pause does not need one: queue pause wraps this gate with state read from its own store ([Pausing and Unpausing a Queue](submitqueue/queue-pause-and-unpause.md)). A shared database-backed `consumergate` implementation would still need its own availability, caching, and administration design, and belongs in a separate change rather than coupled to a queue backend.
- **Admin RPC on each service for pause/resume.** Reaches the same middleware, but costs a new proto surface, port, and auth story on three services. The shared test directory already supplies the out-of-process control required here.
- **Config/env-driven controller enablement plus restart.** Restart granularity is the whole process — it bounces every sibling controller and disturbs in-flight leases, destroying exactly the "others keep running" property mid-scenario. A static topology tool, not a stop/start lever.
- **Fail closed on gate-state errors.** Turns an auxiliary medium's failure into a full consumption stall across every gated consumer. The gate exists to *add* control, not to add a new way for the pipeline to stop on its own.
Expand Down
2 changes: 1 addition & 1 deletion doc/rfc/index.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ Design documents and technical proposals, grouped by scope. Shared/cross-cutting
- [Gateway History APIs](submitqueue/history-api.md) - Request lifecycle history exposed through separate request ID and change ID endpoints
- [Build Runner](submitqueue/build-runner.md) - Vendor-agnostic BuildRunner interface, provider-neutral BuildStatus lifecycle, and how the orchestrator wires it into the build stage
- [Extension Contract](submitqueue/extension-contract.md) - When extensions take orchestrator identity (request/batch) and resolve granular content themselves vs. take controller-resolved data; revises the BuildRunner base/head contract
- [Pausing and Unpausing a Queue](submitqueue/queue-pause-and-unpause.md) - Runtime pause and unpause of a queue as two switches (reject new lands, freeze processing), including unpause and cancel behavior; compares keeping the state on queue config against a separate pause-state extension, with open questions on freeze failure posture and cancel during freeze
- [Pausing and Unpausing a Queue](submitqueue/queue-pause-and-unpause.md) - Runtime pause and unpause of a queue as two switches (reject new lands, freeze processing), including unpause and cancel behavior; the separate pause-state extension, fail-closed enforcement, requirements for implementations, and the rejected queue-config alternative, with open questions on cancel during freeze
- [Gateway Status and List APIs](submitqueue/status-list-api.md) - Gateway-owned request context, materialized current status, sqid or change-URI status lookup, and queue admission listing
- [Speculation](submitqueue/speculation.md) - Why SubmitQueue speculates, the path/tree model, and the two pluggable seams: speculation-tree enumeration and path selection
- [Outcome Scorer](submitqueue/outcome-scorer.md) - How likely a batch is to reach Succeeded: `Score(ctx, batch, paths)` as a logit-linear model — a base content price plus YAML weights on path and batch evidence (`pathPassed`, `pathFailed`, `landing`, `cancelling`)
Expand Down
Loading
Loading