From aeb04857bde9e8611928c9d1fdb755465d5060ea Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 19:41:45 +0000 Subject: [PATCH 1/2] docs(consumer-gate): Correct consumer group naming and point to queue pause --- doc/rfc/consumer-gate.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/doc/rfc/consumer-gate.md b/doc/rfc/consumer-gate.md index 45d9d7452..50e01a053 100644 --- a/doc/rfc/consumer-gate.md +++ b/doc/rfc/consumer-gate.md @@ -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 @@ -27,7 +27,7 @@ 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 From 503c039e13fa19f56b461f6a0ef77083145bc368 Mon Sep 17 00:00:00 2001 From: manjari Date: Wed, 14 Oct 2026 20:55:05 +0000 Subject: [PATCH 2/2] docs(rfc): Fold the queue pause decision into the RFC --- doc/rfc/consumer-gate.md | 4 +- doc/rfc/index.md | 2 +- .../submitqueue/queue-pause-and-unpause.md | 152 ++++-------------- 3 files changed, 35 insertions(+), 123 deletions(-) diff --git a/doc/rfc/consumer-gate.md b/doc/rfc/consumer-gate.md index 50e01a053..27df48960 100644 --- a/doc/rfc/consumer-gate.md +++ b/doc/rfc/consumer-gate.md @@ -33,7 +33,7 @@ A controller's consumer group is its stable runtime name — the natural key for 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 @@ -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. diff --git a/doc/rfc/index.md b/doc/rfc/index.md index ee725c7c3..24f07fbc4 100644 --- a/doc/rfc/index.md +++ b/doc/rfc/index.md @@ -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`) diff --git a/doc/rfc/submitqueue/queue-pause-and-unpause.md b/doc/rfc/submitqueue/queue-pause-and-unpause.md index 0f724ee3c..d5910d297 100644 --- a/doc/rfc/submitqueue/queue-pause-and-unpause.md +++ b/doc/rfc/submitqueue/queue-pause-and-unpause.md @@ -1,11 +1,11 @@ # Pausing and Unpausing a Queue -Operators need to stop a queue during an incident, a repository freeze, or maintenance, and resume it later. We need to pause and unpause a queue at runtime, without redeploying the gateway or orchestrator. This RFC lays out 2 options for where the pause state should live. +Operators need to stop a queue during an incident, a repository freeze, or maintenance, and resume it later, at runtime and without redeploying the gateway or orchestrator. The pause state lives in its own extension, `queuepause`, and OSS code enforces it, so every deployment runs the same rejection and freezing logic and a deployer only supplies where the state comes from. ## Behavior ### Pause -Pause is two independent switches per queue. +Pause is two independent switches per queue, plus an optional free-text reason. | Reject new lands | Freeze processing | Effect | |---|---|---| @@ -14,15 +14,15 @@ Pause is two independent switches per queue. | off | on | Hold: requests are accepted and durable, nothing advances | | on | on | Full pause | -All four combinations are meaningful, so two booleans are enough and no mode enum is needed. Both options share the following decisions: +All four combinations are meaningful, so two booleans are enough and no mode enum is needed. -- **Reject** happens in `Land()` before a request ID is minted, so a refused request consumes no sequence number and writes nothing. It returns a distinct user error, so callers can tell a paused queue from a malformed request or an unknown queue. -- **Freeze** is a live-read `consumergate.Gate` implementation that closes the gate for every orchestrator stage of the paused queue, including the dead-letter consumers, so no processing happens while frozen. The primary and DLQ consumers share one gate, so a gate that matches on queue alone covers both. The gate finds a message's queue from its tenant, which every orchestrator and gateway publish site sets to the queue name. It is a barrier, not preemption: a message already inside a controller runs to completion ([consumer-gate.md](../consumer-gate.md)). Blocked deliveries are postponed, so they spend no retry budget ([consumer-hold.md](../consumer-hold.md)). Unpausing releases within one re-check delay plus the queue poll interval. -- **Defaults**: when nothing is configured, no queue is ever paused. OSS ships a working way to pause; a deployer can back the state with its own config service instead. -- Only `Land()` is gated at the gateway. Status, list, and history reads are unaffected. +- **Reject** happens in `Land()` before a request ID is minted, so a refused request consumes no sequence number and writes nothing. It returns a distinct user error that carries the queue and the operator's reason, and the gateway maps it to `FailedPrecondition`. A caller can tell a paused queue from a malformed request (`InvalidArgument`) or an unknown queue. The pause check runs after the queue is recognized, so an unknown queue is reported as unknown. +- **Freeze** is a live-read `consumergate.Gate` that closes for every orchestrator stage of the paused queue, including the dead-letter consumers, so no processing happens while frozen. The primary and DLQ consumers share one gate, so a gate that matches on queue alone covers both. The gate finds a message's queue from its tenant, which every orchestrator and gateway publish site sets to the queue name. It is a barrier, not preemption: a message already inside a controller runs to completion ([consumer-gate.md](../consumer-gate.md)). Blocked deliveries are postponed, so they spend no retry budget ([consumer-hold.md](../consumer-hold.md)). +- **Both fail closed.** When pause state cannot be read, `Land()` returns the error and the freeze gate stays closed. The consumer treats a gate error as open, so the freeze gate returns a blocked entry on a read error instead of the error. The one exception is a cancelled context, which is passed through so a shutting-down consumer is not logged as a pause backend failure. +- **Defaults**: when nothing is configured, no queue is ever paused. OSS ships a working way to pause, and a deployer can back the state with its own config service. +- Only `Land()` is gated at the gateway. Status, list, and history reads are unaffected. The gateway's own request-log consumer is not frozen, so request status and history keep materializing while a queue is frozen. - All orchestrator stages share the consumer group `orchestrator`, so freeze covers a queue's whole pipeline. Freezing a single stage is out of scope. - ### Unpause - **Reject** stops applying on the first `Land()` after the store reflects the change. There is nothing to drain or replay. @@ -42,131 +42,43 @@ The consequence is that freeze, cancel, then unfreeze does not guarantee the can Supporting cancel during a freeze needs the topic on the gate key and exemptions for cancel, speculate, and build signal. Freeze would then mean no new effects rather than nothing runs. -## Problem -- `queueconfig.Store` is the registry of valid queue names. The YAML implementation is read once at startup and the entity is documented as immutable. -- `consumergate` can stop a controller at runtime, but its only backend is a shared directory of files meant for e2e tests and single-host development. It has no notion of a queue: it is keyed by consumer group and partition. It fails open when its state cannot be read. -- The gateway has no pause check in `Land()`. - -## Option A: pause fields on queue config - -`entity.QueueConfig` gains two booleans, zero value false. The `queueconfig.Store` interface is unchanged. The gateway reads the fields through the existing `Get` in `land.go`. The orchestrator builds a `queueconfig.Store` and a `consumergate.Gate` that wraps it and closes for queues whose freeze field is set. - -The entity gains the two fields: - -```go -type QueueConfig struct { - Name string `json:"name" yaml:"name"` - - // RejectLands is true while the queue refuses new land requests. - RejectLands bool `json:"reject_lands" yaml:"reject_lands"` - - // FreezeProcessing is true while the queue's accepted requests are not advanced. - FreezeProcessing bool `json:"freeze_processing" yaml:"freeze_processing"` -} -``` - -The gateway keeps the cfg it currently discards in `land.go` and rejects before minting an ID: - -```go -cfg, err := c.queueConfigs.Get(ctx, queue) -if err != nil { - // existing not-found and lookup-failure handling -} -if cfg.RejectLands { - return entity.LandResult{}, errs.NewUserError(&QueuePausedError{Queue: queue}) -} -``` - -The orchestrator wires a gate over the same store: - -```go -func (g queueConfigGate) Enter(ctx context.Context, key consumergate.Key) (consumergate.Entry, error) { - cfg, err := g.queueConfigs.Get(ctx, key.Partition.Tenant) // tenant is the queue name - if err != nil { - return nil, err - } - return g.entry(cfg.FreezeProcessing), nil -} -``` - -Pros: -- Smallest change, and no new extension. -- Both switches live on one entity, so a deployer wires a single `Store` backend. -- A `Get` failure already fails `Land()`, so reject fails closed with no new code. - -Cons: -- Adds surface to queue config, setting a precedent for more runtime knobs there. -- The orchestrator gains a `queueconfig` dependency it does not have today: its own `QUEUE_CONFIG_PATH` and store construction. It learns its queues from `Profiles`, so it would then hold two queue lists (`Profiles` and queue config) that can disagree. -- With the OSS YAML store each service holds its own read-once copy, so a pause change takes effect in each service only after that service restarts. -- Puts operational state on an entity documented as an immutable registry of names, with behavior living in extensions, so the entity comment must be rewritten. This is milder than it sounds: each value `Get` returns is still an immutable snapshot, and only what `Get` returns changes over time, as with any live-backed store. The lasting cost is precedent, since the next runtime knob then has an obvious home on `QueueConfig`. - -## Option B: separate pause-state extension - -A new extension, `queuepause`, is a per-queue source of pause state. It returns the two switches for a queue and decides nothing. OSS code enforces them, so OSS and every deployment run the same rejection and freezing logic, and a deployer only supplies where the state comes from. +## Design -The state and the contract: +`queuepause.Store` is a per-queue source of pause state. It returns a `QueuePause` entity for a queue and decides nothing. `QueuePause` holds the two switches and the reason. `Get` is keyed by queue name, like `queueconfig.Store`, and returns the zero state for a queue with no entry, so absence means not paused. Both the gateway and the orchestrator resolve it, so the contract and its implementations live at `submitqueue/extension/queuepause/`. -```go -package entity +OSS ships three implementations: -type QueuePauseState struct { - // RejectLands is true while the queue refuses new land requests. - RejectLands bool +- `noop` never pauses and is the default when nothing is configured. +- `file` re-reads a small YAML file on each `Get`, so an operator pauses or unpauses a queue by editing the file, with no restart. This mirrors the file consumer gate, which re-reads on every delivery. An operator should replace the file by rename so a reader never sees a partial write. +- `fake` is an in-memory store for tests, with injectable state and read errors. It is never used in production. - // FreezeProcessing is true while the queue's accepted requests are not advanced. - FreezeProcessing bool -} -``` +A deployer supplies its own store backed by its config service. Both services select the store through one shared helper in `service/submitqueue/core`. Setting `QUEUE_PAUSE_CONFIG_PATH` selects `file`, and the file must exist at startup. Leaving it unset selects `noop`. -```go -package queuepause +The gateway reads the state in `Land()` after the queue lookup and before minting an ID, and rejects when `RejectLands` is set. The orchestrator wraps its existing consumer gate in a decorator that closes for a queue while `FreezeProcessing` is set and otherwise defers to the wrapped gate. It cannot live under `platform/` because it depends on `queuepause.Store`. -type Store interface { - Get(ctx context.Context, queue string) (entity.QueuePauseState, error) -} -``` +### Requirements for implementations -`Get` is keyed by queue name, like `queueconfig.Store`, and returns the zero state for a queue with no entry, so absence means not paused. Both the gateway and the orchestrator resolve it, so the contract and its implementations live at `submitqueue/extension/queuepause/`. +An implementation of `queuepause.Store` must: -OSS ships two implementations. `noop` never pauses and is the default when nothing is configured. `file` re-reads a small YAML file on each `Get`, so an operator pauses or unpauses a queue by editing the file, with no restart. This mirrors the file consumer gate, which re-reads on every delivery. A deployer supplies its own store backed by its config service. +- **Return the zero state, and no error, for a queue it has no entry for.** Absence means not paused. +- **Return an error when the backend cannot be read, never the zero state.** Callers that enforce a switch fail closed on it, so an outage must not look like an unpaused queue. +- **Reflect a change without a restart.** Any caching bounds how long a pause takes to apply and release, and should stay short, because `Get` runs on every `Land()` and every gated delivery. +- **Be safe for concurrent use.** +- **Fail at construction on misconfiguration**, such as a missing path or invalid contents, so a bad deployment fails at startup and not on the first land. +- **Parse strictly.** Unknown fields and entries with an empty or duplicate queue name are errors, so a misspelled switch cannot silently leave a queue unpaused. +- **Wrap context errors with `%w`**, so a cancelled read is recognized as cancellation and not as a backend failure. -`land.go` reads the state and rejects before minting an ID. A failed read fails the request, so reject fails closed: +## Alternative: pause fields on queue config -```go -state, err := c.queuePauses.Get(ctx, queue) -if err != nil { - return entity.LandResult{}, fmt.Errorf("failed to read pause state for queue=%s: %w", queue, err) -} -if state.RejectLands { - return entity.LandResult{}, errs.NewUserError(&QueuePausedError{Queue: queue}) -} -``` +`entity.QueueConfig` could carry the two booleans, with the gateway reading them through the existing `queueconfig.Store` and the orchestrator wrapping that store in a gate. It is the smaller change, and a failed `Get` already fails `Land()`. It was rejected for these reasons: -A `consumergate.Gate` adapter in the SubmitQueue domain closes the gate for a queue while `FreezeProcessing` is set. It depends on `queuepause.Store`, so it cannot live under `platform/`: - -```go -func (g queuePauseGate) Enter(ctx context.Context, key consumergate.Key) (consumergate.Entry, error) { - state, err := g.pauses.Get(ctx, key.Partition.Tenant) // tenant is the queue name - if err != nil { - return nil, err - } - return entry{blocked: state.FreezeProcessing}, nil -} -``` - -Pros: -- Keeps behavior off queue config. -- OSS users get a working pause, and a deployer needs one implementation per backend. -- Leaves the name `admission` free for a richer, content-based gate later. - -Cons: -- New extension surface: interface, `noop` and `file` implementations, mock, README, and wiring in both services. -- The orchestrator gains a dependency it does not have today: a `queuepause.Store`. -- Queue-name validation still needs a home. If `queueconfig` leaves OSS, nothing rejects an unknown queue: `Profiles.For` falls back to the default profile for any name. +- The orchestrator would gain its own `queueconfig` dependency and a second queue list next to `Profiles`, and the two can disagree. +- The OSS YAML store is read once at startup, so each service would apply a pause change only after a restart. +- It puts operational state on an entity documented as an immutable registry of names, and makes queue config the default home for the next runtime knob. ## Open questions -- **Fail open or closed on freeze.** Reject fails closed when pause state cannot be read, because `Land()` returns the error. However, the consumer treats a gate read error as open, so an adapter that returns the store's error would silently unfreeze a queue during a store outage. The adapter can instead return a blocked entry on error to fail closed. Which posture should freeze take? - **Cancel during freeze.** Should cancel take effect while processing is frozen, as an incident workflow of freeze, cancel, unfreeze would need? If so, freeze becomes no new effects rather than nothing runs, and needs the topic on the gate key ([Cancellation](#cancellation)). -- **OSS backing.** `file` is proposed first because it needs no schema and works with a mounted config file. A MySQL-backed store would serve multi-host OSS deployments, and needs a way for an operator to write it. +- **OSS backing.** `file` needs no schema and works with a mounted config file. A MySQL-backed store would serve multi-host OSS deployments, and needs a way for an operator to write it. - **Read cost.** The `file` store reads on every `Land()` and on every gated delivery. A short cache may be needed. +- **Queue-name validation.** If `queueconfig` leaves OSS, nothing rejects an unknown queue, because `Profiles.For` falls back to the default profile for any name.