diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 94352628c..03e00e1b7 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -5,7 +5,7 @@ Thank you for your interest in contributing to SubmitQueue! Whether you are repo
## Getting Started
1. Read the [Development Setup](doc/howto/DEVELOPMENT.md) guide for prerequisites, building, and running tests.
-2. Review the [Architecture Guide](CLAUDE.md) to understand project layout, conventions, and code style.
+2. Review the [Architecture Guide](AGENTS.md) to understand project layout, conventions, and code style.
3. Check the [Testing Guide](doc/howto/TESTING.md) for testing patterns and requirements.
## Development Workflow
@@ -34,7 +34,7 @@ Thank you for your interest in contributing to SubmitQueue! Whether you are repo
- Include tests for new functionality.
- Ensure all existing tests pass (`make test`).
- Ensure CI passes before requesting review.
-- Follow the existing code style and patterns described in the [Architecture Guide](CLAUDE.md).
+- Follow the existing code style and patterns described in the [Architecture Guide](AGENTS.md).
- Fill out the PR template with a description, motivation, and test plan.
## Code Review
diff --git a/README.md b/README.md
index d05784253..37791cca0 100644
--- a/README.md
+++ b/README.md
@@ -11,7 +11,7 @@ Designed for large monorepos and fast-moving teams where concurrent changes can
## Repository layout
-Cross-domain Go code (errors, metrics, consumer framework, HTTP helpers, shared entities, shared extension contracts) lives under [`platform/`](platform/README.md). Each product domain has its own tree (`submitqueue/`, `stovepipe/`, …) and grows into `gateway/`, `orchestrator/`, `entity/`, `extension/`, and domain-local `core/` — though a domain may start smaller (Stovepipe is currently a single Ping-only service with just `controller/`). See [CLAUDE.md](CLAUDE.md) for conventions and import paths.
+Cross-domain Go code (errors, metrics, consumer framework, HTTP helpers, shared entities, shared extension contracts) lives under [`platform/`](platform/README.md). Each product domain has its own tree (`submitqueue/`, `stovepipe/`, …) and grows into `gateway/`, `orchestrator/`, `entity/`, `extension/`, and domain-local `core/` — though a domain may start smaller (Stovepipe is currently a single Ping-only service with just `controller/`). See [AGENTS.md](AGENTS.md) for conventions and import paths.
## Quick Start
@@ -49,7 +49,7 @@ The queue's own logic is real in all three: validation, batching, conflict analy
| [Development Setup](doc/howto/DEVELOPMENT.md) | Prerequisites, build, environment, IDE setup |
| [Contributing](CONTRIBUTING.md) | How to contribute, workflow, guidelines |
| [Testing Guide](doc/howto/TESTING.md) | Unit, integration, and E2E testing patterns |
-| [Architecture Guide](CLAUDE.md) | Project layout, patterns, conventions |
+| [Architecture Guide](AGENTS.md) | Project layout, patterns, conventions |
| [Examples](service/README.md) | Running services, clients, API reference |
| [RFCs](doc/rfc/index.md) | Design documents and proposals |
diff --git a/doc/howto/TESTING.md b/doc/howto/TESTING.md
index 08893b8fc..643c107d5 100644
--- a/doc/howto/TESTING.md
+++ b/doc/howto/TESTING.md
@@ -73,9 +73,9 @@ make build-all-linux # Build Linux binaries for the local docker-
- Speed: Fast (< 1s typically)
**2. Integration Tests** - Service in isolation with real dependencies
-- Location: `test/integration/submitqueue/{service}/`
-- Run: `make integration-test-{service}`
-- Containers: MySQL + one service
+- Location: `test/integration/submitqueue//` (e.g., `gateway/`, `orchestrator/`, `extension//`)
+- Run: `make integration-test-submitqueue-gateway`, `make integration-test-submitqueue-orchestrator`, `make integration-test-submitqueue-consumer`, or `make integration-test-extensions`
+- Containers: MySQL + one service or the extension's dependencies
- Tests one service isolated from others
**3. E2E Tests** - Complete workflows across all services
@@ -194,14 +194,14 @@ make local-stop
make local-submitqueue-gateway-start
# Test Ping API (port shown by make local-submitqueue-ps)
-grpcurl -plaintext -d '{"message": "hello"}' localhost: submitqueue.SubmitQueueGateway/Ping
+grpcurl -plaintext -d '{"message": "hello"}' localhost: uber.submitqueue.gateway.SubmitQueueGateway/Ping
# Test Land API
grpcurl -plaintext -d '{
"queue": "test-queue",
"change": {"source": "github", "ids": ["PR-123"]},
"strategy": "REBASE"
-}' localhost: submitqueue.SubmitQueueGateway/Land
+}' localhost: uber.submitqueue.gateway.SubmitQueueGateway/Land
# Stop
make local-submitqueue-gateway-stop
@@ -213,7 +213,7 @@ make local-submitqueue-gateway-stop
make local-submitqueue-orchestrator-start
# Test Ping API (port shown by make local-submitqueue-ps)
-grpcurl -plaintext -d '{"message": "hello"}' localhost: submitqueue.SubmitQueueOrchestrator/Ping
+grpcurl -plaintext -d '{"message": "hello"}' localhost: uber.submitqueue.orchestrator.SubmitQueueOrchestrator/Ping
# Stop
make local-submitqueue-orchestrator-stop
@@ -256,18 +256,18 @@ brew install grpcurl # macOS
grpcurl -plaintext localhost: list
# Describe a service
-grpcurl -plaintext localhost: describe submitqueue.SubmitQueueGateway
+grpcurl -plaintext localhost: describe uber.submitqueue.gateway.SubmitQueueGateway
# Call Ping
grpcurl -plaintext -d '{"message": "test"}' \
- localhost: submitqueue.SubmitQueueGateway/Ping
+ localhost: uber.submitqueue.gateway.SubmitQueueGateway/Ping
# Call Land
grpcurl -plaintext -d '{
"queue": "my-queue",
"change": {"source": "github", "ids": ["PR-456"]},
"strategy": "REBASE"
-}' localhost: submitqueue.SubmitQueueGateway/Land
+}' localhost: uber.submitqueue.gateway.SubmitQueueGateway/Land
```
### Available Commands
@@ -363,9 +363,9 @@ docker network ls | grep sq-test | awk '{print $1}' | xargs docker network rm
### Adding Integration Tests
-1. Add test to `test/integration/submitqueue/{service}/suite_test.go`
-2. Use suite's resources (`s.client`, `s.db`)
-3. Run: `make integration-test-{service}`
+1. Add test to `test/integration/submitqueue//suite_test.go` (e.g., `test/integration/submitqueue/gateway/suite_test.go` for Gateway, `test/integration/submitqueue/orchestrator/suite_test.go` for Orchestrator, or a subdirectory under `test/integration/submitqueue/extension/` for extension tests).
+2. Use suite's resources (`s.client`, `s.db`).
+3. Run the matching Makefile target such as `make integration-test-submitqueue-gateway`.
Example:
```go
@@ -387,7 +387,7 @@ assert.Equal(s.T(), "expected", resp.Value)
## See Also
-- [CLAUDE.md](../../CLAUDE.md) - Development guidelines and project structure
+- [AGENTS.md](../../AGENTS.md) - Development guidelines and project structure
- [service/submitqueue/docker-compose.yml](../../service/submitqueue/docker-compose.yml) - Full stack service definitions
- [service/submitqueue/gateway/server/docker-compose.yml](../../service/submitqueue/gateway/server/docker-compose.yml) - Gateway isolation
- [service/submitqueue/orchestrator/server/docker-compose.yml](../../service/submitqueue/orchestrator/server/docker-compose.yml) - Orchestrator isolation
diff --git a/doc/rfc/runway/workflow.md b/doc/rfc/runway/workflow.md
index 568e8d470..b6d1cf464 100644
--- a/doc/rfc/runway/workflow.md
+++ b/doc/rfc/runway/workflow.md
@@ -1,16 +1,16 @@
# Runway Workflow
-Runway is the landing service: it owns VCS operations — mergeability checking and landing — on behalf of SubmitQueue. Runway is a single service (the domain *is* the service): it subscribes to two inbound topics (`merge-conflict-checker`, `merger`) and publishes results to two outbound topics (`merge-conflict-checker-signal`, `merger-signal`). It is a consumer-only service with no gateway; work arrives via topic queues and results leave via topic queues.
+Runway is the landing service: it owns VCS operations — mergeability checking and landing — on behalf of SubmitQueue. Runway is a single service (the domain *is* the service): it subscribes to two inbound topics (`merge-conflict-check`, `runway-merge`) and publishes results to two outbound topics (`merge-conflict-check-signal`, `merge-signal`). It is a consumer-only service with no gateway; work arrives via topic queues and results leave via topic queues.
## Merge-conflict check and merge
-The two queues operate at different granularities:
+The two queues are the same shape but different commit semantics:
-- **merge-conflict-check** is request-level. A merge request carries an ordered sequence of steps (changes + merge strategy). Runway performs a read-only trial merge and publishes per-step mergeability results back.
+- **merge-conflict-check** is a dry run. A merge request carries an ordered sequence of steps (changes + merge strategy). Runway performs a read-only trial merge and publishes per-step mergeability results back.
-- **merge** is batch-level. A merge request carries the same payload but Runway commits the result and reports the revisions it produced (per-step output IDs).
+- **merge** is the committing version. A merge request carries the same payload but Runway commits the result and reports the revisions it produced (per-step output IDs).
-A third operation — **promote** — pushes a commit to a ref as-is (`--ff-only`). The primary use case is forwarding a landed SHA from `main` to `verified/main` without creating a new merge commit. Promote reuses the merge queue with the `PROMOTE` merge strategy; Runway fast-forwards the target ref and reports the same SHA back as the output ID.
+- **promote** is a special `PROMOTE` strategy on the `merge` queue: it pushes a commit to a ref as-is (`--ff-only`). The primary use case is forwarding a landed SHA from `main` to `verified/main` without creating a new merge commit.
These are independent input-output flows. A merge-conflict check can run without a merge ever running, and a merge does not depend on a prior check.
@@ -23,39 +23,39 @@ The outbound topics partition by SubmitQueue queue name, matching SubmitQueue's
## Workflow
```
- ┌─────────────────────────────────────────────────────┐
- │ submitqueue orchestrator │
- └──────────┬───────────────────────────┬──────────────┘
- │ │
- MergeRequest (dry run) MergeRequest (commit)
- │ │
- ▼ ▼
- [merge-conflict-checker] [merger]
- │ │
- merge-conflict-check ctrl merge ctrl
- (read-only) (apply + commit)
- │ │
- MergeResult MergeResult
- │ │
- ▼ ▼
- [merge-conflict-checker-signal] [merger-signal]
- │ │
- ▼ ▼
- ┌──────────┬───────────────────────────┬──────────────┐
- │ merge-conflict-check- merge-signal ctrl │
- │ signal ctrl (update batch state, │
- │ (update request fan out to conclude) │
- │ mergeability) │
- │ submitqueue orchestrator │
- └─────────────────────────────────────────────────────┘
+ ┌─────────────────────────────────────────────────────┐
+ │ submitqueue orchestrator │
+ └──────────┬───────────────────────────┬──────────────┘
+ │ │
+ MergeRequest (dry run) MergeRequest (commit)
+ │ │
+ ▼ ▼
+ [merge-conflict-check] [runway-merge]
+ │ │
+ merge-conflict-check ctrl merge ctrl
+ (read-only) (apply + commit)
+ │ │
+ MergeResult MergeResult
+ │ │
+ ▼ ▼
+ [merge-conflict-check-signal] [merge-signal]
+ │ │
+ ▼ ▼
+ ┌──────────┬───────────────────────────┬──────────────┐
+ │ merge-conflict-check- merge-signal ctrl │
+ │ signal ctrl (update batch state, │
+ │ (update request fan out to conclude) │
+ │ mergeability) │
+ │ submitqueue orchestrator │
+ └─────────────────────────────────────────────────────┘
```
## Per-controller summary
| Controller | In | Out | One-line role |
|---|---|---|---|
-| **merge-conflict-check** | MergeRequest | MergeResult -> merge-conflict-checker-signal | Dry-run merge: check mergeability of ordered steps against the target branch (read-only) |
-| **merge** | MergeRequest | MergeResult -> merger-signal | Apply, commit, and report per-step output IDs |
+| **merge-conflict-check** | MergeRequest | MergeResult -> merge-conflict-check-signal | Dry-run merge: check mergeability of ordered steps against the target branch (read-only) |
+| **merge** | MergeRequest | MergeResult -> merge-signal | Apply, commit, and report per-step output IDs |
The merge-conflict-check controller always publishes a result — even when all steps are mergeable — so SubmitQueue receives a definitive answer. On infrastructure error it nacks for retry.
@@ -83,8 +83,8 @@ Runway has no persistent state — no request store, no job store, no database.
### Runway
-Runway is a single service. It subscribes to two inbound topics (`merge-conflict-checker`, `merger`), performs VCS operations through a pluggable extension, and publishes results to two outbound topics (`merge-conflict-checker-signal`, `merger-signal`). It owns no persistent data.
+Runway is a single service. It subscribes to two inbound topics (`merge-conflict-check`, `runway-merge`), performs VCS operations through a pluggable extension, and publishes results to two outbound topics (`merge-conflict-check-signal`, `merge-signal`). It owns no persistent data.
### Shared: the messaging queue
-Runway communicates with SubmitQueue only through the messaging queue. The inbound topics are owned by runway; the outbound topics are owned by SubmitQueue.
+Runway communicates with SubmitQueue only through the messaging queue. The contract is owned by Runway and published under `api/runway/messagequeue/`; both inbound and outbound topic keys live there. SubmitQueue publishes `MergeRequest` messages and consumes the `MergeResult` signals.
diff --git a/doc/rfc/stovepipe/steps/build.md b/doc/rfc/stovepipe/steps/build.md
index 0078ecd2d..0501c9900 100644
--- a/doc/rfc/stovepipe/steps/build.md
+++ b/doc/rfc/stovepipe/steps/build.md
@@ -151,7 +151,7 @@ So `build`'s `Trigger` gets its own shape under `stovepipe/extension/buildrunner
### Stovepipe `BuildRunner` contract (design sketch)
-Not implemented here. `BuildID`, `BuildStatus`, and `BuildMetadata` are defined locally in `stovepipe/entity`, shaped the same as SubmitQueue's equivalents in `submitqueue/entity` but not the same Go types — per the reviewer preference recorded in [Alternatives considered for sharing the contract](#alternatives-considered-for-sharing-the-contract), a shared `platform/base`/`platform/extension/buildrunner` contract was considered and set aside in favor of keeping each domain's interface separate and reusing at the implementation layer instead. `stovepipe/extension/buildrunner` holds `Trigger`, `Status`, `Cancel`, `Config`, and the `Factory` interface, per [CLAUDE.md](CLAUDE.md)'s extension rules.
+Not implemented here. `BuildID`, `BuildStatus`, and `BuildMetadata` are defined locally in `stovepipe/entity`, shaped the same as SubmitQueue's equivalents in `submitqueue/entity` but not the same Go types — per the reviewer preference recorded in [Alternatives considered for sharing the contract](#alternatives-considered-for-sharing-the-contract), a shared `platform/base`/`platform/extension/buildrunner` contract was considered and set aside in favor of keeping each domain's interface separate and reusing at the implementation layer instead. `stovepipe/extension/buildrunner` holds `Trigger`, `Status`, `Cancel`, `Config`, and the `Factory` interface, per [AGENTS.md](AGENTS.md)'s extension rules.
```go
// package buildrunner (stovepipe/extension/buildrunner)
@@ -266,7 +266,7 @@ Key the `Build` by identity derived from the Request — `buildKey(R) = R.ID` fo
| Pros | Cons |
|---|---|
| Redelivery dedup by direct get: checking `BuildStore.Get(buildKey(R))` before triggering means at-least-once delivery never starts a second build | A second id concept (`Build.ID` beside `Build.RunnerBuildID`) carried by every entity, signature, and reader forever |
-| `Request` → `Build` navigation with no reverse index, per the KV key-derivation rule in CLAUDE.md | No current reader needs to *derive* a build id — the id travels in every message hop, so each consumer already holds the key it needs |
+| `Request` → `Build` navigation with no reverse index, per the KV key-derivation rule in [AGENTS.md](AGENTS.md) | No current reader needs to *derive* a build id — the id travels in every message hop, so each consumer already holds the key it needs |
| Enforces (rather than assumes) the direct-navigation property SubmitQueue's speculate takes on faith | Diverges entity shape and controller flow from SubmitQueue, weakening the "structurally the same controller" claim and dual-implementing-backend symmetry |
Trade-offs: the dedup guards a rare event at a permanent modeling cost. The duplicate it prevents arises only from a redelivery inside the trigger window — rare, and already harmless (identical scope; `buildsignal`'s superseded short-circuit and its first-writer-wins outcome CAS make the loser a no-op — see [Idempotency](#idempotency)). The prospective key-derivers — a future canceller, or `analyze` reaching back to the Phase-1 target graph — would need to be handed the id by their producing stage instead, if those designs land.
@@ -326,9 +326,9 @@ Plus the `BuildID{ID string}` wire type in `stovepipe/entity` (same "id only tra
- `Create(ctx, build entity.Build) error` — `ErrAlreadyExists` if the id is taken.
- `Get(ctx, id string) (entity.Build, error)` — `ErrNotFound` if absent.
-- `Update(ctx, build entity.Build, oldVersion, newVersion int32) error` — pure conditional write; `ErrVersionMismatch` on a stale guard. The controller computes `newVersion = oldVersion + 1`, calls the store, and assigns `build.Version = newVersion` only on success (see [CLAUDE.md](CLAUDE.md) and the [storage README](submitqueue/extension/storage/README.md)).
+- `Update(ctx, build entity.Build, oldVersion, newVersion int32) error` — pure conditional write; `ErrVersionMismatch` on a stale guard. The controller computes `newVersion = oldVersion + 1`, calls the store, and assigns `build.Version = newVersion` only on success (see [AGENTS.md](AGENTS.md) and the [storage README](submitqueue/extension/storage/README.md)).
-Single-key reads/writes only — no list-by-request, no query-by-attribute — per the key/value-shaped extension rule in [CLAUDE.md](CLAUDE.md).
+Single-key reads/writes only — no list-by-request, no query-by-attribute — per the key/value-shaped extension rule in [AGENTS.md](AGENTS.md).
**`Request` additions** (extending the existing entity, which already has `ID/Queue/URI/State/Version`):
diff --git a/doc/rfc/stovepipe/steps/process.md b/doc/rfc/stovepipe/steps/process.md
index 7f48603ac..4bece0a8a 100644
--- a/doc/rfc/stovepipe/steps/process.md
+++ b/doc/rfc/stovepipe/steps/process.md
@@ -199,7 +199,7 @@ Per-queue knobs such as `max_concurrent` live outside this row — see [Per-Queu
| *(owned by buildsignal)* succeeded / failed / cancelled | Phase 1 build outcome | **yes** |
| *(later)* building, recording, … | Finer states as downstream stages need them | — |
-Transitions use the repo's optimistic-locking pattern: compute `newVersion = oldVersion + 1`, call `RequestStore.Update(ctx, req, oldVersion, newVersion)`, assign `req.Version = newVersion` only on success (see [storage README](../../../../submitqueue/extension/storage/README.md) and [CLAUDE.md](../../../../CLAUDE.md)).
+Transitions use the repo's optimistic-locking pattern: compute `newVersion = oldVersion + 1`, call `RequestStore.Update(ctx, req, oldVersion, newVersion)`, assign `req.Version = newVersion` only on success (see [storage README](../../../../submitqueue/extension/storage/README.md) and [AGENTS.md](../../../../AGENTS.md)).
## Storage contract additions
diff --git a/doc/rfc/stovepipe/workflow.md b/doc/rfc/stovepipe/workflow.md
index c7b62442c..73bba9045 100644
--- a/doc/rfc/stovepipe/workflow.md
+++ b/doc/rfc/stovepipe/workflow.md
@@ -56,7 +56,7 @@ The ref is a *cache* of the last-green URI, not a second record of greenness. It
| **SourceControl** | Resolve a Queue name to its current head URI; answer ancestry/comparison questions between two URIs (is the new head a fast-forward descendant of the last green, or was history rewritten?); enumerate commits in a range; advance the Queue's **promotion ref** to a commit. The sole owner of URI semantics, including which refs a Queue name resolves to. |
| **build-runner** | Build a scope at a URI (optionally relative to a baseline URI), returning pass/fail and the target graph. See [build-runner.md](../submitqueue/build-runner.md). |
| **Hooks** | Deliver Stovepipe's greenness events to downstream systems — "this URI / this project is now green (or not green)". Fire-and-forget notification, decoupled so Stovepipe does not know or care who consumes the event. Not implemented yet; it will be the shared cross-domain hook seam rather than a Stovepipe-specific extension. See [hook-framework.md](../hook-framework.md). |
-| **Storage** | Persist Queues (incl. last-green URI), Requests, build records, and per-URI / per-project greenness. Key/value-shaped per the extension-design rules in [CLAUDE.md](../../../CLAUDE.md). |
+| **Storage** | Persist Queues (incl. last-green URI), Requests, build records, and per-URI / per-project greenness. Key/value-shaped per the extension-design rules in [AGENTS.md](../../../AGENTS.md). |
Hooks are the notification boundary. When a validation fact is recorded — whole-repo green/not-green, or later a project green/not-green — the event reaches deployment systems, dashboards, and developer tooling without any of them polling Stovepipe's store, and each environment can route it to its own downstream (a deploy gate, a Slack notifier, an event bus) without changing the pipeline. The mechanism is the cross-domain hook framework rather than a call out of the recording stage: `record` publishes a `HookEvent` to Stovepipe's `hook` topic, and a dispatcher stage consumes it and invokes the wired hooks, so a slow or failing downstream cannot add latency to the pipeline. Neither half exists yet; see [record.md](steps/record.md#hooks) for the fact-to-event mapping and its open questions.
diff --git a/doc/rfc/submitqueue/extension-contract.md b/doc/rfc/submitqueue/extension-contract.md
index 81e086e5b..80c241ff2 100644
--- a/doc/rfc/submitqueue/extension-contract.md
+++ b/doc/rfc/submitqueue/extension-contract.md
@@ -14,7 +14,7 @@ Both unblock with the shape `conflict` already uses: accept identity, resolve in
## Principle
- **Decision/action extensions** take orchestrator identity at their stage granularity and resolve granular content through narrowly-injected dependencies. Request stage → `entity.Request`; batch stage → `entity.Batch` / `[]entity.Batch`. Both are thin reference entities (a `Request` carries URIs, not diffs; a `Batch` carries IDs, not changes).
-- **Resolution targets** — `storage`, `changestore`, `queueconfig` — stay key/value-shaped. They are what the others resolve *through* (see [storage/README.md](../../../submitqueue/extension/storage/README.md) and CLAUDE.md). Refinement: the storage *aggregate* has since gained the same per-queue factory resolution every other seam has — the stores it hands back remain strictly key/value, bound to their queue, while the cross-queue read-model stores stay individually-injected singletons.
+- **Resolution targets** — `storage`, `changestore`, `queueconfig` — stay key/value-shaped. They are what the others resolve *through* (see [storage/README.md](../../../submitqueue/extension/storage/README.md) and [AGENTS.md](../../../AGENTS.md)). Refinement: the storage *aggregate* has since gained the same per-queue factory resolution every other seam has — the stores it hands back remain strictly key/value, bound to their queue, while the cross-queue read-model stores stay individually-injected singletons.
- **Output mirrors the input unit.** Each output element self-identifies with the input it corresponds to — `changeprovider`'s `ChangeInfo` carries its `URI`, `conflict`'s `Conflict` carries its `BatchID` — so a flat list suffices and the caller correlates results back to inputs without re-deriving boundaries. A *wrapper* entity (`entity.BatchChanges`) is introduced only to aggregate *up* to a coarser unit than the elements — the scorer needs batch-wide line/file totals, so the rollup earns its keep; no `RequestChanges` exists because nothing needs request-wide rollups. And when the input is a *collection* of independently-actioned units, the output groups by them: `pusher`, fed `[]entity.Batch`, returns outcomes grouped per batch, the same way `conflict` already tags each `Conflict` with its in-flight `BatchID`.
### What each stage resolves today
diff --git a/doc/rfc/submitqueue/modular-queue-wiring.md b/doc/rfc/submitqueue/modular-queue-wiring.md
index c63a59479..1acdddb89 100644
--- a/doc/rfc/submitqueue/modular-queue-wiring.md
+++ b/doc/rfc/submitqueue/modular-queue-wiring.md
@@ -12,7 +12,7 @@ The orchestrator's example `main.go` (`example/submitqueue/orchestrator/server/m
Adding a new queue today requires changes in **three places**: YAML config (`queues.yaml`), Go code (`newQueueRegistry`), and a recompile. Adding a new pipeline stage requires **two coordinated edits** (topic list + controller registration). The topic → subscription → DLQ subscription → DLQ controller linkage is maintained by copy-paste across 12 stages, where forgetting any half creates a silent failure.
-The [TODO on line 475](../../../service/submitqueue/orchestrator/server/main.go) already flags the queue-registry pattern as a candidate for promotion into the domain layer, contingent on a trigger: a second consumer needing the same wiring, data-driven config, or lifecycle requirements.
+The queue-registry pattern is flagged as a candidate for promotion into the domain layer once a second consumer needs the same wiring, data-driven config, or lifecycle requirements. Today the orchestrator's `main.go` wires it inline.
## Vocabulary
diff --git a/doc/rfc/submitqueue/speculation-generator-best-first.md b/doc/rfc/submitqueue/speculation-generator-best-first.md
index cfe87f37e..458e5e007 100644
--- a/doc/rfc/submitqueue/speculation-generator-best-first.md
+++ b/doc/rfc/submitqueue/speculation-generator-best-first.md
@@ -447,7 +447,7 @@ The ordering stays the same. `CandidatePath.RankingScore` contains this logarith
- `Succeeded` fixes an assumption to succeeds.
- `Failed` or `Cancelled` fixes an assumption to fails.
- `Cancelling` remains undecided because cancellation may lose a race with completion.
-- `Merging` also remains undecided, because a merge can fail. It is tempting to treat it as committed to landing and skip the scorer call, but that puts a state-specific policy inside the search: whether a path betting against a merging batch is worth funding is a question of price, and price belongs to the scorer. The allocator draws the same line — "no batch state enters this decision" — and the generator holds it too. Nothing is lost by staying open, because a head can never merge ahead of a dependency it took a position on (see [speculation.md](speculation.md)); the cost of an unlikely path is budget, which is the allocator's to ration.
+- `Merging` also remains undecided, because a merge can fail. It is tempting to treat it as committed to landing and skip the scorer call, but that puts a state-specific policy inside the search: whether a path betting against a merging batch is worth funding is a question of price, and price belongs to the scorer. The allocator draws the same line — "no batch state enters this decision" — and the generator holds it too. Nothing is lost by staying open: a single passed path still waits for the merge result, while passed paths covering every outcome let the controller bypass the dependency (see [speculation.md](speculation.md)). Funding the unlikely side spends budget, which is the allocator's to ration.
- A fixed assumption stays in the returned path but contributes probability 1 and has no flip.
- A shared dependency is scored once per run.
diff --git a/doc/rfc/submitqueue/speculation.md b/doc/rfc/submitqueue/speculation.md
index ebdd49896..1887cff7e 100644
--- a/doc/rfc/submitqueue/speculation.md
+++ b/doc/rfc/submitqueue/speculation.md
@@ -4,7 +4,7 @@ A merge queue that verifies one change at a time is limited by its slowest build
Work enters SubmitQueue as **batches** — changes verified and merged together. Two batches **conflict** when they touch the same code, which makes the earlier one a **dependency** of the later. A **path** is one set of assumptions about how a batch's dependencies resolve, and the batch it builds is the path's **head**.
-On every queue update the **speculate controller** reruns from scratch: it reads the current state, applies the incoming signals, asks a pluggable **Speculator** which paths are worth building within the CI budget, and persists only those. Everything else is recomputed next time, never stored. Merging stays strict: a batch merges only after its dependencies resolve and a matching build has passed.
+On every queue update the **speculate controller** reruns from scratch: it reads the current state, applies the incoming signals, asks a pluggable **Speculator** which paths are worth building within the CI budget, and persists only those. Everything else is recomputed next time, never stored. A batch normally merges after its dependencies resolve and a matching build has passed; complete passed coverage of every unresolved outcome lets it bypass those dependencies.
## The speculation run
@@ -54,7 +54,7 @@ Every write is a compare-and-swap: a writer that loses re-reads on a later run.
Verdicts are controller-owned facts: the Speculator can neither compute nor veto them.
-- **Merge (strict).** Each path carries an assumption about every dependency — *succeeds* (built on top of) or *fails* (built without). Once a path's build has passed and every dependency has finished the way the path assumed — one assumed *succeeds* has merged, one assumed *fails* has failed or been cancelled — the speculate controller moves the head to Merging and hands it to Runway. A dependency that is merely *merging* has not finished, because a merge can fail, so it is still waited on. If that hand-off is lost, the next run re-sends it. The same run sets the head's remaining in-flight paths *cancelling*: once one path has passed the others cannot help, and they hold CI slots until they stop. The mergesignal controller records Runway's terminal result: success marks the head Succeeded, while failure marks it Failed. The result publishes a single dirty signal — no per-dependent fan-out — and the next run refutes paths whose assumption disagrees with the result: *fails* assumptions after success, *succeeds* assumptions after failure. The hand-off is idempotent, so Runway reports success without another merge when the change is already present. Down a chain, each head waits for its predecessors to settle, so a chain merges one at a time.
+- **Merge.** Each path carries an assumption about every dependency — *succeeds* (built on top of) or *fails* (built without). Normally, once a path's build has passed and every dependency has finished the way the path assumed — one assumed *succeeds* has merged, one assumed *fails* has failed or been cancelled — the speculate controller moves the head to Merging and hands it to Runway. A dependency that is merely *merging* has not finished, because a merge can fail, so a single matching path still waits for the answer. Complete passed coverage is the exception described in Bypass large diff: it lets a head merge before those answers arrive. If the hand-off is lost, the next run re-sends it. The same run sets the head's remaining in-flight paths *cancelling*: once the head can merge they cannot help, and they hold CI slots until they stop. The mergesignal controller records Runway's terminal result: success marks the head Succeeded, while failure marks it Failed. The result publishes a single dirty signal — no per-dependent fan-out — and the next run refutes paths whose assumption disagrees with the result: *fails* assumptions after success, *succeeds* assumptions after failure. The hand-off is idempotent, so Runway reports success without another merge when the change is already present. A chain ordinarily merges one at a time, but a fully covered head can bypass its unsettled predecessors.
- **Failure (no viable path).** A batch fails when every possible future has a failed build — no path can pass, so it can never merge.
- **Cancel.** A cancelled batch is driven terminal: its in-flight paths are set *cancelling*, then the batch is marked Cancelled once they stop (see Cancellation).
@@ -74,9 +74,9 @@ Example of the payoff either way: `H` conflicts with `B1` and weak `B2`. Relax `
If a batch's passed builds cover *every* way its dependencies could resolve, the outcome is the same either way — so it can merge now, ahead of them. Classic case: a small change stuck behind a slow one is built both with and without it; both pass, and it merges immediately.
-The default Speculator covers the whole space only when doing so is cheap enough, and funds the extra candidates within the build budget. The controller merges early only when a passed path exists for every combination of the dependencies — it reads that straight off the path records. If any combination is missing or unbuilt, the head waits normally.
+The controller checks coverage over only the dependencies that have not settled yet. Settled dependencies pin each surviving path to the outcome that actually happened; for every combination of the remaining dependencies, the path set must contain a passed, unbroken path with that combination of assumptions. If any combination is missing, unbuilt, failed, or contradicted by a settled dependency, the head waits normally. The check only observes paths the Speculator already funded — it does not fund the exponential path space itself or alter the queue's build budget.
-**Not yet implemented on the controller side.** `decide`/`mergeablePath` gate on a single passed path whose assumptions have all been settled by the dependency's actual state; nothing enumerates the combinations. The distinction matters: a *single* passed path that assumed a dependency would fail is not complete coverage, and merging on it while that dependency is still live would put a combination on the trunk that no build validated. Coverage is what makes early merge sound — one path betting the right way is not.
+Coverage makes the bypass sound because whichever way the dependencies later resolve, a passed build already validated the resulting set of changes. The build order and merge order differ: a path assuming dependency `D` succeeds validates `D` then head `H`, while bypass lands `H` before `D`. SubmitQueue treats those orders as content-equivalent. Runway still performs the real merge, so if the reordered changes conflict textually, the older dependency can fail after the newer head has bypassed it; this is an accepted cost of landing the fully covered head early rather than a licence to put unmergeable content on the target.
### Cancellation
diff --git a/doc/rfc/submitqueue/workflow.md b/doc/rfc/submitqueue/workflow.md
index e34eb85e9..b50e3958c 100644
--- a/doc/rfc/submitqueue/workflow.md
+++ b/doc/rfc/submitqueue/workflow.md
@@ -1,6 +1,6 @@
# Orchestrator Workflow
-The orchestrator processes land requests through a queue-driven pipeline of small, single-purpose controllers. The gateway accepts a request over RPC and hands it off asynchronously; from there each controller consumes one topic, advances the request or batch, and publishes to the next topic. Most hops carry only an ID — the controller fetches the entity from storage — while a few entry points (`start`, `buildsignal`, `log`) carry the full payload because there is no row to fetch yet. Some stages cross a service boundary: they publish a full payload to the other service's queue and consume a full payload back, because neither service can read the other's storage. (The `validate` and `merge` stages both hand work to runway — a merge-conflict check and the merge itself — and consume its result on `mergeconflictsignal` / `mergesignal`.) See the queue-payload-boundary rule in [CLAUDE.md](../../../CLAUDE.md).
+The orchestrator processes land requests through a queue-driven pipeline of small, single-purpose controllers. The gateway accepts a request over RPC and hands it off asynchronously; from there each controller consumes one topic, advances the request or batch, and publishes to the next topic. Most hops carry only an ID — the controller fetches the entity from storage — while a few entry points (`start`, `buildsignal`, `log`) carry the full payload because there is no row to fetch yet. Some stages cross a service boundary: they publish a full payload to the other service's queue and consume a full payload back, because neither service can read the other's storage. (The `validate` and `merge` stages both hand work to runway — a merge-conflict check and the merge itself — and consume its result on `mergeconflictsignal` / `mergesignal`.) See the queue-payload-boundary rule in [AGENTS.md](../../../AGENTS.md).
The pipeline has two cycles: `speculate → build → buildsignal → speculate` (CI feedback loop) and `merge → runway → mergesignal → speculate` (land the batch out of process, then advance the next). `conclude` is the only stage that transitions a request to a terminal state; `log` is an append-only sink that any controller can publish to via `submitqueue/core/request.PublishLog`.
diff --git a/service/README.md b/service/README.md
index 2e49af737..939808e29 100644
--- a/service/README.md
+++ b/service/README.md
@@ -13,9 +13,9 @@ Each domain has its own subdirectory with a dedicated README:
| Service | Port | Domain | RPCs | Backing stores |
|---------|------|--------|------|----------------|
| **SubmitQueue Gateway** | 8081 | `submitqueue` | `Ping`, `Land`, `Cancel`, `GetRequestSummaryByID`, `GetRequestSummaryByChangeURI`, `List`, `GetRequestHistoryByID`, `GetRequestHistoryByChangeURI` | MySQL app + queue |
-| **SubmitQueue Orchestrator** | 8082 | `submitqueue` | `Ping` (+ consumes 9 pipeline topics) | MySQL app + queue |
-| **Stovepipe** | 8083 | `stovepipe` | `Ping`, `Ingest` (+ consumes the process topic) | MySQL storage + queue |
-| **Runway** | 8086 | `runway` | `Ping` (+ consumes merge-conflict-check & merge topics) | MySQL queue |
+| **SubmitQueue Orchestrator** | 8082 | `submitqueue` | `Ping` (+ consumes pipeline topics: start, cancel, validate, batch, dependency-analysis, speculate, build, buildsignal, submitqueue-merge, conclude, log, plus DLQ topics, and the two Runway signal topics) | MySQL app + queue |
+| **Stovepipe** | 8083 | `stovepipe` | `Ping`, `Ingest` (+ consumes the process, build, buildsignal, and record topics, plus DLQ topics) | MySQL storage + queue |
+| **Runway** | 8086 | `runway` | `Ping` (+ consumes merge-conflict-check & runway-merge topics) | MySQL queue |
Ports above are the `go run` defaults; under Docker Compose each server listens on `:8080` inside its container and is published on a random ephemeral host port (use `make local-*-ps` / `docker port` to discover it).
@@ -34,7 +34,8 @@ service/
│ └── client/ # Orchestrator ping client
├── stovepipe/
│ ├── docker-compose.yml # Stovepipe service + storage MySQL + queue MySQL
-│ ├── server/ # Stovepipe gRPC server + Dockerfile
+│ ├── docker-compose.debug.yml # Debug variant with delve
+│ ├── server/ # Stovepipe gRPC server + Dockerfile + compose
│ └── client/ # Stovepipe ping client
└── runway/
├── server/ # Runway gRPC server + Dockerfile + compose
@@ -53,6 +54,7 @@ make local-submitqueue-orchestrator-start # orchestrator-only stack
# Stovepipe service (gRPC service + storage MySQL + queue MySQL)
make local-stovepipe-start
+make local-stovepipe-logs
# Runway service (consumer + queue MySQL)
make local-runway-start
diff --git a/service/stovepipe/README.md b/service/stovepipe/README.md
index 4a378bfed..2e8253802 100644
--- a/service/stovepipe/README.md
+++ b/service/stovepipe/README.md
@@ -1,14 +1,18 @@
# Stovepipe Service
-Runnable wiring for the **Stovepipe** domain — a single-service domain (the domain *is* the service). The server exposes two RPCs and runs one internal pipeline stage as a queue consumer:
+Runnable wiring for the **Stovepipe** domain — a single-service domain (the domain *is* the service). The server exposes two RPCs and runs the internal pipeline stages as queue consumers:
- **`Ping`** — health check.
- **`Ingest`** — resolves a queue's head commit, persists a `Request` (and its head URI) to storage, and publishes the request to the **process** stage.
- **process consumer** (`TopicKeyProcess`) — reloads the persisted `Request` from storage and runs the process stage (`stovepipe/controller/process`).
+- **build consumer** (`TopicKeyBuild`) — reloads the persisted `Request` and triggers the build-runner, then publishes to `buildsignal`.
+- **buildsignal consumer** (`TopicKeyBuildSignal`) — polls/records the build's terminal status and releases the queue's in-flight slot, then publishes to `record`.
+- **record consumer** (`TopicKeyRecord`) — writes the whole-repo validation fact, advances the queue's last-green bookmark and promotion ref, and publishes hook events.
+- **DLQ reconciler** — for each internal topic, a `_dlq` consumer that drives stuck requests to a conservative terminal state so the queue's slot is freed.
-The ingest → process hop stays inside one service and one store, so only the request **ID** travels on the queue; the consumer reloads from storage (the source of truth), which keeps messages small and redelivery idempotent. The process topic key and its internal wire contract are owned by the domain under `stovepipe/core/messagequeue/`.
+The ingest → process → build → buildsignal → record hop stays inside one service and one store, so the queue messages carry only request **IDs**; the consumers reload from storage (the source of truth), which keeps messages small and redelivery idempotent. The process, build, buildsignal, and record topic keys and their internal wire contract are owned by the domain under `stovepipe/core/messagequeue/`.
-Stovepipe therefore needs two MySQL databases: a **storage** database (the `request` and `request_uri` tables) and a **queue** database (messaging infrastructure).
+Stovepipe therefore needs two MySQL databases: a **storage** database (the `queue`, `request`, `request_uri`, and `build` tables) and a **queue** database (messaging infrastructure).
## Wiring notes
@@ -22,14 +26,15 @@ Stovepipe therefore needs two MySQL databases: a **storage** database (the `requ
```
stovepipe/
├── docker-compose.yml # Stovepipe service + storage MySQL + queue MySQL
+├── docker-compose.debug.yml # Debug variant with delve
├── server/
-│ ├── main.go # gRPC server (Ping, Ingest) + process-stage consumer wiring
+│ ├── main.go # gRPC server (Ping, Ingest) + pipeline consumer wiring
│ └── Dockerfile
└── client/
└── main.go # Ping client (default :8083)
```
-The Stovepipe controllers live under [`stovepipe/controller/`](../../stovepipe/controller) and its extensions under [`stovepipe/extension/`](../../stovepipe/extension); this directory only contains the runnable wiring and a Docker Compose stack for manual testing.
+The Stovepipe controllers live under [`stovepipe/controller/`](../../stovepipe/controller) (subdirectories for `ingest`, `process`, `build`, `buildsignal`, `record`, and `dlq`) and its extensions under [`stovepipe/extension/`](../../stovepipe/extension); this directory only contains the runnable wiring and a Docker Compose stack for manual testing.
## Configuration
diff --git a/stovepipe/README.md b/stovepipe/README.md
index 475de8846..269e8388d 100644
--- a/stovepipe/README.md
+++ b/stovepipe/README.md
@@ -1,7 +1,7 @@
# Stovepipe
-Stovepipe is currently a single Ping-only service. Its layout:
+Stovepipe is a post-merge validation service. Its layout:
-- `controller/` — business logic (transport-agnostic). Currently exposes the `Ping` RPC.
+- `controller/` — business logic (transport-agnostic). Exposes the `Ping` and `Ingest` RPCs, and consumes the internal pipeline stages (`process`, `build`, `buildsignal`, `record`) plus a DLQ reconciler.
-The wire contract lives under `api/stovepipe/` (`proto/` for the `.proto` source, `protopb/` for the committed generated stubs). Entities, extensions, and the orchestration pipeline will be added back as the service grows.
+The wire contract lives under `api/stovepipe/` (`proto/` for the `.proto` source, `protopb/` for the committed generated stubs). The internal queue contract and topic keys live under `stovepipe/core/messagequeue/`. Storage, source-control, and build-runner extensions live under `stovepipe/extension/`.
diff --git a/stovepipe/extension/storage/README.md b/stovepipe/extension/storage/README.md
index 530b07877..8ced668dc 100644
--- a/stovepipe/extension/storage/README.md
+++ b/stovepipe/extension/storage/README.md
@@ -8,7 +8,7 @@ This is a separate contract from `submitqueue/extension/storage` — same shape
## Optimistic locking contract
-Entities that support concurrent mutation (`Request`, `Build`) carry an `int32 Version` field. `Update` methods take both `oldVersion` (the where-clause guard) and `newVersion` (the value to write) — the store performs a pure conditional write and never computes `oldVersion + 1` itself. Version arithmetic is owned by the controller: it computes `newVersion`, calls `Update`, and only assigns `entity.Version = newVersion` after the call succeeds. See [CLAUDE.md](../../../CLAUDE.md) and the [submitqueue storage README](../../../submitqueue/extension/storage/README.md#optimistic-locking-contract) for the full rationale and the caller pattern — the convention is identical here.
+Entities that support concurrent mutation (`Request`, `Build`) carry an `int32 Version` field. `Update` methods take both `oldVersion` (the where-clause guard) and `newVersion` (the value to write) — the store performs a pure conditional write and never computes `oldVersion + 1` itself. Version arithmetic is owned by the controller: it computes `newVersion`, calls `Update`, and only assigns `entity.Version = newVersion` after the call succeeds. See [AGENTS.md](../../../AGENTS.md) and the [submitqueue storage README](../../../submitqueue/extension/storage/README.md#optimistic-locking-contract) for the full rationale and the caller pattern — the convention is identical here.
## Read-after-write consistency
diff --git a/submitqueue/README.md b/submitqueue/README.md
index 6147d254a..11bcf3b38 100644
--- a/submitqueue/README.md
+++ b/submitqueue/README.md
@@ -6,6 +6,6 @@ SubmitQueue service layout:
- `orchestrator/` — Orchestrator service: coordinates the land pipeline (batch, speculate, build, merge, conclude, ...).
- `extension/` — SubmitQueue-specific extension implementations (storage, counter, changestore, mergechecker, pusher, scorer, conflict, queueconfig, buildrunner, ...).
- `entity/` — SubmitQueue-specific domain entities.
-- `core/` — Infrastructure shared across SubmitQueue's own services (gateway and orchestrator): the queue `consumer` framework and the `request` lifecycle. The SubmitQueue-scoped analogue of the repo-level `core/`.
+- `core/` — Infrastructure shared across SubmitQueue's own services (gateway and orchestrator): the queue `consumer` framework, the `request` lifecycle, and topic keys. The SubmitQueue-scoped analogue of the repo-level `platform/`.
-Cross-domain building blocks live outside this directory: shared entities in `entity/`, shared extensions in `extension/`, and cross-domain infrastructure in the top-level `core/`.
+Cross-domain building blocks live outside this directory: shared entities in `platform/base/`, shared extensions in `platform/extension/`, and cross-domain infrastructure in `platform/`.
diff --git a/submitqueue/entity/request_log.go b/submitqueue/entity/request_log.go
index eac65c040..72bad9ef6 100644
--- a/submitqueue/entity/request_log.go
+++ b/submitqueue/entity/request_log.go
@@ -58,11 +58,11 @@ const (
RequestStatusBatched RequestStatus = "batched"
// RequestStatusSpeculating indicates that the batch containing the request is in speculation:
- // planning, building, or waiting for its dependencies to settle. None of those leaves it able to land.
+ // planning, building, or waiting until either its dependencies settle or passed paths cover every possible outcome.
RequestStatusSpeculating RequestStatus = "speculating"
// RequestStatusSpeculated indicates that the batch containing the request has finished speculating:
- // a build passed on a path whose assumptions all held, and the batch has been cleared to merge.
+ // either a passed path's assumptions all held, or passed paths cover every outcome of its unsettled dependencies.
RequestStatusSpeculated RequestStatus = "speculated"
// RequestStatusLanding indicates that the request is actively being landed (e.g., source control operation is in progress to push the change to the target branch).
diff --git a/submitqueue/extension/speculation/speculator/standard/README.md b/submitqueue/extension/speculation/speculator/standard/README.md
index 02e9f5f7e..c5d869e65 100644
--- a/submitqueue/extension/speculation/speculator/standard/README.md
+++ b/submitqueue/extension/speculation/speculator/standard/README.md
@@ -4,7 +4,7 @@ The `standard` `Speculator` funds the queue's most promising speculation paths f
Each run it considers candidate paths in descending order of their probability of being the future that actually happens, and proposes builds down that ranking. Paths already pending or building keep the slot they hold rather than restarting; paths whose builds already finished are skipped for as long as their records remain in the supplied path sets, so a finished path can be proposed again — for a retry, say — once retention drops it; new builds fill whatever budget remains.
-When the budget runs out, everything below the cut waits for a later run. That is safe because a batch's verdict never depends on what was funded — only on how its dependencies resolve and which builds pass.
+When the budget runs out, everything below the cut waits for a later run. That is safe because the propose-side cannot invent a batch verdict: the speculate controller still decides merge from the persisted paths, including complete coverage of unsettled dependencies.
Both halves are swappable. The ranking is the `Generator`'s: the default `bestfirst` scores each path by the probability that all its assumptions hold. The budget policy is the `Allocator`'s: the default `sticky` fills only free slots and never preempts, where a preempting allocator would cancel a low-value in-flight path to fund a better one.
diff --git a/submitqueue/extension/storage/README.md b/submitqueue/extension/storage/README.md
index 21002ee37..2672851f8 100644
--- a/submitqueue/extension/storage/README.md
+++ b/submitqueue/extension/storage/README.md
@@ -48,7 +48,7 @@ A `Get` immediately following a successful write (`Create`/`Update`) — by the
## Key-value contract
-Store interfaces are designed for the storage technology *space*, not for SQL (see the Extensions section of the repo `CLAUDE.md`): every method must be satisfiable by a plain key-value backend (DynamoDB, Bigtable, an in-memory map) as cheaply as by MySQL. Concretely, a store exposes only get/put/conditional-update **by primary key**. No lookups by other attributes, no listings filtered server-side, no joins.
+Store interfaces are designed for the storage technology *space*, not for SQL (see the Extensions section of the repo [AGENTS.md](../../../../AGENTS.md)): every method must be satisfiable by a plain key-value backend (DynamoDB, Bigtable, an in-memory map) as cheaply as by MySQL. Concretely, a store exposes only get/put/conditional-update **by primary key**. No lookups by other attributes, no listings filtered server-side, no joins.
**The smell test is the index.** If implementing a proposed store method in MySQL requires adding a secondary index (`KEY idx_*`) to the schema, the method is a query-by-attribute in disguise and the contract has left the key-value space — a KV backend would need a global secondary index or a hand-maintained index table to fake it. Treat a new `KEY` line in a schema diff as a design review flag, not a tuning detail.
diff --git a/submitqueue/orchestrator/README.md b/submitqueue/orchestrator/README.md
index 0fd2cf1ee..c2405ffc1 100644
--- a/submitqueue/orchestrator/README.md
+++ b/submitqueue/orchestrator/README.md
@@ -1 +1,22 @@
-SubmitQueue Orchestrator
+# SubmitQueue Orchestrator
+
+The orchestrator runs the SubmitQueue land pipeline. It consumes the internal topics declared in `submitqueue/core/topickey/` and advances requests and batches through the stages that lead from `accepted` to a terminal state.
+
+## Pipeline stages
+
+The pipeline is queue-driven: each stage consumes one topic, advances one entity, and publishes to the next topic.
+
+- **start** — receives `LandRequest` from the gateway, persists the `Request` entity, and emits `Started`.
+- **validate** — checks for duplicates, resolves change metadata, and publishes a `MergeRequest` to Runway's `merge-conflict-check` topic.
+- **mergeconflictsignal** — correlates the dry-run result, fails the request on conflict, or forwards it to batching.
+- **batch** — groups the request into a `Batch` with its dependencies.
+- **speculate** — decides which speculative paths to validate (CI) versus land directly.
+- **build** — triggers a CI build for a speculative path.
+- **buildsignal** — records the CI result and loops back to `speculate`.
+- **merge** — publishes a committing `MergeRequest` to Runway's `runway-merge` topic.
+- **mergesignal** — correlates the merge result and fans out to `conclude` and back to `speculate`.
+- **conclude** — maps the terminal batch state to the request states.
+- **log** — persists gateway-owned request-log events published by the orchestrator.
+- **DLQ reconcilers** — one per primary consumed topic, driving stuck requests/batches to a conservative terminal `failed` state.
+
+See [doc/rfc/submitqueue/workflow.md](../../doc/rfc/submitqueue/workflow.md) for the full pipeline diagram and ownership rules.
diff --git a/submitqueue/orchestrator/controller/speculate/BUILD.bazel b/submitqueue/orchestrator/controller/speculate/BUILD.bazel
index 01b4113af..c0fd13583 100644
--- a/submitqueue/orchestrator/controller/speculate/BUILD.bazel
+++ b/submitqueue/orchestrator/controller/speculate/BUILD.bazel
@@ -3,6 +3,7 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
go_library(
name = "go_default_library",
srcs = [
+ "bypass.go",
"check.go",
"dispatch.go",
"doc.go",
diff --git a/submitqueue/orchestrator/controller/speculate/bypass.go b/submitqueue/orchestrator/controller/speculate/bypass.go
new file mode 100644
index 000000000..9177a2d46
--- /dev/null
+++ b/submitqueue/orchestrator/controller/speculate/bypass.go
@@ -0,0 +1,79 @@
+// Copyright (c) 2025 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package speculate
+
+import "github.com/uber/submitqueue/submitqueue/entity"
+
+// bypassablePath returns a passed path once passed builds cover every possible
+// outcome of the head's unsettled dependencies. Settled dependencies stay
+// pinned to reality through assumptionBroken.
+//
+// A path's combination is read positionally — the i-th assumption belongs to
+// the head's i-th dependency — which isWellFormed's order check is what
+// licenses: two paths with the same assumptions in different orders are
+// different stacks, not the same outcome.
+func bypassablePath(head entity.Batch, set entity.SpeculationPathSet, snap snapshot) (entity.SpeculationPathEntry, bool) {
+ unsettled := unsettledDependencyIndices(head, snap)
+ if len(unsettled) == 0 {
+ return entity.SpeculationPathEntry{}, false
+ }
+
+ required := 1
+ for range unsettled {
+ if required > len(set.Paths)/2 {
+ return entity.SpeculationPathEntry{}, false
+ }
+ required *= 2
+ }
+
+ seen := make(map[string]struct{}, required)
+ var winner entity.SpeculationPathEntry
+ for _, entry := range set.Paths {
+ if entry.Status != entity.SpeculationPathStatusPassed ||
+ assumptionBroken(entry.Path, snap) ||
+ !isWellFormed(entry.Path, head) {
+ continue
+ }
+
+ signature := make([]byte, len(unsettled))
+ for i, depIndex := range unsettled {
+ if entry.Path.Dependencies[depIndex].Assumption == entity.DependencyAssumptionFails {
+ signature[i] = 'f'
+ } else {
+ signature[i] = 's'
+ }
+ }
+ key := string(signature)
+ if _, exists := seen[key]; exists {
+ continue
+ }
+ seen[key] = struct{}{}
+ if len(seen) == 1 {
+ winner = entry
+ }
+ }
+
+ return winner, len(seen) == required
+}
+
+func unsettledDependencyIndices(head entity.Batch, snap snapshot) []int {
+ var indices []int
+ for i, depID := range head.Dependencies {
+ if !snap.batchState(depID).IsTerminal() {
+ indices = append(indices, i)
+ }
+ }
+ return indices
+}
diff --git a/submitqueue/orchestrator/controller/speculate/check.go b/submitqueue/orchestrator/controller/speculate/check.go
index f82cd8a21..14528c913 100644
--- a/submitqueue/orchestrator/controller/speculate/check.go
+++ b/submitqueue/orchestrator/controller/speculate/check.go
@@ -29,7 +29,7 @@ const (
rejectHeadNotSpeculating rejection = "head_not_speculating"
// rejectMalformedPath is a path whose assumptions do not line up with its
// head's dependency list: one missing or extra, a duplicate, a wrong head,
- // or a made-up assumption value.
+ // an assumption out of queue order, or a made-up assumption value.
rejectMalformedPath rejection = "malformed_path"
// rejectBrokenAssumption is a path with an assumption a finished
// dependency has already proven wrong.
@@ -122,7 +122,13 @@ func rejectionReason(proposal entity.Speculation, snap snapshot) (rejection, boo
// isWellFormed reports whether a path is a proper guess about its head:
// exactly one assumption for each of the head's dependencies, no more and no
-// fewer, and every assumption a real value.
+// fewer, in the head's dependency order, and every assumption a real value.
+//
+// Order is load-bearing: Base() projects the path positionally, so a path
+// whose dependencies are permuted describes a stack the build runner applied
+// in a different order — a different tree, which no verdict may count as the
+// combination its assumptions name. With the length check and position-wise
+// equality, a missing, extra, or duplicate dependency is also impossible.
//
// A malformed path is not merely suboptimal, it is unmergeable — the merge
// preconditions are read off the path's assumptions (see mergeablePath), so a
@@ -135,16 +141,10 @@ func isWellFormed(path entity.SpeculationPath, head entity.Batch) bool {
return false
}
- required := make(map[string]struct{}, len(head.Dependencies))
- for _, dep := range head.Dependencies {
- required[dep] = struct{}{}
- }
-
- for _, dep := range path.Dependencies {
- if _, ok := required[dep.Batch]; !ok {
+ for i, dep := range path.Dependencies {
+ if dep.Batch != head.Dependencies[i] {
return false
}
- delete(required, dep.Batch)
switch dep.Assumption {
case entity.DependencyAssumptionSucceeds,
@@ -154,7 +154,7 @@ func isWellFormed(path entity.SpeculationPath, head entity.Batch) bool {
}
}
- return len(required) == 0
+ return true
}
// findPath returns the entry for a path ID in the set.
diff --git a/submitqueue/orchestrator/controller/speculate/check_test.go b/submitqueue/orchestrator/controller/speculate/check_test.go
index b095f9274..ee2e25f5f 100644
--- a/submitqueue/orchestrator/controller/speculate/check_test.go
+++ b/submitqueue/orchestrator/controller/speculate/check_test.go
@@ -102,6 +102,18 @@ func TestFilterProposals_Rejects(t *testing.T) {
snap: checkSnapshot(entity.BatchStateSpeculating),
want: rejectMalformedPath,
},
+ {
+ name: "path with dependencies out of queue order",
+ proposal: entity.Speculation{
+ Path: entity.SpeculationPath{Head: head, Dependencies: []entity.PathDependency{
+ {Batch: dep2, Assumption: entity.DependencyAssumptionFails},
+ {Batch: dep1, Assumption: entity.DependencyAssumptionSucceeds},
+ }},
+ Action: entity.PathActionBuild,
+ },
+ snap: checkSnapshot(entity.BatchStateSpeculating),
+ want: rejectMalformedPath,
+ },
{
name: "cancel on a path that is not stored",
proposal: entity.Speculation{Path: valid, Action: entity.PathActionCancel},
@@ -208,12 +220,12 @@ func TestIsWellFormed(t *testing.T) {
want: true,
},
{
- name: "order does not matter",
+ name: "dependencies out of queue order",
path: entity.SpeculationPath{Head: head, Dependencies: []entity.PathDependency{
{Batch: dep2, Assumption: entity.DependencyAssumptionSucceeds},
{Batch: dep1, Assumption: entity.DependencyAssumptionFails},
}},
- want: true,
+ want: false,
},
{
name: "missing a dependency",
diff --git a/submitqueue/orchestrator/controller/speculate/doc.go b/submitqueue/orchestrator/controller/speculate/doc.go
index 1a5e82263..a8927cdbb 100644
--- a/submitqueue/orchestrator/controller/speculate/doc.go
+++ b/submitqueue/orchestrator/controller/speculate/doc.go
@@ -24,7 +24,8 @@
// speculation everything is serial: C waits for B, B waits for A. Speculation
// builds a batch against a guess about how its dependencies turn out. When
// the guess holds, the batch merges the moment the guessed-on dependencies
-// land — it never waits for a build of its own to start afterwards.
+// land. If passed paths cover every possible outcome, the batch can merge
+// before those dependencies settle.
//
// # Paths
//
@@ -46,6 +47,8 @@
//
// Fund both and every future is covered:
//
+// - While A is still unresolved, both P1 and P2 passing lets B bypass A and
+// merge immediately: either possible future has already been validated.
// - A succeeds and P1 passed: B merges the moment A lands. P2's guess
// ("A fails") is broken — it can no longer come true — so its build is
// cancelled to free the slot.
@@ -75,9 +78,8 @@
// building, and every pending, building, and cancelling path holds its slot
// until its build stops. A path is broken once a dependency's actual result
// proves one of its assumptions wrong: its guess can no longer come true, so
-// its build is cancelled to free the slot. A path is superseded when a
-// sibling path of the same head passes — that sibling will carry the head out
-// of the queue, so the others are cancelled too.
+// its build is cancelled to free the slot. A path is superseded when its head
+// becomes mergeable, so any still-running siblings are cancelled too.
//
// Cancelling is intent, not fact: the build keeps its slot until CI actually
// stops it, and only an observation of that stop (or proof nothing was ever
@@ -89,8 +91,8 @@
//
// # The life of a batch, as seen from here
//
-// Created ──admit──► Speculating ──┬── merge ──► Merging (merge stage takes over)
-// └── fail ───► Failed
+// Created ──admit──► Speculating ──┬── merge or bypass ──► Merging
+// └── fail ─────────────► Failed
// user cancel (cancel stage):
// ... ──► Cancelling ── every path stopped ──► Cancelled
//
diff --git a/submitqueue/orchestrator/controller/speculate/finalize.go b/submitqueue/orchestrator/controller/speculate/finalize.go
index 8096e0863..fb61e91ac 100644
--- a/submitqueue/orchestrator/controller/speculate/finalize.go
+++ b/submitqueue/orchestrator/controller/speculate/finalize.go
@@ -33,9 +33,9 @@ import (
// snap.speculating holding the heads still open to new work.
//
// Everything here is a fact, not a choice: a path a resolved dependency ruled
-// out is dead, a head whose passed build's assumptions all came true
-// merges, and a batch the user cancelled is finished once its last build
-// stops. Finalizing before the Speculator is asked is what keeps its work from
+// out is dead, a head whose passed builds establish a merge verdict merges,
+// and a batch the user cancelled is finished once its last build stops.
+// Finalizing before the Speculator is asked is what keeps its work from
// being wasted — asked first, it would propose builds for a head that is
// already merging.
//
@@ -98,7 +98,10 @@ func (c *Controller) finalize(ctx context.Context, snap *snapshot) error {
// The winning path carries the head out of the queue; its
// siblings cannot help it any more and are still holding CI
// slots the rest of the queue could use.
- winner, _ := mergeablePath(set, *snap)
+ winner, ok := mergeablePath(set, *snap)
+ if !ok {
+ winner, _ = bypassablePath(batch, set, *snap)
+ }
if supersede(&set, winner.ID, nowMs) {
snap.pathSets[batch.ID] = set
snap.markDirty(batch.ID)
@@ -449,9 +452,9 @@ func cancelBrokenPathsInSet(set *entity.SpeculationPathSet, snap snapshot, nowMs
})
}
-// supersede stops every path other than the winner. Once one path has passed,
-// its siblings cannot help the head any more — but they are still holding CI
-// slots the rest of the queue could use.
+// supersede stops every path other than the winner once the head can merge.
+// Its live siblings cannot help any more but still hold CI slots the rest of
+// the queue could use.
func supersede(set *entity.SpeculationPathSet, winnerID string, nowMs int64) bool {
return markCancelling(set, nowMs, func(entry entity.SpeculationPathEntry) bool {
return entry.ID == winnerID
diff --git a/submitqueue/orchestrator/controller/speculate/outcome.go b/submitqueue/orchestrator/controller/speculate/outcome.go
index 44f60bbb2..cd6c0c2c7 100644
--- a/submitqueue/orchestrator/controller/speculate/outcome.go
+++ b/submitqueue/orchestrator/controller/speculate/outcome.go
@@ -25,8 +25,9 @@ type outcome string
const (
// outcomeWait means the batch's outcome is not decided yet.
outcomeWait outcome = "wait"
- // outcomeMerge means a passed path's assumptions have all come true, so
- // the head can be handed to the merge stage.
+ // outcomeMerge means the head can be handed to the merge stage: either a
+ // passed path's assumptions have all come true, or passed paths cover every
+ // possible outcome of its unsettled dependencies.
outcomeMerge outcome = "merge"
// outcomeFail means no future remains in which the head could pass.
outcomeFail outcome = "fail"
@@ -54,6 +55,9 @@ func decide(head entity.Batch, set entity.SpeculationPathSet, snap snapshot) out
if _, ok := mergeablePath(set, snap); ok {
return outcomeMerge
}
+ if _, ok := bypassablePath(head, set, snap); ok {
+ return outcomeMerge
+ }
if hasNoViableFuture(head, set, snap) {
return outcomeFail
}
diff --git a/submitqueue/orchestrator/controller/speculate/outcome_test.go b/submitqueue/orchestrator/controller/speculate/outcome_test.go
index c21e6e43d..f1b6fd591 100644
--- a/submitqueue/orchestrator/controller/speculate/outcome_test.go
+++ b/submitqueue/orchestrator/controller/speculate/outcome_test.go
@@ -154,6 +154,137 @@ func TestMergeablePath_ExcludesBrokenPassedPath(t *testing.T) {
assert.False(t, ok)
}
+func TestBypassablePath(t *testing.T) {
+ const (
+ succeeds = entity.DependencyAssumptionSucceeds
+ fails = entity.DependencyAssumptionFails
+ )
+ allPaths := []entity.SpeculationPathEntry{
+ passedPath(succeeds, succeeds),
+ passedPath(succeeds, fails),
+ passedPath(fails, succeeds),
+ passedPath(fails, fails),
+ }
+ headBatch := entity.Batch{ID: head, Dependencies: []string{dep1, dep2}}
+
+ tests := []struct {
+ name string
+ head entity.Batch
+ set entity.SpeculationPathSet
+ dep1State entity.BatchState
+ dep2State entity.BatchState
+ want bool
+ }{
+ {
+ name: "bypasses when every outcome of two unsettled dependencies passed",
+ head: headBatch,
+ set: setOf(allPaths...),
+ dep1State: entity.BatchStateSpeculating,
+ dep2State: entity.BatchStateMerging,
+ want: true,
+ },
+ {
+ name: "waits when one outcome is missing",
+ head: headBatch,
+ set: setOf(allPaths[:3]...),
+ dep1State: entity.BatchStateSpeculating,
+ dep2State: entity.BatchStateSpeculating,
+ },
+ {
+ name: "covers only the unsettled dependency when another has succeeded",
+ head: headBatch,
+ set: setOf(
+ passedPath(succeeds, succeeds),
+ passedPath(succeeds, fails),
+ ),
+ dep1State: entity.BatchStateSucceeded,
+ dep2State: entity.BatchStateSpeculating,
+ want: true,
+ },
+ {
+ name: "does not count a path contradicted by a settled dependency",
+ head: headBatch,
+ set: setOf(
+ passedPath(succeeds, succeeds),
+ passedPath(fails, fails),
+ ),
+ dep1State: entity.BatchStateSucceeded,
+ dep2State: entity.BatchStateSpeculating,
+ },
+ {
+ name: "does not count a duplicate outcome twice",
+ head: headBatch,
+ set: setOf(
+ passedPath(succeeds, succeeds),
+ passedPath(succeeds, succeeds),
+ passedPath(fails, succeeds),
+ passedPath(fails, fails),
+ ),
+ dep1State: entity.BatchStateSpeculating,
+ dep2State: entity.BatchStateSpeculating,
+ },
+ {
+ name: "does not count an unpassed outcome",
+ head: headBatch,
+ set: setOf(
+ allPaths[0],
+ allPaths[1],
+ allPaths[2],
+ entryFor(pathOver(fails, fails), entity.SpeculationPathStatusFailed),
+ ),
+ dep1State: entity.BatchStateSpeculating,
+ dep2State: entity.BatchStateSpeculating,
+ },
+ {
+ name: "does not count a malformed path",
+ head: headBatch,
+ set: setOf(
+ allPaths[0],
+ allPaths[1],
+ allPaths[2],
+ passedPath(fails),
+ ),
+ dep1State: entity.BatchStateSpeculating,
+ dep2State: entity.BatchStateSpeculating,
+ },
+ {
+ // A path stacked [dep2, dep1] built a different tree than [dep1, dep2]:
+ // the base feeds the runner in path order (see build.loadBase). It
+ // cannot stand in for the canonical combination its assumptions name.
+ name: "does not count a reordered path",
+ head: headBatch,
+ set: setOf(
+ passedPath(succeeds, succeeds),
+ passedPath(succeeds, fails),
+ passedPath(fails, succeeds),
+ entryFor(entity.SpeculationPath{Head: head, Dependencies: []entity.PathDependency{
+ {Batch: dep2, Assumption: fails},
+ {Batch: dep1, Assumption: fails},
+ }}, entity.SpeculationPathStatusPassed),
+ ),
+ dep1State: entity.BatchStateSpeculating,
+ dep2State: entity.BatchStateSpeculating,
+ },
+ {
+ name: "leaves fully settled dependencies to strict merge",
+ head: headBatch,
+ set: setOf(allPaths...),
+ dep1State: entity.BatchStateSucceeded,
+ dep2State: entity.BatchStateFailed,
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ winner, ok := bypassablePath(tt.head, tt.set, snapWith(tt.dep1State, tt.dep2State))
+ assert.Equal(t, tt.want, ok)
+ if tt.want {
+ assert.NotEmpty(t, winner.ID)
+ }
+ })
+ }
+}
+
// livePassedPath is mergeablePath without the settled requirement, and the gap
// between the two is the head's waiting room: its own work is done and all that
// remains is other batches finishing. That window is reported to the members,
@@ -279,6 +410,15 @@ func TestDecide(t *testing.T) {
// A passed path wins over a failed sibling: one way through is enough.
assert.Equal(t, outcomeMerge, decide(headBatch, setOf(failed, passed), allResolved))
+
+ allUnresolved := snapWith(entity.BatchStateSpeculating, entity.BatchStateSpeculating)
+ fullCoverage := setOf(
+ passedPath(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionSucceeds),
+ passedPath(entity.DependencyAssumptionSucceeds, entity.DependencyAssumptionFails),
+ passedPath(entity.DependencyAssumptionFails, entity.DependencyAssumptionSucceeds),
+ passedPath(entity.DependencyAssumptionFails, entity.DependencyAssumptionFails),
+ )
+ assert.Equal(t, outcomeMerge, decide(headBatch, fullCoverage, allUnresolved))
}
// Once a path has passed, its siblings cannot help the head but are still
diff --git a/submitqueue/orchestrator/controller/speculate/run_test.go b/submitqueue/orchestrator/controller/speculate/run_test.go
index 306373640..0ae432fbe 100644
--- a/submitqueue/orchestrator/controller/speculate/run_test.go
+++ b/submitqueue/orchestrator/controller/speculate/run_test.go
@@ -1549,6 +1549,37 @@ func TestRun_MergingHeadReportsSpeculatedAndNoWait(t *testing.T) {
assert.Equal(t, head, h.logs[0].Metadata["batch_id"])
}
+func TestRun_BypassesUnsettledDependenciesWithFullCoverage(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ const (
+ succeeds = entity.DependencyAssumptionSucceeds
+ fails = entity.DependencyAssumptionFails
+ )
+
+ h := newRunHarness(t, ctrl, &scriptedSpeculator{}, []entity.Batch{memberHead()})
+ h.batches.EXPECT().Get(gomock.Any(), dep1).Return(entity.Batch{ID: dep1, State: entity.BatchStateSpeculating}, nil)
+ h.batches.EXPECT().Get(gomock.Any(), dep2).Return(entity.Batch{ID: dep2, State: entity.BatchStateMerging}, nil)
+ h.pathSets.EXPECT().Get(gomock.Any(), head).Return(entity.SpeculationPathSet{
+ Head: head,
+ Paths: []entity.SpeculationPathEntry{
+ passedPath(succeeds, succeeds),
+ passedPath(succeeds, fails),
+ passedPath(fails, succeeds),
+ passedPath(fails, fails),
+ },
+ Version: 1,
+ }, nil).AnyTimes()
+ h.batches.EXPECT().
+ Update(gomock.Any(), updateTo{id: head, state: entity.BatchStateMerging}, int32(1), int32(2)).Return(nil)
+
+ require.NoError(t, h.run(head))
+
+ assert.Equal(t, []string{"submitqueue-merge"}, h.published)
+ assert.Zero(t, h.spec.calls)
+ require.Len(t, h.logs, 1)
+ assert.Equal(t, entity.RequestStatusSpeculated, h.logs[0].Status)
+}
+
// The merge stage publishes landing as its first act on the dispatch. Both
// statuses are non-terminal, so the summary is decided on timestamp alone and
// a speculated sent afterwards would beat the landing it precedes.
diff --git a/test/e2e/submitqueue/BUILD.bazel b/test/e2e/submitqueue/BUILD.bazel
index 2733aa86a..680b0f6d3 100644
--- a/test/e2e/submitqueue/BUILD.bazel
+++ b/test/e2e/submitqueue/BUILD.bazel
@@ -48,6 +48,7 @@ go_test(
"//submitqueue/core/batch:go_default_library",
"//submitqueue/core/topickey:go_default_library",
"//submitqueue/entity:go_default_library",
+ "//submitqueue/extension/storage:go_default_library",
"//submitqueue/extension/storage/mysql:go_default_library",
"//test/testutil:go_default_library",
"@com_github_stretchr_testify//assert:go_default_library",
diff --git a/test/e2e/submitqueue/harness_test.go b/test/e2e/submitqueue/harness_test.go
index cb149ae5a..4ee147f49 100644
--- a/test/e2e/submitqueue/harness_test.go
+++ b/test/e2e/submitqueue/harness_test.go
@@ -42,6 +42,7 @@ import (
corebatch "github.com/uber/submitqueue/submitqueue/core/batch"
"github.com/uber/submitqueue/submitqueue/core/topickey"
"github.com/uber/submitqueue/submitqueue/entity"
+ "github.com/uber/submitqueue/submitqueue/extension/storage"
"go.uber.org/zap"
)
@@ -321,6 +322,39 @@ func (s *E2EIntegrationSuite) awaitBatchState(queue, batchID string, want entity
})
}
+// seedPassedPath writes one already-passed speculation path directly, so a
+// test can hold the other side of a dependency back deterministically.
+func (s *E2EIntegrationSuite) seedPassedPath(queue string, path entity.SpeculationPath) {
+ t := s.T()
+ store, err := s.appStorage.For(queue)
+ require.NoError(t, err, "failed to resolve operating store for queue %s", queue)
+ pathSets := store.GetSpeculationPathSetStore()
+
+ set, err := pathSets.Get(s.ctx, path.Head)
+ if storage.IsNotFound(err) {
+ set = entity.SpeculationPathSet{Queue: queue, Head: path.Head}
+ } else {
+ require.NoError(t, err, "failed to read path set for %s", path.Head)
+ }
+
+ set.Paths = append(set.Paths, entity.SpeculationPathEntry{
+ ID: path.ID(),
+ Path: path,
+ Status: entity.SpeculationPathStatusPassed,
+ // Attempt 1 is the first build of a path and matches every entry the
+ // speculate run itself writes.
+ Attempt: 1,
+ Version: 1,
+ })
+ newVersion := set.Version + 1
+ if set.Version == 0 {
+ require.NoError(t, pathSets.Create(s.ctx, set), "failed to create path set for %s", path.Head)
+ } else {
+ require.NoError(t, pathSets.Update(s.ctx, set, set.Version, newVersion),
+ "failed to seed passed path %s for %s", path.ID(), path.Head)
+ }
+}
+
// strandInCreated puts a batch back into Created, reproducing the state a batch
// is left in when it is promoted but its announcement never reaches speculate.
// Nothing will name it on the speculate topic again, so only a run that looks
diff --git a/test/e2e/submitqueue/suite_test.go b/test/e2e/submitqueue/suite_test.go
index 7097647d6..e54ef60da 100644
--- a/test/e2e/submitqueue/suite_test.go
+++ b/test/e2e/submitqueue/suite_test.go
@@ -276,34 +276,10 @@ func (s *E2EIntegrationSuite) TestLand_HappyPath_ReachesLanded() {
"operating store should show request %s in terminal state landed", req.sqid)
}
-// TestDependentBatch_IsWokenByTheMergeAhead proves that a batch waiting on
-// another is woken when that one merges — the edge CODEM-303 was silently
-// dropping.
-//
-// A merged batch fans out to speculate so its dependents can re-plan. That
-// message used to reuse the bare batch ID, which the batch controller had
-// already published to the same topic and partition when the batch was
-// created. The queue deduplicates against rows it has not collected yet,
-// consumed ones included, so the wake-up was reported as a success, stored
-// nothing, and never arrived.
-//
-// Ordinarily something else re-plans the queue soon enough to hide that. This
-// test removes every other source of a wake-up, as stop → observe → start:
-//
-// 1. Stop: close the gate for runway-merge on this queue, before landing, so
-// the lead batch cannot complete its merge.
-// 2. Land the lead. It runs to the merge hand-off and parks there.
-// 3. Land the dependent. The queue's analyzer serializes conservatively, so
-// its batch depends on the lead's, which is in-flight (Merging counts).
-// 4. Observe: wait for the dependent to record "waiting" — its speculative
-// build has already passed, so its own build signals are finished. From
-// here the only thing that can advance it is the lead merging.
-// 5. Start: open the gate. The lead merges and fans out.
-//
-// The dependent reaching "landed" is therefore attributable to the fan-out
-// alone. Against the old code it rests at "speculating" and the suite runs to
-// Bazel's timeout, which is how the harness reports a pipeline that stalled.
-func (s *E2EIntegrationSuite) TestDependentBatch_IsWokenByTheMergeAhead() {
+// TestDependentBatch_BypassesMergingDependency proves that a dependency still
+// waiting on Runway is unresolved for strict merge but can be bypassed once
+// passed paths cover both of its possible outcomes.
+func (s *E2EIntegrationSuite) TestDependentBatch_BypassesMergingDependency() {
t := s.T()
const queue = "e2e-chain-queue"
@@ -337,12 +313,11 @@ func (s *E2EIntegrationSuite) TestDependentBatch_IsWokenByTheMergeAhead() {
require.Contains(t, got.Dependencies, leadBatch,
"batch %s must depend on the in-flight %s for this test to exercise anything", dependentBatch, leadBatch)
- // Its speculative build passes while the lead is still parked, so by the
- // time the gate opens the dependent has no build signals left to wake it.
- // That rest is an event, not a status: a batch blocked on a dependency has
- // not finished speculating, so it stays "speculating" until it can merge.
- s.awaitEvent(dependent, entity.RequestEventWaiting)
- s.log.Logf("Dependent %s has passed its build and waits only on %s", dependent.sqid, leadBatch)
+ // Both paths pass while the lead is still parked. A Merging dependency is
+ // unresolved because its merge can fail, so the durable Merging state proves
+ // the dependent advanced through complete coverage rather than strict merge.
+ s.awaitBatchState(queue, dependentBatch, entity.BatchStateMerging)
+ s.log.Logf("Dependent %s bypassed merging batch %s", dependent.sqid, leadBatch)
// Start: the lead merges, and its fan-out is now the only thing that can
// move the dependent.
@@ -352,8 +327,100 @@ func (s *E2EIntegrationSuite) TestDependentBatch_IsWokenByTheMergeAhead() {
s.awaitStatus(lead, entity.RequestStatusLanded)
s.awaitStatus(dependent, entity.RequestStatusLanded)
- assert.Equal(t, entity.RequestStateLanded, s.terminalState(dependent),
- "the dependent must land once the batch it waited on merged")
+ assert.Equal(t, entity.RequestStateLanded, s.terminalState(dependent))
+}
+
+// TestDependentBatch_BypassedHeadLandsFirst proves the bypass does not just
+// dispatch a merge: the dependent lands while its dependency is still held
+// mid-build, and only then does the dependency proceed.
+func (s *E2EIntegrationSuite) TestDependentBatch_BypassedHeadLandsFirst() {
+ t := s.T()
+
+ const queue = "e2e-chain-queue"
+ const gateGroup = "orchestrator"
+
+ lead := s.land(queue, "github://github.example.com/uber/e2e-bypass/pull/1/aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
+ heldBatch := s.awaitBatchID(lead)
+ s.closeGate(gateGroup, heldBatch, "e2e: hold the leader's build so the follower can fully cover it")
+ defer s.openGate(gateGroup, heldBatch)
+ s.awaitBatchState(queue, heldBatch, entity.BatchStateSpeculating)
+
+ follower := s.land(queue, "github://github.example.com/uber/e2e-bypass/pull/2/bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
+ followerBatch := s.awaitBatchID(follower)
+ require.NotEqual(t, heldBatch, followerBatch)
+
+ // The follower builds with and without the held leader while the leader's
+ // own build is parked; complete coverage hands it to the merge stage.
+ s.awaitBatchState(queue, followerBatch, entity.BatchStateMerging)
+ s.awaitStatus(follower, entity.RequestStatusLanded)
+ assert.Equal(t, entity.RequestStatusSpeculating, s.mustStatus(lead),
+ "the follower must land while its dependency is still held")
+
+ s.openGate(gateGroup, heldBatch)
+ s.awaitStatus(lead, entity.RequestStatusLanded)
+
+ s.assertStatusesInOrder(follower,
+ entity.RequestStatusSpeculating,
+ entity.RequestStatusSpeculated,
+ entity.RequestStatusLanding,
+ entity.RequestStatusLanded,
+ )
+}
+
+// TestDependentBatch_NoBypassWhenCoverageIsIncomplete proves the complement:
+// with only the "dependency succeeds" side passed, a head waits for its
+// dependency to resolve and merges strictly, never dispatching ahead of it.
+//
+// Partial coverage is seeded directly: the follower's batch is stranded in
+// Created (build held), its "succeeds" path is written as passed, and a
+// trigger request wakes the queue. The follower's real speculative builds are
+// parked on its held batch partition, so the funded set stays exactly one
+// path. The single passed path is a live passed path, so the run reports the
+// wait — the signal this test then uses to prove nothing else advanced it.
+func (s *E2EIntegrationSuite) TestDependentBatch_NoBypassWhenCoverageIsIncomplete() {
+ t := s.T()
+
+ const queue = "e2e-chain-queue"
+ const gateGroup = "orchestrator"
+
+ lead := s.land(queue, "github://github.example.com/uber/e2e-nobypass/pull/1/cccccccccccccccccccccccccccccccccccccccc")
+ leadBatch := s.awaitBatchID(lead)
+ s.awaitBatchState(queue, leadBatch, entity.BatchStateSpeculating)
+
+ follower := s.land(queue, "github://github.example.com/uber/e2e-nobypass/pull/2/dddddddddddddddddddddddddddddddddddddddd")
+ heldBatch := s.awaitBatchID(follower)
+ s.closeGate(gateGroup, heldBatch, "e2e: hold the follower's builds so only the seeded path exists")
+ defer s.openGate(gateGroup, heldBatch)
+
+ // Strand the follower in Created, then seed exactly one passed path: the
+ // guess that the lead succeeds. Its builds are parked, so nothing can add
+ // the "fails" path while the queue is quiet.
+ s.strandInCreated(queue, heldBatch)
+ s.seedPassedPath(queue, entity.SpeculationPath{
+ Head: heldBatch,
+ Dependencies: []entity.PathDependency{
+ {Batch: leadBatch, Assumption: entity.DependencyAssumptionSucceeds},
+ },
+ })
+
+ trigger := s.land(queue, "github://github.example.com/uber/e2e-nobypass/pull/3/eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee")
+ s.awaitBatchID(trigger)
+
+ // The run that admits the follower also reports the wait: its one passed
+ // path covers only one of the lead's two outcomes. The follower's batch
+ // must still be Speculating — incomplete coverage must never hand it to
+ // the merge stage ahead of the lead.
+ s.awaitEvent(follower, entity.RequestEventWaiting)
+ assert.Equal(t, entity.BatchStateSpeculating, s.batchState(queue, heldBatch),
+ "incomplete coverage must not move the batch to merging")
+
+ // Let the queue finish. The follower's own speculative builds were parked
+ // on the held partition; releasing it lets the surviving path complete and
+ // land after the lead. How it ultimately converges is exercised by the
+ // other tests; this one exists to prove the wait, not the landing.
+ s.openGate(gateGroup, heldBatch)
+ s.awaitStatus(lead, entity.RequestStatusLanded)
+ s.awaitStatus(trigger, entity.RequestStatusLanded)
}
// TestReadAPIs validates all five request read endpoints against receipts
@@ -434,23 +501,16 @@ func (s *E2EIntegrationSuite) TestReadAPIs() {
assert.Equal(t, secondSummary.Request.LastError, secondEvents[len(secondEvents)-1].LastError)
}
-// TestLand_DependentBatch_StaysSpeculatingAcrossAnUnresolvedDependency covers
-// the oscillation the request log used to report as a regression: a head
-// speculates while a dependency is unresolved, the dependency then fails, and
-// the head re-plans and lands anyway. Throughout, it is speculating exactly
-// once — the trail must never revisit a stage.
+// TestLand_DependentBatch_BypassesAnUnresolvedDependency proves the complete
+// payoff: a follower built both with and without a held leader lands before the
+// leader resolves.
//
// The wait is forced rather than raced. Batch IDs come from a per-queue counter
// as "/batch/", so the leader on a fresh queue is batch/1, and the
// build topic partitions by batch — closing the gate on that partition before
// anything is published holds the leader's build and nothing else, so the
// follower reaches a passed path while its dependency is still outstanding.
-//
-// No invalidated event is asserted. A passed path stops occupying build budget,
-// so by the time the leader fails the follower has usually funded the other side
-// of the guess too; it never loses its last live passed path, which is what
-// invalidated reports. The unit tests cover that state directly.
-func (s *E2EIntegrationSuite) TestLand_DependentBatch_StaysSpeculatingAcrossAnUnresolvedDependency() {
+func (s *E2EIntegrationSuite) TestLand_DependentBatch_BypassesAnUnresolvedDependency() {
const queue = "e2e-respeculate-queue"
const gateGroup = "orchestrator"
leaderBatch := queue + "/batch/1"
@@ -462,21 +522,18 @@ func (s *E2EIntegrationSuite) TestLand_DependentBatch_StaysSpeculatingAcrossAnUn
follower := s.land(queue, "github://github.example.com/uber/e2e-respeculate/pull/2/2222222222222222222222222222222222222222")
s.log.Logf("Landed leader=%s (build held) follower=%s", leader.sqid, follower.sqid)
- // The baseline analyzer serializes the queue, so the follower depends on the
- // leader and speculates on it succeeding. That build passes while the leader
- // is still held: the follower's own work is done and only the leader is
- // outstanding, which is the wait.
- s.awaitEvent(follower, entity.RequestEventWaiting)
- assert.Equal(s.T(), entity.RequestStatusSpeculating, s.mustStatus(follower),
- "a head waiting on its dependency has not finished speculating")
+ // The baseline analyzer serializes the queue, and the build budget lets the
+ // follower validate both possible outcomes while the leader is held.
+ s.awaitStatus(follower, entity.RequestStatusLanded)
+ assert.Equal(s.T(), entity.RequestStatusSpeculating, s.mustStatus(leader),
+ "the follower must land before the held leader resolves")
- // Release the leader. Its build fails, contradicting the guess the follower
- // speculated on, and the follower has to reach the trunk another way.
+ // Release the leader only after the follower has landed. Its later failure is
+ // one of the outcomes the follower already validated.
s.openGate(gateGroup, leaderBatch)
assert.Equal(s.T(), entity.RequestStatusError, s.awaitTerminal(leader),
"the leader's build carries a failure marker, so it must not land")
- s.awaitStatus(follower, entity.RequestStatusLanded)
s.assertStatusesInOrder(follower,
entity.RequestStatusSpeculating,
entity.RequestStatusSpeculated,