From a87a5864f04aaa4a6fe128c26ca926f80235836b Mon Sep 17 00:00:00 2001 From: jamy Date: Tue, 30 Jun 2026 15:47:14 +0000 Subject: [PATCH 1/9] docs(submitqueue): add modular queue wiring RFC Propose extracting per-queue extension profiles, topic-registry builder, and controller registration from the orchestrator example main.go into reusable domain packages, plus data-driven profile hints in QueueConfig. Co-Authored-By: Claude Opus 4.6 --- doc/rfc/index.md | 1 + doc/rfc/submitqueue/modular-queue-wiring.md | 242 ++++++++++++++++++++ 2 files changed, 243 insertions(+) create mode 100644 doc/rfc/submitqueue/modular-queue-wiring.md diff --git a/doc/rfc/index.md b/doc/rfc/index.md index 571bdb4c..0c4bf92a 100644 --- a/doc/rfc/index.md +++ b/doc/rfc/index.md @@ -16,6 +16,7 @@ Design documents and technical proposals, grouped by scope. Shared/cross-cutting - [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 - [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 +- [Modular Queue Wiring](submitqueue/modular-queue-wiring.md) - Extract per-queue extension profiles, topic-registry builder, and controller registration into reusable domain packages; add data-driven profile hints to QueueConfig ## Stovepipe diff --git a/doc/rfc/submitqueue/modular-queue-wiring.md b/doc/rfc/submitqueue/modular-queue-wiring.md new file mode 100644 index 00000000..fbe6d233 --- /dev/null +++ b/doc/rfc/submitqueue/modular-queue-wiring.md @@ -0,0 +1,242 @@ +# Modular Queue Wiring + +Design notes for making the orchestrator's per-queue extension wiring and topic-registry setup modular, data-driven, and reusable across deployers. Decisions and rationale only; the code changes land after this RFC is reviewed. + +## Problem + +The orchestrator's example `main.go` (`example/submitqueue/orchestrator/server/main.go`) is ~950 lines that mixes three distinct concerns: + +1. **Infrastructure bootstrap** — DB connections, logger, metrics, gRPC server, signal handling (~200 lines of generic boilerplate, much of it duplicated between gateway and orchestrator). +2. **Queue topology / topic registry** — `newTopicRegistry` is a static list of 12+ pipeline stages, each with a primary subscription and a mirrored DLQ subscription, plus publish-only topics. Adding or removing a pipeline stage requires editing this function in lockstep with controller registration. +3. **Per-queue extension wiring** — `queueRegistry`, `newQueueRegistry`, and four thin `*Factory` adapter types. The only way to configure which scorer / analyzer / change-provider / build-runner a queue uses is to edit Go code in this file, recompile, and redeploy. + +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). This makes adoption harder for new integrators who want to deploy SubmitQueue with their own queues and extension profiles. + +The [TODO on line 477](../../example/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. + +## Principle + +- **The wiring layer assembles; the domain layer provides reusable building blocks.** Today the domain layer owns controllers, entities, and extension interfaces — but the *composition* of extensions into per-queue profiles and the *registration* of controllers into consumers is copy-pasted into each deployer's main.go. These compositions are mechanical and identical across deployers; they belong in the domain. +- **Data-driven where practical, code-driven where necessary.** Queue *names* are already data-driven (YAML via `queueconfig.Store`). Queue *extension profiles* — which scorer, which conflict analyzer — should also be declarable as data. Custom extension *implementations* remain code (a deployer writes a new `scorer.Scorer` impl), but selecting among known implementations should not require a recompile. +- **No DI framework.** The wiring stays explicit Go code. This refactor reduces its volume, not its nature. + +## Proposal + +### 1. Promote queue-profile registry into `submitqueue/core/queueprofile` + +Extract the `queueExtensions` struct (renamed `Profile`) and `queueRegistry` (renamed `Registry`) from the example into `submitqueue/core/queueprofile/`. This is the domain-internal analogue of `submitqueue/core/topickey` — infrastructure shared between the orchestrator and (potentially) future services, but private to the SubmitQueue domain. + +```go +// submitqueue/core/queueprofile/profile.go +package queueprofile + +// Profile is the full set of extension implementations for a single queue. +// Grouping per queue (rather than per extension) lets the wiring read as +// "for this queue, here are its scorer, analyzer, change provider, …" +// and lets a profile start from a baseline and override only what differs. +type Profile struct { + // ChangeProvider resolves change metadata for land requests in this queue. + ChangeProvider changeprovider.ChangeProvider + + // BuildRunner triggers and polls CI builds for batches in this queue. + BuildRunner buildrunner.BuildRunner + + // Scorer computes success probability for batches in this queue. + Scorer scorer.Scorer + + // Analyzer detects conflicts between batches in this queue. + Analyzer conflict.Analyzer +} + +// Registry maps a queue name to its Profile, falling back to a default +// for queues without an explicit entry. It is the single place that knows +// the queue topology; extension packages remain queue-agnostic. +type Registry struct { … } + +func NewRegistry(def Profile, perQueue map[string]Profile) Registry +func (r Registry) Get(queue string) Profile +``` + +The four thin factory adapters (`changeProviderFactory`, `buildRunnerFactory`, `scorerFactory`, `analyzerFactory`) also move into this package as exported types. They are mechanical — `For(cfg) → registry.Get(cfg.QueueName).X` — and every deployer needs them identically. + +**Why a new package instead of expanding `queueconfig`:** `queueconfig` is a resolution target (key/value store of queue names). The profile registry is a *consumer* of queue names that additionally bundles behavioral extension instances. Mixing them would give `queueconfig` a dependency on every extension interface, violating the "stores are resolution targets, not aggregators" principle from CLAUDE.md. + +### 2. Extract topic-registry builder into `submitqueue/core/topicregistry` + +Replace `newTopicRegistry` with a reusable builder that declaratively constructs primary + DLQ topic pairs from a slice of stage specs, plus publish-only topics. + +```go +// submitqueue/core/topicregistry/builder.go +package topicregistry + +// StageSpec declares one pipeline stage that needs a primary subscription +// and an auto-generated DLQ subscription. +type StageSpec struct { + // Key is the consumer.TopicKey for this stage. + Key consumer.TopicKey + + // Name is the wire topic name (e.g. "start", "batch"). + Name string + + // GroupSuffix is the consumer-group suffix (e.g. "orchestrator-start"). + GroupSuffix string +} + +// BuildParams configures the topic registry builder. +type BuildParams struct { + // Queue is the message queue backend. + Queue extqueue.Queue + + // SubscriberName identifies this subscriber for partition leases. + SubscriberName string + + // Stages are the primary pipeline stages. Each gets a paired DLQ + // subscription with DLQ disabled (no _dlq_dlq cascade) and a high + // MaxAttempts for convergent reconciliation. + Stages []StageSpec + + // PublishOnly are topics the service publishes to but never consumes. + PublishOnly []consumer.TopicConfig +} + +func Build(p BuildParams) (consumer.TopicRegistry, error) +``` + +This eliminates the manual duplication of the primary/DLQ pairing pattern across 12 stages. Each stage is one `StageSpec` entry; the builder guarantees every primary stage gets a correctly-configured DLQ subscription (disabled DLQ-of-DLQ, high MaxAttempts) without copy-paste. Adding or removing a pipeline stage becomes adding or removing one line. + +### 3. Extract controller-registration helpers into `submitqueue/orchestrator/controller/wire` + +Create a `wire` subpackage under `submitqueue/orchestrator/controller/` with two functions: + +```go +// submitqueue/orchestrator/controller/wire/wire.go +package wire + +// PrimaryParams holds the dependencies needed to construct and register +// all primary pipeline controllers. +type PrimaryParams struct { + Consumer consumer.Consumer + Logger *zap.SugaredLogger + Scope tally.Scope + Registry consumer.TopicRegistry + ChangeProviderF changeprovider.Factory + BuildRunnerF buildrunner.Factory + ScorerF scorer.Factory + ConflictF conflict.Factory + Counter counter.Counter + Store storage.Storage +} + +// RegisterPrimary creates and registers all primary pipeline controllers. +// Returns the count of registered controllers. +func RegisterPrimary(p PrimaryParams) (int, error) + +// DLQParams holds the dependencies needed to construct and register +// all DLQ reconciliation controllers. +type DLQParams struct { + Consumer consumer.Consumer + Logger *zap.SugaredLogger + Scope tally.Scope + Store storage.Storage +} + +// RegisterDLQ creates and registers all DLQ reconciliation controllers. +// Returns the count of registered controllers. +func RegisterDLQ(p DLQParams) (int, error) +``` + +This keeps the controller list in the domain layer (testable, importable) and reduces the wiring main.go to: build dependencies → call `wire.RegisterPrimary` → call `wire.RegisterDLQ`. Adding a new pipeline stage becomes a single-file edit in this package. + +### 4. Extend `QueueConfig` with optional profile hints + +Add an optional `Profile` field to `entity.QueueConfig` so the YAML file can declare which scorer / conflict / build-runner strategy each queue uses: + +```yaml +queues: + - name: test-queue + profile: + scorer: heuristic + conflict: file-overlap + - name: e2e-test-queue + profile: + scorer: composite + conflict: none + - name: e2e-cancel-queue + # No profile — inherits the baseline. +``` + +```go +// submitqueue/entity/queue_config.go +type QueueConfig struct { + // Name uniquely identifies this queue within the system. + Name string `json:"name" yaml:"name"` + + // Profile carries optional hints for which extension implementations + // this queue uses. The wiring layer maps hint strings to concrete + // extension instances; the entity does not import extension packages. + // Zero value means "use the deployer's baseline profile." + Profile QueueProfile `json:"profile,omitempty" yaml:"profile,omitempty"` +} + +// QueueProfile carries string-typed hints for extension selection. +// Each field names a known implementation (e.g. "heuristic", "composite", +// "file-overlap", "none", "all"). Deployers register the mapping from +// hint → implementation in the wiring layer. An empty string means +// "inherit from the baseline." +type QueueProfile struct { + // Scorer names the scoring strategy (e.g. "heuristic", "composite"). + Scorer string `json:"scorer,omitempty" yaml:"scorer,omitempty"` + + // Conflict names the conflict-analysis strategy (e.g. "all", "none", "file-overlap"). + Conflict string `json:"conflict,omitempty" yaml:"conflict,omitempty"` + + // BuildRunner names the build-runner backend (e.g. "fake", "jenkins"). + BuildRunner string `json:"build_runner,omitempty" yaml:"build_runner,omitempty"` + + // ChangeProvider names the change-provider backend (e.g. "github", "phabricator", "routing"). + ChangeProvider string `json:"change_provider,omitempty" yaml:"change_provider,omitempty"` +} +``` + +**Constraints:** `QueueConfig` stays a simple data carrier — it does NOT import extension packages. The mapping from hint string → extension instance remains in the wiring layer (`example/.../main.go` or a deployer's equivalent). This preserves the clean architecture boundary: entities are pure, factories are injected. + +### 5. Refactor the example orchestrator main.go + +Rewrite the example to compose the new packages: + +1. Build infrastructure (DB, logger, metrics) — stays in main.go (deployment-specific). +2. Load queue configs from YAML — stays in main.go. +3. Build queue profiles using `queueprofile.NewRegistry(...)` — stays in main.go but is now ~30 lines instead of ~100, and can optionally be driven by profile hints from the YAML. +4. Build topic registry using `topicregistry.Build(...)` — one call, ~10 lines instead of ~90. +5. Register controllers using `wire.RegisterPrimary(...)` + `wire.RegisterDLQ(...)` — two calls instead of ~170 lines. +6. Start consumers and gRPC server — stays in main.go. + +**Result:** main.go drops from ~950 to ~300 lines. The domain-layer packages are independently testable. A new integrator copies the example, edits `queues.yaml` (including optional profile hints), and optionally customizes the `queueprofile.Registry` population — no need to understand the full pipeline topology. + +## What each extraction produces + +| Extraction | New package | Key types | Lines removed from main.go | +|---|---|---|---| +| Queue profiles | `submitqueue/core/queueprofile` | `Profile`, `Registry`, `ChangeProviderFactory`, `BuildRunnerFactory`, `ScorerFactory`, `AnalyzerFactory` | ~100 (queueExtensions, queueRegistry, 4 factory types, newQueueRegistry) | +| Topic registry | `submitqueue/core/topicregistry` | `StageSpec`, `BuildParams`, `Build()` | ~90 (newTopicRegistry) | +| Controller wire | `submitqueue/orchestrator/controller/wire` | `PrimaryParams`, `DLQParams`, `RegisterPrimary()`, `RegisterDLQ()` | ~170 (registerPrimaryControllers, registerDLQControllers) | +| Profile hints | `submitqueue/entity` (extended) | `QueueProfile` (added to `QueueConfig`) | 0 (additive) | + +## Rejected + +- **DI framework (wire/dig/fx).** Adds indirection and a build-time dependency for a problem that explicit code solves. The refactor reduces the volume of explicit wiring, not its nature. +- **Hot-reload of queue configs.** Out of scope. The YAML is loaded at startup. Hot-reload can build on this foundation later — `queueconfig.Store` already abstracts the read path, so swapping the YAML impl for a watching impl is a future, independent change. +- **Changing the Factory interface contract.** The existing `Factory.For(Config)` pattern is sound and is the way controllers resolve per-queue extension instances. We add a first-class registry that factories resolve *against*, not a new factory contract. +- **Promoting `newChangeProvider` / `newGitHubChangeProvider` / `newPhabChangeProvider` out of the example.** These are deployment-specific (token sources, HTTP clients, timeouts). They stay in the wiring layer. +- **Merging `queueprofile` into `queueconfig`.** The config store is a resolution target (key/value); the profile registry aggregates behavioral instances. Mixing them gives `queueconfig` a dependency on every extension interface, violating the "stores are resolution targets" principle. +- **A generic "service bootstrap" package.** The duplicated boilerplate between gateway and orchestrator (logger, metrics, DB, gRPC server, signal handling) is real but is a separate, orthogonal concern. Folding it into this RFC would conflate infrastructure and domain — extract it separately if/when a third service lands. + +## Triggers + +Per the existing TODO, the extractions should land when any of these occur: + +1. A second consumer needs the same wiring (a real production server, or an e2e harness building real per-queue profiles). +2. Per-queue config becomes data-driven (build profiles from `queueconfig.Store` / `queues.yaml` instead of Go literals) — step 4 of this proposal. +3. The bundle grows lifecycle (Close / health / hot-reload). + +Steps 1–3 can land independently as mechanical extractions with zero behavioral change. Step 4 (profile hints) is additive and can follow once the structural extractions stabilize. From 2e10005d47a00467b96bbf33f8e78f20755b3e15 Mon Sep 17 00:00:00 2001 From: jamy Date: Mon, 13 Jul 2026 22:31:40 +0000 Subject: [PATCH 2/9] docs(submitqueue): add trade-off analysis for profile hints vs QueueConfig removal Analyze three options for per-queue extension selection in light of potential QueueConfig removal: hints in the entity, a separate profile config file, or keeping profiles in Go code only. Recommends landing the mechanical extractions (steps 1-3) unconditionally and deferring step 4 until the QueueConfig question is resolved. Co-Authored-By: Claude Opus 4.6 --- doc/rfc/submitqueue/modular-queue-wiring.md | 32 ++++++++++++++++++++- 1 file changed, 31 insertions(+), 1 deletion(-) diff --git a/doc/rfc/submitqueue/modular-queue-wiring.md b/doc/rfc/submitqueue/modular-queue-wiring.md index fbe6d233..4a2e062b 100644 --- a/doc/rfc/submitqueue/modular-queue-wiring.md +++ b/doc/rfc/submitqueue/modular-queue-wiring.md @@ -222,7 +222,37 @@ Rewrite the example to compose the new packages: | Controller wire | `submitqueue/orchestrator/controller/wire` | `PrimaryParams`, `DLQParams`, `RegisterPrimary()`, `RegisterDLQ()` | ~170 (registerPrimaryControllers, registerDLQControllers) | | Profile hints | `submitqueue/entity` (extended) | `QueueProfile` (added to `QueueConfig`) | 0 (additive) | -## Rejected +## Trade-off: profile hints vs. removing QueueConfig entirely + +Step 4 (profile hints in `QueueConfig`) deserves separate scrutiny because there is an open question about whether `QueueConfig` should exist at all. + +### Current role of QueueConfig + +`QueueConfig` today is a single-field entity (`Name string`). Its sole consumer is the gateway's `LandController`, which calls `queueconfig.Store.Get(ctx, queue)` to reject requests targeting unknown queues — a pure name-validation gate. The orchestrator does not import `queueconfig` at all; it maintains its own hardcoded `queueRegistry` with no programmatic link to the YAML config. The TODO on line 477 of the orchestrator example envisions bridging the two ("see also queueconfig.Store, which holds the per-queue data half"), but that bridge does not exist today. + +### Three options for per-queue extension selection + +| Option | Description | Pros | Cons | +|---|---|---|---| +| **A: Profile hints in QueueConfig (step 4 as proposed)** | Add `QueueProfile` fields to the entity; deployers declare scorer/conflict/etc. in `queues.yaml`; the wiring layer maps hint strings → instances. | Single source of truth for queue identity + behavior. YAML-only queue addition for known extension types. | Expands `QueueConfig` from a pure name registry into a config carrier — if QueueConfig is later removed, these fields need a new home. The entity gains fields the gateway doesn't use (profile hints are orchestrator-only). | +| **B: Separate profile config file** | Leave `QueueConfig` as-is (name-only). Create a separate `queue-profiles.yaml` (or a `profiles:` section in a new file) consumed only by the orchestrator wiring. The orchestrator loads both queue names and profiles; the gateway loads only names. | Clean separation: gateway validates names, orchestrator resolves profiles. `QueueConfig` stays minimal and removable. No entity-level coupling. | Two config files to keep in sync (queue names must match). More moving parts in the wiring layer. | +| **C: No data-driven profiles — keep profiles in Go code** | Drop step 4 entirely. Steps 1–3 (queueprofile, topicregistry, wire) still land as mechanical extractions. Per-queue profiles stay in Go, just using the promoted `queueprofile.Registry` instead of the current inline types. | Simplest change. No new config surface. Full type safety — a misspelled scorer name is a compile error, not a runtime lookup miss. Consistent with the existing philosophy ("all behavioral and VCS configuration lives in the extension factory implementations"). | Adding a new queue still requires a recompile. Doesn't address the "three-place edit" problem for deployers who don't write custom extensions. | + +### If QueueConfig is removed + +If the direction is to remove `QueueConfig` entirely (perhaps because queue name validation moves to a different mechanism — e.g. the orchestrator's `queueprofile.Registry` becomes the implicit registry of valid queues, and the gateway queries it or the profile store), then: + +- **Option A becomes wasted work** — we'd add profile fields to an entity that's about to be deleted. +- **Option B is resilient** — the separate profile config survives independently. +- **Option C is neutral** — no config-layer dependency either way. + +If queue name validation stays but moves out of the entity (e.g. the gateway calls `queueprofile.Registry.Get()` directly, treating the profile registry as the source of truth for "which queues exist"), then `QueueConfig` + `queueconfig.Store` can be removed without losing the validation gate. The gateway would depend on the profile registry instead of a name-only store. This would be a clean removal: the `queueconfig` extension package, its YAML impl, its mock, and the `QueueConfig` entity all go away; the `queueprofile.Registry` subsumes both name validation and extension resolution. + +### Recommendation + +**Land steps 1–3 unconditionally** — they are mechanical extractions with zero behavioral change and no dependency on the QueueConfig question. **Defer step 4** until the QueueConfig question is resolved. If QueueConfig stays, option A is the natural evolution; if it's removed, option B or C is cleaner, and the promoted `queueprofile.Registry` from step 1 already provides the foundation either way. + + - **DI framework (wire/dig/fx).** Adds indirection and a build-time dependency for a problem that explicit code solves. The refactor reduces the volume of explicit wiring, not its nature. - **Hot-reload of queue configs.** Out of scope. The YAML is loaded at startup. Hot-reload can build on this foundation later — `queueconfig.Store` already abstracts the read path, so swapping the YAML impl for a watching impl is a future, independent change. From 56ff589d28e604d3270d5240885ca82048fe876f Mon Sep 17 00:00:00 2001 From: jamy Date: Tue, 14 Jul 2026 19:27:26 +0000 Subject: [PATCH 3/9] docs(submitqueue): rewrite RFC around declare-don't-assemble engine approach Incorporate the counter-proposal from the design gist: replace the original 5-step extraction plan with a unified pipeline.Construct[D] engine + lifecycle.Component approach. Services declare themselves as Deps struct + Stages slice + Controllers struct; the engine handles topic registry, DLQ pairing, consumer lifecycle, and startup ordering. Key changes: - Add vocabulary section and library/host boundary diagram - Replace steps 1-5 with the engine-based Steps 1-5 (lifecycle, pipeline, service self-declaration, host profiles, host main) - Simplify QueueConfig trade-off: recommend option C (profiles stay in Go code, host-private) over data-driven profile hints - Show how the pattern generalizes across all four services - Add "what the engine enforces" invariants table Co-Authored-By: Claude Opus 4.6 --- doc/rfc/submitqueue/modular-queue-wiring.md | 450 ++++++++++++-------- 1 file changed, 263 insertions(+), 187 deletions(-) diff --git a/doc/rfc/submitqueue/modular-queue-wiring.md b/doc/rfc/submitqueue/modular-queue-wiring.md index 4a2e062b..6a28216b 100644 --- a/doc/rfc/submitqueue/modular-queue-wiring.md +++ b/doc/rfc/submitqueue/modular-queue-wiring.md @@ -1,6 +1,6 @@ # Modular Queue Wiring -Design notes for making the orchestrator's per-queue extension wiring and topic-registry setup modular, data-driven, and reusable across deployers. Decisions and rationale only; the code changes land after this RFC is reviewed. +Design notes for making the orchestrator's per-queue extension wiring and topic-registry setup modular, reusable, and importable by external deployers. Decisions and rationale only; the code changes land after this RFC is reviewed. ## Problem @@ -10,221 +10,307 @@ The orchestrator's example `main.go` (`example/submitqueue/orchestrator/server/m 2. **Queue topology / topic registry** — `newTopicRegistry` is a static list of 12+ pipeline stages, each with a primary subscription and a mirrored DLQ subscription, plus publish-only topics. Adding or removing a pipeline stage requires editing this function in lockstep with controller registration. 3. **Per-queue extension wiring** — `queueRegistry`, `newQueueRegistry`, and four thin `*Factory` adapter types. The only way to configure which scorer / analyzer / change-provider / build-runner a queue uses is to edit Go code in this file, recompile, and redeploy. -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). This makes adoption harder for new integrators who want to deploy SubmitQueue with their own queues and extension profiles. +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 477](../../example/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. -## Principle - -- **The wiring layer assembles; the domain layer provides reusable building blocks.** Today the domain layer owns controllers, entities, and extension interfaces — but the *composition* of extensions into per-queue profiles and the *registration* of controllers into consumers is copy-pasted into each deployer's main.go. These compositions are mechanical and identical across deployers; they belong in the domain. -- **Data-driven where practical, code-driven where necessary.** Queue *names* are already data-driven (YAML via `queueconfig.Store`). Queue *extension profiles* — which scorer, which conflict analyzer — should also be declarable as data. Custom extension *implementations* remain code (a deployer writes a new `scorer.Scorer` impl), but selecting among known implementations should not require a recompile. -- **No DI framework.** The wiring stays explicit Go code. This refactor reduces its volume, not its nature. - -## Proposal +## Vocabulary -### 1. Promote queue-profile registry into `submitqueue/core/queueprofile` +``` +seam an extension interface the library defines and the deployer fills + (storage, buildrunner, sourcecontrol, …) — always an interface, never an impl +stage one pipeline step: a topic being consumed + the controller consuming it + (+ optionally its dead-letter reconciler) +engine pipeline.Construct — the ONE shared assembly routine +profile the host's per-queue choice of seam impls +host the deployer binary: our own example/ mains, or an external repo's fx/plain-main app +``` -Extract the `queueExtensions` struct (renamed `Profile`) and `queueRegistry` (renamed `Registry`) from the example into `submitqueue/core/queueprofile/`. This is the domain-internal analogue of `submitqueue/core/topickey` — infrastructure shared between the orchestrator and (potentially) future services, but private to the SubmitQueue domain. +## Principle -```go -// submitqueue/core/queueprofile/profile.go -package queueprofile +- **Declare, don't assemble.** A service is defined by three declarations — a seam contract (struct), a topology table (slice of stages), and a controller set (struct). The engine (`pipeline.Construct`) consumes these declarations and produces one lifecycle handle. The host fills seams and starts the handle. Assembly logic lives in exactly one place (the engine), not in every deployer's main.go. +- **Library vs. host.** The library (controllers, engine, extension interfaces) owns no server, reads no environment variables, catches no signals, and calls no `os.Exit`. The host owns transport (gRPC/HTTP), process model (signals, exit codes, health checks), and policy (credentials, impl selection, queue routing). This boundary makes the library importable by any host — fx, plain main, test harness — without unwanted side effects. +- **Profiles stay host-private.** The per-queue routing decision ("monorepo/main gets buildkite, monorepo/exp gets local runner") is host policy. It lives in the host's `profiles.go`, never crosses the library boundary, and maps to the library's `Factory` interfaces through thin adapters at the seam boundary. -// Profile is the full set of extension implementations for a single queue. -// Grouping per queue (rather than per extension) lets the wiring read as -// "for this queue, here are its scorer, analyzer, change provider, …" -// and lets a profile start from a baseline and override only what differs. -type Profile struct { - // ChangeProvider resolves change metadata for land requests in this queue. - ChangeProvider changeprovider.ChangeProvider +``` +┌───────────────────────── HOST ──────────────────────────┐ +│ transport process model policy │ +│ gRPC/HTTP — mount fx or plain main, config, creds,│ +│ controllers behind signals, exit codes impl selection,│ +│ YOUR proto+server health checks queue routing │ +└──────────┬──────────────────┬──────────────────┬─────────┘ + │ glue: pb svc → │ Start(ctx) │ Deps + │ controllers ▼ Stop(ctx) ▼ +┌──────────────────────── LIBRARY ────────────────────────┐ +│ controllers ONE lifecycle handle extension │ +│ (pure logic) from pipeline.Construct seams │ +│ NO server · NO env reads · NO signals · NO exit │ +└─────────────────────────────────────────────────────────┘ +``` - // BuildRunner triggers and polls CI builds for batches in this queue. - BuildRunner buildrunner.BuildRunner +## Proposal - // Scorer computes success probability for batches in this queue. - Scorer scorer.Scorer +### Step 1 · `platform/lifecycle` — the one interface a host ever sees - // Analyzer detects conflicts between batches in this queue. - Analyzer conflict.Analyzer +```go +// Component is anything with a lifecycle. Construct returns one; hosts drive it. +type Component interface { + Start(ctx context.Context) error + Stop(ctx context.Context) error } -// Registry maps a queue name to its Profile, falling back to a default -// for queues without an explicit entry. It is the single place that knows -// the queue topology; extension packages remain queue-agnostic. -type Registry struct { … } - -func NewRegistry(def Profile, perQueue map[string]Profile) Registry -func (r Registry) Get(queue string) Profile +// Group runs an ordered list of Components as one Component. +// +// Start: members in order; if member i fails to start, members i-1…0 are +// stopped in reverse and the error is returned — no half-started state. +// Stop: members in REVERSE order (work-acceptors drain before the +// connections under them close); errors joined, none swallowed. +func NewGroup(ordered ...Component) *Group ``` -The four thin factory adapters (`changeProviderFactory`, `buildRunnerFactory`, `scorerFactory`, `analyzerFactory`) also move into this package as exported types. They are mechanical — `For(cfg) → registry.Get(cfg.QueueName).X` — and every deployer needs them identically. - -**Why a new package instead of expanding `queueconfig`:** `queueconfig` is a resolution target (key/value store of queue names). The profile registry is a *consumer* of queue names that additionally bundles behavioral extension instances. Mixing them would give `queueconfig` a dependency on every extension interface, violating the "stores are resolution targets, not aggregators" principle from CLAUDE.md. +The engine uses Group internally; hosts can also nest Groups (e.g. two services in one process). This replaces the ad-hoc `sync.WaitGroup` + `chan` + manual error-joining in today's main.go. -### 2. Extract topic-registry builder into `submitqueue/core/topicregistry` - -Replace `newTopicRegistry` with a reusable builder that declaratively constructs primary + DLQ topic pairs from a slice of stage specs, plus publish-only topics. +### Step 2 · `platform/pipeline` — the engine ```go -// submitqueue/core/topicregistry/builder.go -package topicregistry - -// StageSpec declares one pipeline stage that needs a primary subscription -// and an auto-generated DLQ subscription. -type StageSpec struct { - // Key is the consumer.TopicKey for this stage. +// Stage is one row of a service's topology table. D is that service's Deps type. +type Stage[D any] struct { + // Key is the stage's LOGICAL topic key (e.g. topickey.Start). The engine maps + // it to a physical topic name via the TopicNames option — the mapping is host + // data, so two deployments can run the same pipeline on different topic names. Key consumer.TopicKey - // Name is the wire topic name (e.g. "start", "batch"). - Name string + // New builds the stage's controller from the service's Deps. The engine calls + // it ONCE, eagerly, inside Construct — so a nil/missing dependency fails at + // boot with the stage's name on it, never mid-delivery. + New func(D) (consumer.Controller, error) - // GroupSuffix is the consumer-group suffix (e.g. "orchestrator-start"). - GroupSuffix string + // DLQ, when non-nil, declares "this stage dead-letters". The engine then + // derives the paired DLQ topic (_dlq, retry budget, DLQ-of-DLQ + // disabled) AND registers this reconciler on the DLQ consumer. Declaring + // one without getting the other is impossible — that's the invariant. + DLQ func(D) (consumer.Controller, error) } +``` -// BuildParams configures the topic registry builder. -type BuildParams struct { - // Queue is the message queue backend. - Queue extqueue.Queue - - // SubscriberName identifies this subscriber for partition leases. - SubscriberName string - - // Stages are the primary pipeline stages. Each gets a paired DLQ - // subscription with DLQ disabled (no _dlq_dlq cascade) and a high - // MaxAttempts for convergent reconciliation. - Stages []StageSpec - - // PublishOnly are topics the service publishes to but never consumes. - PublishOnly []consumer.TopicConfig +```go +// Construct is the ONLY assembly code in the repo. Schematic: +func Construct[D any](deps D, stages []Stage[D], opts ...Option) (lifecycle.Component, error) { + o := applyOptions(opts) // TopicNames map, Classifiers, extra Components + + registry := consumer.NewTopicRegistry() + primary := consumer.New(o.queues, registry, o.classifiers, subscriberName()) + dlq := consumer.New(o.queues, registry, errs.AlwaysRetryableProcessor, subscriberName()) + + for _, s := range stages { + topic := o.physicalName(s.Key) // logical key → deployment's topic name + registry.Add(s.Key, subscription(topic)) + + ctl, err := s.New(deps) // eager ⇒ boot-time, named failure + if err != nil { return nil, fmt.Errorf("stage %s: %w", s.Key, err) } + primary.Register(ctl) + + if s.DLQ != nil { // declared ⇒ pair + reconciler, derived together + registry.Add(dlqKey(s.Key), dlqSubscription(topic)) + rec, err := s.DLQ(deps) + if err != nil { return nil, fmt.Errorf("stage %s dlq: %w", s.Key, err) } + dlq.Register(rec) + } + } + + // order is decided HERE, once: infra → publishers → consumers; Stop reverses it + return lifecycle.NewGroup(o.infra, o.publishers, primary, dlq), nil } - -func Build(p BuildParams) (consumer.TopicRegistry, error) ``` -This eliminates the manual duplication of the primary/DLQ pairing pattern across 12 stages. Each stage is one `StageSpec` entry; the builder guarantees every primary stage gets a correctly-configured DLQ subscription (disabled DLQ-of-DLQ, high MaxAttempts) without copy-paste. Adding or removing a pipeline stage becomes adding or removing one line. +This unifies topic-registry construction, subscription configuration, controller creation, DLQ pairing, and consumer lifecycle into one call. Each stage is a single row in a typed table; the engine enforces the invariant that every DLQ-declaring stage gets a paired DLQ subscription and reconciler, derived together. The host never sees consumer internals. -### 3. Extract controller-registration helpers into `submitqueue/orchestrator/controller/wire` +### Step 3 · Service self-declaration — `submitqueue/orchestrator/pipeline.go` -Create a `wire` subpackage under `submitqueue/orchestrator/controller/` with two functions: +A service's **entire** definition is three declarations — a struct (seams), a slice (topology), a constructor (controllers). No assembly code: ```go -// submitqueue/orchestrator/controller/wire/wire.go -package wire - -// PrimaryParams holds the dependencies needed to construct and register -// all primary pipeline controllers. -type PrimaryParams struct { - Consumer consumer.Consumer - Logger *zap.SugaredLogger - Scope tally.Scope - Registry consumer.TopicRegistry - ChangeProviderF changeprovider.Factory - BuildRunnerF buildrunner.Factory - ScorerF scorer.Factory - ConflictF conflict.Factory - Counter counter.Counter - Store storage.Storage +// ① Deps: one field per dependency the pipeline needs. +// This struct IS the service's public API toward deployers. +type Deps struct { + Logger *zap.SugaredLogger + Scope tally.Scope + Storage storage.Storage // singleton seams + Queues messagequeue.Stores + BuildRunner buildrunner.Factory // per-queue seams: Factory is a RESOLVER — + ChangeProvider changeprovider.Factory // For(Config{QueueName}) → impl for THAT queue + Scorer scorer.Factory + Analyzer conflict.Factory + Counter counter.Counter } -// RegisterPrimary creates and registers all primary pipeline controllers. -// Returns the count of registered controllers. -func RegisterPrimary(p PrimaryParams) (int, error) - -// DLQParams holds the dependencies needed to construct and register -// all DLQ reconciliation controllers. -type DLQParams struct { - Consumer consumer.Consumer - Logger *zap.SugaredLogger - Scope tally.Scope - Store storage.Storage +// ② Stages: the pipeline topology as a typed table. +// Adding a stage = adding one row. Nothing else, anywhere. +var Stages = []pipeline.Stage[Deps]{ + { + Key: topickey.Start, + New: func(d Deps) (consumer.Controller, error) { + return start.NewController(d.Logger, d.Scope, d.Storage, d.Queues), nil + }, + DLQ: func(d Deps) (consumer.Controller, error) { + return dlq.NewRequestController(d.Logger, d.Scope, d.Storage, + dlq.DecodeLandRequestID, dlq.TopicKey(topickey.Start), + "orchestrator-start-dlq"), nil + }, + }, + { Key: topickey.Validate, /* same shape */ }, + { Key: topickey.Batch, /* same shape */ }, + // … all 12 stages } -// RegisterDLQ creates and registers all DLQ reconciliation controllers. -// Returns the count of registered controllers. -func RegisterDLQ(p DLQParams) (int, error) +// ③ Controllers: RPC-facing controllers, constructed but NOT bound to any +// wire contract. Binding to a proto service + transport is host glue, +// because consumers may use different protos or transports. +type Controllers struct { + Ping *controller.PingController +} + +func NewControllers(d Deps) Controllers { + return Controllers{Ping: controller.NewPingController(d.Logger, d.Scope)} +} ``` -This keeps the controller list in the domain layer (testable, importable) and reduces the wiring main.go to: build dependencies → call `wire.RegisterPrimary` → call `wire.RegisterDLQ`. Adding a new pipeline stage becomes a single-file edit in this package. +### Step 4 · Host-private profiles — `service/.../profiles.go` -### 4. Extend `QueueConfig` with optional profile hints +Profiles stay entirely in the host. Nothing profile-shaped crosses the library boundary: -Add an optional `Profile` field to `entity.QueueConfig` so the YAML file can declare which scorer / conflict / build-runner strategy each queue uses: +```go +type Profile struct { + BuildRunner buildrunner.BuildRunner + ChangeProvider changeprovider.Provider + Scorer scorer.Scorer + Analyzer conflict.Analyzer +} -```yaml -queues: - - name: test-queue - profile: - scorer: heuristic - conflict: file-overlap - - name: e2e-test-queue - profile: - scorer: composite - conflict: none - - name: e2e-cancel-queue - # No profile — inherits the baseline. -``` +type Profiles struct { + byQueue map[string]Profile + defaultProfile Profile +} -```go -// submitqueue/entity/queue_config.go -type QueueConfig struct { - // Name uniquely identifies this queue within the system. - Name string `json:"name" yaml:"name"` - - // Profile carries optional hints for which extension implementations - // this queue uses. The wiring layer maps hint strings to concrete - // extension instances; the entity does not import extension packages. - // Zero value means "use the deployer's baseline profile." - Profile QueueProfile `json:"profile,omitempty" yaml:"profile,omitempty"` +func (p Profiles) For(queue string) Profile { + if prof, ok := p.byQueue[queue]; ok { return prof } + return p.defaultProfile +} + +func newProfiles(cfg Config) Profiles { + return Profiles{ + byQueue: map[string]Profile{ + "monorepo/main": {BuildRunner: buildkite.New(cfg.CI), Scorer: heuristic.New()}, + "monorepo/exp": {BuildRunner: local.New(), Scorer: heuristic.New()}, + }, + defaultProfile: Profile{BuildRunner: noop.New(), Scorer: constant.New()}, + } } -// QueueProfile carries string-typed hints for extension selection. -// Each field names a known implementation (e.g. "heuristic", "composite", -// "file-overlap", "none", "all"). Deployers register the mapping from -// hint → implementation in the wiring layer. An empty string means -// "inherit from the baseline." -type QueueProfile struct { - // Scorer names the scoring strategy (e.g. "heuristic", "composite"). - Scorer string `json:"scorer,omitempty" yaml:"scorer,omitempty"` +// Thin adapters crossing the boundary — the http.HandlerFunc trick: +type buildRunnerFunc func(buildrunner.Config) (buildrunner.BuildRunner, error) +func (f buildRunnerFunc) For(c buildrunner.Config) (buildrunner.BuildRunner, error) { return f(c) } - // Conflict names the conflict-analysis strategy (e.g. "all", "none", "file-overlap"). - Conflict string `json:"conflict,omitempty" yaml:"conflict,omitempty"` +func (p Profiles) BuildRunnerFactory() buildrunner.Factory { + return buildRunnerFunc(func(c buildrunner.Config) (buildrunner.BuildRunner, error) { + return p.For(c.QueueName).BuildRunner, nil + }) +} +// … same pattern for Scorer, Analyzer, ChangeProvider +``` - // BuildRunner names the build-runner backend (e.g. "fake", "jenkins"). - BuildRunner string `json:"build_runner,omitempty" yaml:"build_runner,omitempty"` +### Step 5 · Host main.go — `service/.../main.go` - // ChangeProvider names the change-provider backend (e.g. "github", "phabricator", "routing"). - ChangeProvider string `json:"change_provider,omitempty" yaml:"change_provider,omitempty"` +```go +func run(ctx context.Context) error { + cfg := loadConfig() // env/flags: host-owned + logger, scope := newLogger(cfg), newScope(cfg) + + store := storagemysql.New(cfg.DB, logger, scope) + queues := mqmysql.New(cfg.QueueDB, logger, scope) + profiles := newProfiles(cfg) // Step 4 + + deps := orchestrator.Deps{ + Logger: logger, Scope: scope, + Storage: store, Queues: queues, + BuildRunner: profiles.BuildRunnerFactory(), + ChangeProvider: profiles.ChangeProviderFactory(), + Scorer: profiles.ScorerFactory(), + Analyzer: profiles.AnalyzerFactory(), + } + + pl, err := pipeline.Construct(deps, orchestrator.Stages, + pipeline.TopicNames(cfg.TopicNames), // logical → physical topic names + pipeline.Classifiers(backendClassifiers()), + ) + if err != nil { return err } + + srv := grpc.NewServer() + ctls := orchestrator.NewControllers(deps) + pb.RegisterSubmitQueueOrchestratorServer(srv, rpcServer{c: ctls}) + + if err := pl.Start(ctx); err != nil { return err } + defer pl.Stop(context.Background()) + return serveUntilDone(ctx, srv) } ``` -**Constraints:** `QueueConfig` stays a simple data carrier — it does NOT import extension packages. The mapping from hint string → extension instance remains in the wiring layer (`example/.../main.go` or a deployer's equivalent). This preserves the clean architecture boundary: entities are pure, factories are injected. +## At delivery time — all the pieces meeting -### 5. Refactor the example orchestrator main.go +``` +row appears on topic "start", partition key "monorepo/exp" + └─▶ primary consumer (built in Step 2) holds the partition lease, fetches delivery + └─▶ start controller (built by Step 3's row) .Process(ctx, delivery) + └─▶ needs a runner for THIS queue: + deps.BuildRunner.For(Config{QueueName: "monorepo/exp"}) + └─▶ Step 4's adapter → profiles.For("monorepo/exp").BuildRunner → local runner + (the SAME Deps field answers "monorepo/main" with buildkite + on the next delivery — that's the Factory-as-resolver contract, + identical to today's buildRunnerFactory{queues} in main.go) + controller returns nil ⇒ ack + controller returns err ⇒ Step 5's classifier decides: retry (nack) or not; + after retry budget ⇒ row moves to "start_dlq" ⇒ Step 2's derived pairing + guarantees the reconciler from Step 3's DLQ field is listening there +``` -Rewrite the example to compose the new packages: +## Generalizes across all four services -1. Build infrastructure (DB, logger, metrics) — stays in main.go (deployment-specific). -2. Load queue configs from YAML — stays in main.go. -3. Build queue profiles using `queueprofile.NewRegistry(...)` — stays in main.go but is now ~30 lines instead of ~100, and can optionally be driven by profile hints from the YAML. -4. Build topic registry using `topicregistry.Build(...)` — one call, ~10 lines instead of ~90. -5. Register controllers using `wire.RegisterPrimary(...)` + `wire.RegisterDLQ(...)` — two calls instead of ~170 lines. -6. Start consumers and gRPC server — stays in main.go. +``` + gateway orchestrator stovepipe runway +──────────────────────────────────────────────────────────────────────────────────────────── + Deps seams counter · storage · changeprovider · storage · counter · storage · + queueconfig.Store · buildrunner · scorer sourcecontrol. merger Factory + requestlog store analyzer · validator Factory · + (+7 speculation) queueconfig.Store + + Stages log start · cancel · process mergeconflictcheck · + (rows) validate · batch · merge + … (+ DLQ column) + + Controllers Gateway Orchestrator Stovepipe Runway + Ping·Land·Cancel Ping Ping·Ingest Ping + + host keeps impl selection · creds · TopicKey→name map · classifiers · transport · signals + — identical across all four columns: a new service copies any column and + fills in two rows of DATA +──────────────────────────────────────────────────────────────────────────────────────────── +``` -**Result:** main.go drops from ~950 to ~300 lines. The domain-layer packages are independently testable. A new integrator copies the example, edits `queues.yaml` (including optional profile hints), and optionally customizes the `queueprofile.Registry` population — no need to understand the full pipeline topology. +Two integration surfaces fall out — the Go library surface above (Deps · Stages · Controllers · `pipeline.Construct`), and the proto contracts in `api/` for cross-language consumers. The Go library never dictates the wire contract; binding controllers to a proto service + transport is host glue. -## What each extraction produces +## What the engine enforces -| Extraction | New package | Key types | Lines removed from main.go | -|---|---|---|---| -| Queue profiles | `submitqueue/core/queueprofile` | `Profile`, `Registry`, `ChangeProviderFactory`, `BuildRunnerFactory`, `ScorerFactory`, `AnalyzerFactory` | ~100 (queueExtensions, queueRegistry, 4 factory types, newQueueRegistry) | -| Topic registry | `submitqueue/core/topicregistry` | `StageSpec`, `BuildParams`, `Build()` | ~90 (newTopicRegistry) | -| Controller wire | `submitqueue/orchestrator/controller/wire` | `PrimaryParams`, `DLQParams`, `RegisterPrimary()`, `RegisterDLQ()` | ~170 (registerPrimaryControllers, registerDLQControllers) | -| Profile hints | `submitqueue/entity` (extended) | `QueueProfile` (added to `QueueConfig`) | 0 (additive) | +| Concern | Enforced by | +|---|---| +| Start/Stop ordering, rollback on partial failure | `lifecycle.Group` — one implementation, tested once | +| DLQ pair + reconciler always present together | Engine property: any row declaring `DLQ:` gets both the DLQ subscription and the reconciler, derived together | +| Missing seam at boot | Eager ctor run in `Construct` ⇒ named boot error with the failing stage | +| Naming drift | Nothing to name — services export data (Deps struct + Stages slice), not assembly functions | +| Wrong data (bad row) | Typechecking + a trivial data test (`Stages` keys unique) | ## Trade-off: profile hints vs. removing QueueConfig entirely -Step 4 (profile hints in `QueueConfig`) deserves separate scrutiny because there is an open question about whether `QueueConfig` should exist at all. +Profile selection (which scorer/conflict/build-runner a queue uses) deserves separate scrutiny because there is an open question about whether `QueueConfig` should exist at all. ### Current role of QueueConfig @@ -234,39 +320,29 @@ Step 4 (profile hints in `QueueConfig`) deserves separate scrutiny because there | Option | Description | Pros | Cons | |---|---|---|---| -| **A: Profile hints in QueueConfig (step 4 as proposed)** | Add `QueueProfile` fields to the entity; deployers declare scorer/conflict/etc. in `queues.yaml`; the wiring layer maps hint strings → instances. | Single source of truth for queue identity + behavior. YAML-only queue addition for known extension types. | Expands `QueueConfig` from a pure name registry into a config carrier — if QueueConfig is later removed, these fields need a new home. The entity gains fields the gateway doesn't use (profile hints are orchestrator-only). | -| **B: Separate profile config file** | Leave `QueueConfig` as-is (name-only). Create a separate `queue-profiles.yaml` (or a `profiles:` section in a new file) consumed only by the orchestrator wiring. The orchestrator loads both queue names and profiles; the gateway loads only names. | Clean separation: gateway validates names, orchestrator resolves profiles. `QueueConfig` stays minimal and removable. No entity-level coupling. | Two config files to keep in sync (queue names must match). More moving parts in the wiring layer. | -| **C: No data-driven profiles — keep profiles in Go code** | Drop step 4 entirely. Steps 1–3 (queueprofile, topicregistry, wire) still land as mechanical extractions. Per-queue profiles stay in Go, just using the promoted `queueprofile.Registry` instead of the current inline types. | Simplest change. No new config surface. Full type safety — a misspelled scorer name is a compile error, not a runtime lookup miss. Consistent with the existing philosophy ("all behavioral and VCS configuration lives in the extension factory implementations"). | Adding a new queue still requires a recompile. Doesn't address the "three-place edit" problem for deployers who don't write custom extensions. | +| **A: Profile hints in QueueConfig** | Add `QueueProfile` fields to the entity; deployers declare scorer/conflict/etc. in `queues.yaml`; the wiring layer maps hint strings → instances. | Single source of truth for queue identity + behavior. YAML-only queue addition for known extension types. | Expands `QueueConfig` from a pure name registry into a config carrier — if QueueConfig is later removed, these fields need a new home. The entity gains fields the gateway doesn't use (profile hints are orchestrator-only). | +| **B: Separate profile config file** | Leave `QueueConfig` as-is (name-only). Create a separate config consumed only by the host's profiles.go. | Clean separation: gateway validates names, host resolves profiles. `QueueConfig` stays minimal and removable. No entity-level coupling. | Two config files to keep in sync (queue names must match). More moving parts in the wiring layer. | +| **C: Profiles stay in Go code (recommended)** | Per-queue profiles stay in the host's `profiles.go` as Go code. Full type safety — a misspelled scorer name is a compile error, not a runtime lookup miss. Consistent with the existing philosophy ("all behavioral and VCS configuration lives in the extension factory implementations"). | Simplest change. No new config surface. Type-safe. Matches the "host owns policy" principle. `core/queueprofile` stays trigger-gated per the in-code TODO: if a trigger fires, `profiles.go` moves wholesale — nothing profile-shaped ever crossed the boundary. | Adding a new queue requires a recompile. | ### If QueueConfig is removed -If the direction is to remove `QueueConfig` entirely (perhaps because queue name validation moves to a different mechanism — e.g. the orchestrator's `queueprofile.Registry` becomes the implicit registry of valid queues, and the gateway queries it or the profile store), then: - -- **Option A becomes wasted work** — we'd add profile fields to an entity that's about to be deleted. -- **Option B is resilient** — the separate profile config survives independently. -- **Option C is neutral** — no config-layer dependency either way. - -If queue name validation stays but moves out of the entity (e.g. the gateway calls `queueprofile.Registry.Get()` directly, treating the profile registry as the source of truth for "which queues exist"), then `QueueConfig` + `queueconfig.Store` can be removed without losing the validation gate. The gateway would depend on the profile registry instead of a name-only store. This would be a clean removal: the `queueconfig` extension package, its YAML impl, its mock, and the `QueueConfig` entity all go away; the `queueprofile.Registry` subsumes both name validation and extension resolution. - -### Recommendation - -**Land steps 1–3 unconditionally** — they are mechanical extractions with zero behavioral change and no dependency on the QueueConfig question. **Defer step 4** until the QueueConfig question is resolved. If QueueConfig stays, option A is the natural evolution; if it's removed, option B or C is cleaner, and the promoted `queueprofile.Registry` from step 1 already provides the foundation either way. - +If queue name validation moves to a different mechanism (e.g. the host's profile registry becomes the implicit registry of valid queues, and the gateway queries it), then `QueueConfig` + `queueconfig.Store` can be removed without losing the validation gate. This would be a clean removal: the `queueconfig` extension package, its YAML impl, its mock, and the `QueueConfig` entity all go away. Option C is resilient to this removal because profiles never depended on `QueueConfig` in the first place. +## Rejected -- **DI framework (wire/dig/fx).** Adds indirection and a build-time dependency for a problem that explicit code solves. The refactor reduces the volume of explicit wiring, not its nature. -- **Hot-reload of queue configs.** Out of scope. The YAML is loaded at startup. Hot-reload can build on this foundation later — `queueconfig.Store` already abstracts the read path, so swapping the YAML impl for a watching impl is a future, independent change. -- **Changing the Factory interface contract.** The existing `Factory.For(Config)` pattern is sound and is the way controllers resolve per-queue extension instances. We add a first-class registry that factories resolve *against*, not a new factory contract. -- **Promoting `newChangeProvider` / `newGitHubChangeProvider` / `newPhabChangeProvider` out of the example.** These are deployment-specific (token sources, HTTP clients, timeouts). They stay in the wiring layer. -- **Merging `queueprofile` into `queueconfig`.** The config store is a resolution target (key/value); the profile registry aggregates behavioral instances. Mixing them gives `queueconfig` a dependency on every extension interface, violating the "stores are resolution targets" principle. -- **A generic "service bootstrap" package.** The duplicated boilerplate between gateway and orchestrator (logger, metrics, DB, gRPC server, signal handling) is real but is a separate, orthogonal concern. Folding it into this RFC would conflate infrastructure and domain — extract it separately if/when a third service lands. +- **DI framework (wire/dig/fx).** `pipeline.Construct` is not DI: no runtime graph, no reflection, no topo-sort — a typed engine over declarative data. One-offs enter via `Option`s. Hosts that use fx can wrap the engine in `fx.Provide` / `fx.Hook` without the engine knowing. +- **Hot-reload of queue configs.** Out of scope. The YAML is loaded at startup. Hot-reload can build on this foundation later. +- **Changing the Factory interface contract.** The existing `Factory.For(Config)` pattern is sound. The engine consumes factories exactly as today's main.go does; profiles produce them through the same thin adapters currently in main.go. +- **Promoting `newChangeProvider` / `newGitHubChangeProvider` / `newPhabChangeProvider` out of the example.** These are deployment-specific (token sources, HTTP clients, timeouts). They stay in the host. +- **Merging profiles into `queueconfig`.** The config store is a resolution target (key/value); profiles aggregate behavioral instances. Mixing them gives `queueconfig` a dependency on every extension interface, violating the "stores are resolution targets" principle. -## Triggers +## Migration path -Per the existing TODO, the extractions should land when any of these occur: +The refactor can land incrementally: -1. A second consumer needs the same wiring (a real production server, or an e2e harness building real per-queue profiles). -2. Per-queue config becomes data-driven (build profiles from `queueconfig.Store` / `queues.yaml` instead of Go literals) — step 4 of this proposal. -3. The bundle grows lifecycle (Close / health / hot-reload). +1. **`platform/lifecycle`** — pure addition, no existing code changes. Tested independently. +2. **`platform/pipeline`** — pure addition, no existing code changes. Tested with mock stages. +3. **Service self-declarations** (`orchestrator/pipeline.go`, etc.) — pure addition alongside the existing controllers. +4. **Example main.go rewrite** — the one breaking change: replace ~950 lines with ~100 lines composing the new packages. The existing behavior is identical; the diff is large but mechanical. -Steps 1–3 can land independently as mechanical extractions with zero behavioral change. Step 4 (profile hints) is additive and can follow once the structural extractions stabilize. +Steps 1–3 can land independently as pure additions with zero behavioral change. Step 4 is the switch-over. From 0387fe70556e1d2540d2288f2af4ef6ace65f61a Mon Sep 17 00:00:00 2001 From: jamy Date: Wed, 15 Jul 2026 17:36:55 +0000 Subject: [PATCH 4/9] docs(submitqueue): add fluent builder API section to modular queue wiring RFC --- doc/rfc/submitqueue/modular-queue-wiring.md | 107 ++++++++++++++++++++ 1 file changed, 107 insertions(+) diff --git a/doc/rfc/submitqueue/modular-queue-wiring.md b/doc/rfc/submitqueue/modular-queue-wiring.md index 6a28216b..5e95b6ed 100644 --- a/doc/rfc/submitqueue/modular-queue-wiring.md +++ b/doc/rfc/submitqueue/modular-queue-wiring.md @@ -308,6 +308,113 @@ Two integration surfaces fall out — the Go library surface above (Deps · Stag | Naming drift | Nothing to name — services export data (Deps struct + Stages slice), not assembly functions | | Wrong data (bad row) | Typechecking + a trivial data test (`Stages` keys unique) | +## Fluent builder API — convenience layer on top of the engine + +The `pipeline.Construct[D]` engine is the foundational API: typed, composable, and testable. For deployers who wire a single orchestrator with a handful of queues, a **fluent builder** provides a more readable entry point without hiding or replacing the engine. + +### Usage + +```go +app, err := submitqueue.New(). + WithStorage(store). + WithMessageQueue(mq). + WithQueue( + submitqueue.Queue("go-code"). + WithChangeProvider(ghProvider). + WithBuildRunner(buildkiteBuildRunner). + WithScorer(defaultScorer). + WithConflictAnalyzer(tango), + ). + WithQueue( + submitqueue.Queue("monorepo/exp"). + WithChangeProvider(ghProvider). + WithBuildRunner(localRunner). + WithScorer(defaultScorer). + WithConflictAnalyzer(fileOverlap), + ). + Build() + +if err != nil { return err } +if err := app.Start(ctx); err != nil { return err } +defer app.Stop(context.Background()) +``` + +### Implementation sketch + +```go +// submitqueue/builder.go +package submitqueue + +// Builder accumulates configuration for a SubmitQueue orchestrator app. +// It is a convenience layer — Build() populates a Deps struct, constructs +// profiles, and calls pipeline.Construct under the hood. +type Builder struct { + storage storage.Storage + queues messagequeue.Stores + perQueue map[string]Profile + opts []pipeline.Option + errs []error +} + +func New() *Builder { return &Builder{perQueue: map[string]Profile{}} } + +func (b *Builder) WithStorage(s storage.Storage) *Builder { + b.storage = s; return b +} + +func (b *Builder) WithMessageQueue(q messagequeue.Stores) *Builder { + b.queues = q; return b +} + +func (b *Builder) WithQueue(qb QueueBuilder) *Builder { + b.perQueue[qb.name] = qb.profile; return b +} + +func (b *Builder) WithOption(o pipeline.Option) *Builder { + b.opts = append(b.opts, o); return b +} + +func (b *Builder) Build() (*App, error) { + // Validate required fields (storage, queues, at least one queue). + // Populate Deps from the accumulated state. + // Build profiles registry from perQueue map. + // Call pipeline.Construct(deps, orchestrator.Stages, b.opts...). + // Return App wrapping the lifecycle.Component. +} + +// QueueBuilder accumulates per-queue extension selections. +type QueueBuilder struct { + name string + profile Profile +} + +func Queue(name string) QueueBuilder { return QueueBuilder{name: name} } + +func (q QueueBuilder) WithChangeProvider(cp changeprovider.ChangeProvider) QueueBuilder { + q.profile.ChangeProvider = cp; return q +} + +func (q QueueBuilder) WithBuildRunner(br buildrunner.BuildRunner) QueueBuilder { + q.profile.BuildRunner = br; return q +} + +func (q QueueBuilder) WithScorer(s scorer.Scorer) QueueBuilder { + q.profile.Scorer = s; return q +} + +func (q QueueBuilder) WithConflictAnalyzer(a conflict.Analyzer) QueueBuilder { + q.profile.Analyzer = a; return q +} +``` + +### Design constraints + +- **Convenience, not replacement.** The builder calls `pipeline.Construct` — it does not bypass or duplicate the engine. Deployers who need full control (custom `Option`s, multi-service composition, fx integration) use the engine directly. +- **Compile-time type safety.** Each `With*` method takes the concrete extension interface, not a string hint. A missing or mistyped extension is a compile error. +- **`QueueBuilder` is a value type.** The fluent chain returns copies, not pointers, so partial builders are safe to reuse as templates (e.g. a `baseQueue` with defaults that each real queue overrides). +- **`Build()` validates eagerly.** Missing required fields (no storage, no queues, zero queue profiles) produce a clear error at build time, not a nil-pointer panic at runtime. +- **No global state.** `New()` returns an isolated builder. Multiple orchestrator apps can coexist in the same process (useful for integration tests). + ## Trade-off: profile hints vs. removing QueueConfig entirely Profile selection (which scorer/conflict/build-runner a queue uses) deserves separate scrutiny because there is an open question about whether `QueueConfig` should exist at all. From 202618b66d7a6ac94a09f1e34e130cb09dc97d56 Mon Sep 17 00:00:00 2001 From: jamy Date: Wed, 15 Jul 2026 18:28:10 +0000 Subject: [PATCH 5/9] docs(submitqueue): add complete usage examples for core engine and fluent builder APIs --- doc/rfc/submitqueue/modular-queue-wiring.md | 267 ++++++++++++++++++-- 1 file changed, 243 insertions(+), 24 deletions(-) diff --git a/doc/rfc/submitqueue/modular-queue-wiring.md b/doc/rfc/submitqueue/modular-queue-wiring.md index 5e95b6ed..fc504e06 100644 --- a/doc/rfc/submitqueue/modular-queue-wiring.md +++ b/doc/rfc/submitqueue/modular-queue-wiring.md @@ -221,6 +221,8 @@ func (p Profiles) BuildRunnerFactory() buildrunner.Factory { ### Step 5 · Host main.go — `service/.../main.go` +The host's main.go shrinks to infrastructure setup, profile construction, and a single `pipeline.Construct` call. See "Usage examples" below for complete, runnable examples of both the core engine and fluent builder paths. + ```go func run(ctx context.Context) error { cfg := loadConfig() // env/flags: host-owned @@ -308,37 +310,237 @@ Two integration surfaces fall out — the Go library surface above (Deps · Stag | Naming drift | Nothing to name — services export data (Deps struct + Stages slice), not assembly functions | | Wrong data (bad row) | Typechecking + a trivial data test (`Stages` keys unique) | -## Fluent builder API — convenience layer on top of the engine +## Usage examples + +Two APIs serve different deployer needs. The core engine (`pipeline.Construct`) gives full control; the fluent builder (`submitqueue.New()`) wraps it for the common case. + +### When to use which -The `pipeline.Construct[D]` engine is the foundational API: typed, composable, and testable. For deployers who wire a single orchestrator with a handful of queues, a **fluent builder** provides a more readable entry point without hiding or replacing the engine. +| Scenario | API | Why | +|---|---|---| +| Single orchestrator, handful of queues | Fluent builder | Reads top-to-bottom; no boilerplate | +| Multiple services in one process | Core engine | Compose multiple `lifecycle.Component`s into a `lifecycle.Group` | +| fx / custom DI integration | Core engine | `Construct` returns a `lifecycle.Component` that slots into `fx.Hook` | +| Custom `pipeline.Option`s (classifiers, extra components) | Either | Fluent builder exposes `WithOption()`; core engine takes variadic `Option`s directly | +| Integration / e2e tests | Either | Both produce a `lifecycle.Component` with `Start`/`Stop` | -### Usage +### Core engine — complete example + +A single file showing the full `pipeline.Construct` path end-to-end. This is what Steps 3–5 above produce when stitched together. ```go +package main + +import ( + "context" + "os/signal" + + "github.com/uber/submitqueue/platform/lifecycle" + "github.com/uber/submitqueue/platform/pipeline" + "github.com/uber/submitqueue/submitqueue/orchestrator" + storagemysql "github.com/uber/submitqueue/submitqueue/extension/storage/mysql" + mqmysql "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" +) + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) + defer stop() + if err := run(ctx); err != nil { log.Fatal(err) } +} + +func run(ctx context.Context) error { + // ── infrastructure (host-owned) ───────────────────────────────── + cfg := loadConfig() + logger, scope := newLogger(cfg), newScope(cfg) + store := storagemysql.New(cfg.DB, logger, scope) + queues := mqmysql.New(cfg.QueueDB, logger, scope) + + // ── per-queue profiles (host-private, Step 4) ─────────────────── + profiles := newProfiles(cfg) + + // ── populate Deps (the library's public API, Step 3) ──────────── + deps := orchestrator.Deps{ + Logger: logger, + Scope: scope, + Storage: store, + Queues: queues, + // Factory fields — profiles produce thin adapters that cross + // the host/library boundary via the existing Factory interface: + BuildRunner: profiles.BuildRunnerFactory(), + ChangeProvider: profiles.ChangeProviderFactory(), + Scorer: profiles.ScorerFactory(), + Analyzer: profiles.AnalyzerFactory(), + } + + // ── assemble the pipeline (Step 2) ────────────────────────────── + // orchestrator.Stages is a []pipeline.Stage[orchestrator.Deps] declared + // once in the library (Step 3). The host never lists stages or controllers. + pl, err := pipeline.Construct(deps, orchestrator.Stages, + pipeline.TopicNames(cfg.TopicNames), // logical → physical topic names + pipeline.Classifiers(backendClassifiers()), // error classification per backend + ) + if err != nil { return err } + + // ── transport (host-owned) ────────────────────────────────────── + srv := grpc.NewServer() + ctls := orchestrator.NewControllers(deps) + pb.RegisterSubmitQueueOrchestratorServer(srv, rpcServer{c: ctls}) + + // ── lifecycle ─────────────────────────────────────────────────── + if err := pl.Start(ctx); err != nil { return err } + defer pl.Stop(context.Background()) + return serveUntilDone(ctx, srv) +} +``` + +**Advanced: two services in one process.** The core engine returns `lifecycle.Component`, so composing multiple services is a `lifecycle.NewGroup` call: + +```go +orchPl, _ := pipeline.Construct(orchDeps, orchestrator.Stages, orchOpts...) +gwPl, _ := pipeline.Construct(gwDeps, gateway.Stages, gwOpts...) + +combined := lifecycle.NewGroup(orchPl, gwPl) // start in order, stop in reverse +if err := combined.Start(ctx); err != nil { return err } +defer combined.Stop(context.Background()) +``` + +**Advanced: fx integration.** The engine has no opinion on DI frameworks — `lifecycle.Component` maps directly to fx hooks: + +```go +fx.New( + fx.Provide(newDeps, newProfiles), + fx.Invoke(func(lc fx.Lifecycle, deps orchestrator.Deps) error { + pl, err := pipeline.Construct(deps, orchestrator.Stages) + if err != nil { return err } + lc.Append(fx.Hook{ + OnStart: pl.Start, + OnStop: pl.Stop, + }) + return nil + }), +).Run() +``` + +### Fluent builder — complete example + +The same orchestrator, expressed with the builder API. `Build()` calls `pipeline.Construct` internally — the engine is always the assembly mechanism. + +```go +package main + +import ( + "context" + "os/signal" + + "github.com/uber/submitqueue/submitqueue" + storagemysql "github.com/uber/submitqueue/submitqueue/extension/storage/mysql" + mqmysql "github.com/uber/submitqueue/platform/extension/messagequeue/mysql" +) + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt) + defer stop() + if err := run(ctx); err != nil { log.Fatal(err) } +} + +func run(ctx context.Context) error { + cfg := loadConfig() + logger, scope := newLogger(cfg), newScope(cfg) + store := storagemysql.New(cfg.DB, logger, scope) + mq := mqmysql.New(cfg.QueueDB, logger, scope) + + app, err := submitqueue.New(). + WithLogger(logger). + WithScope(scope). + WithStorage(store). + WithMessageQueue(mq). + WithQueue( + submitqueue.Queue("go-code"). + WithChangeProvider(github.New(cfg.GitHub)). + WithBuildRunner(buildkite.New(cfg.CI)). + WithScorer(heuristic.New()). + WithConflictAnalyzer(tango.New(cfg.Tango)), + ). + WithQueue( + submitqueue.Queue("monorepo/exp"). + WithChangeProvider(github.New(cfg.GitHub)). + WithBuildRunner(local.New()). + WithScorer(heuristic.New()). + WithConflictAnalyzer(fileoverlap.New()), + ). + WithOption(pipeline.TopicNames(cfg.TopicNames)). + WithOption(pipeline.Classifiers(backendClassifiers())). + Build() + if err != nil { return err } + + if err := app.Start(ctx); err != nil { return err } + defer app.Stop(context.Background()) + return app.ServeGRPC(ctx) // convenience: app also holds the RPC controllers +} +``` + +**Template reuse.** `QueueBuilder` is a value type, so a partial builder can serve as a baseline that each queue overrides: + +```go +// Common baseline: every queue uses GitHub and heuristic scoring. +base := submitqueue.Queue(""). + WithChangeProvider(github.New(cfg.GitHub)). + WithScorer(heuristic.New()) + app, err := submitqueue.New(). WithStorage(store). WithMessageQueue(mq). - WithQueue( - submitqueue.Queue("go-code"). - WithChangeProvider(ghProvider). - WithBuildRunner(buildkiteBuildRunner). - WithScorer(defaultScorer). - WithConflictAnalyzer(tango), + // Override only what differs per queue: + WithQueue(base.Named("go-code"). + WithBuildRunner(buildkite.New(cfg.CI)). + WithConflictAnalyzer(tango.New(cfg.Tango)), ). - WithQueue( - submitqueue.Queue("monorepo/exp"). - WithChangeProvider(ghProvider). - WithBuildRunner(localRunner). - WithScorer(defaultScorer). - WithConflictAnalyzer(fileOverlap), + WithQueue(base.Named("monorepo/exp"). + WithBuildRunner(local.New()). + WithConflictAnalyzer(fileoverlap.New()), + ). + WithQueue(base.Named("monorepo/test"). + WithBuildRunner(noop.New()). + WithConflictAnalyzer(noop.NewAnalyzer()), ). Build() +``` -if err != nil { return err } -if err := app.Start(ctx); err != nil { return err } -defer app.Stop(context.Background()) +**Integration test.** The builder produces a `lifecycle.Component` just like the engine, so tests start and stop the full pipeline without special harness code: + +```go +func TestOrchestrator(t *testing.T) { + store := inmemory.NewStorage() + mq := inmemory.NewQueues() + + app, err := submitqueue.New(). + WithStorage(store). + WithMessageQueue(mq). + WithQueue( + submitqueue.Queue("test-queue"). + WithChangeProvider(fake.NewChangeProvider()). + WithBuildRunner(fake.NewBuildRunner()). + WithScorer(constant.New(1.0)). + WithConflictAnalyzer(noop.NewAnalyzer()), + ). + Build() + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + require.NoError(t, app.Start(ctx)) + defer app.Stop(context.Background()) + + // Publish a message to the "start" topic and assert controller behavior. + mq.Publish(t, topickey.Start, landRequestPayload("test-queue", "PR-42")) + // … assertions on store state … +} ``` +## Fluent builder — design and implementation + +The `pipeline.Construct[D]` engine is the foundational API: typed, composable, and testable. The fluent builder wraps it for deployers who wire a single orchestrator with a handful of queues. + ### Implementation sketch ```go @@ -349,6 +551,8 @@ package submitqueue // It is a convenience layer — Build() populates a Deps struct, constructs // profiles, and calls pipeline.Construct under the hood. type Builder struct { + logger *zap.SugaredLogger + scope tally.Scope storage storage.Storage queues messagequeue.Stores perQueue map[string]Profile @@ -358,6 +562,9 @@ type Builder struct { func New() *Builder { return &Builder{perQueue: map[string]Profile{}} } +func (b *Builder) WithLogger(l *zap.SugaredLogger) *Builder { b.logger = l; return b } +func (b *Builder) WithScope(s tally.Scope) *Builder { b.scope = s; return b } + func (b *Builder) WithStorage(s storage.Storage) *Builder { b.storage = s; return b } @@ -375,14 +582,20 @@ func (b *Builder) WithOption(o pipeline.Option) *Builder { } func (b *Builder) Build() (*App, error) { - // Validate required fields (storage, queues, at least one queue). - // Populate Deps from the accumulated state. - // Build profiles registry from perQueue map. - // Call pipeline.Construct(deps, orchestrator.Stages, b.opts...). - // Return App wrapping the lifecycle.Component. + // 1. Validate required fields (storage, queues, at least one queue profile). + // 2. Build a Profiles struct from b.perQueue (same as host profiles.go). + // 3. Populate orchestrator.Deps from the accumulated state: + // deps.BuildRunner = profiles.BuildRunnerFactory() + // deps.ChangeProvider = profiles.ChangeProviderFactory() + // deps.Scorer = profiles.ScorerFactory() + // deps.Analyzer = profiles.AnalyzerFactory() + // 4. Call pipeline.Construct(deps, orchestrator.Stages, b.opts...). + // 5. Return App wrapping the lifecycle.Component + RPC controllers. } // QueueBuilder accumulates per-queue extension selections. +// It is a VALUE TYPE — fluent methods return copies, so partial builders +// are safe to reuse as templates. type QueueBuilder struct { name string profile Profile @@ -390,6 +603,12 @@ type QueueBuilder struct { func Queue(name string) QueueBuilder { return QueueBuilder{name: name} } +// Named returns a copy of this builder with a different queue name. +// Use with template reuse: base := Queue("").WithScorer(...); base.Named("q1") +func (q QueueBuilder) Named(name string) QueueBuilder { + q.name = name; return q +} + func (q QueueBuilder) WithChangeProvider(cp changeprovider.ChangeProvider) QueueBuilder { q.profile.ChangeProvider = cp; return q } @@ -411,7 +630,7 @@ func (q QueueBuilder) WithConflictAnalyzer(a conflict.Analyzer) QueueBuilder { - **Convenience, not replacement.** The builder calls `pipeline.Construct` — it does not bypass or duplicate the engine. Deployers who need full control (custom `Option`s, multi-service composition, fx integration) use the engine directly. - **Compile-time type safety.** Each `With*` method takes the concrete extension interface, not a string hint. A missing or mistyped extension is a compile error. -- **`QueueBuilder` is a value type.** The fluent chain returns copies, not pointers, so partial builders are safe to reuse as templates (e.g. a `baseQueue` with defaults that each real queue overrides). +- **`QueueBuilder` is a value type.** The fluent chain returns copies, not pointers, so partial builders are safe to reuse as templates (e.g. a `baseQueue` with defaults that each real queue overrides — see the template-reuse example above). - **`Build()` validates eagerly.** Missing required fields (no storage, no queues, zero queue profiles) produce a clear error at build time, not a nil-pointer panic at runtime. - **No global state.** `New()` returns an isolated builder. Multiple orchestrator apps can coexist in the same process (useful for integration tests). From f343ca825257cadb33202ddac87d06a428c67eab Mon Sep 17 00:00:00 2001 From: Jamy Timmermans Date: Thu, 16 Jul 2026 11:17:04 -0700 Subject: [PATCH 6/9] fix refs --- doc/rfc/submitqueue/modular-queue-wiring.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/doc/rfc/submitqueue/modular-queue-wiring.md b/doc/rfc/submitqueue/modular-queue-wiring.md index fc504e06..23df0e94 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 477](../../example/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 [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. ## Vocabulary @@ -640,7 +640,7 @@ Profile selection (which scorer/conflict/build-runner a queue uses) deserves sep ### Current role of QueueConfig -`QueueConfig` today is a single-field entity (`Name string`). Its sole consumer is the gateway's `LandController`, which calls `queueconfig.Store.Get(ctx, queue)` to reject requests targeting unknown queues — a pure name-validation gate. The orchestrator does not import `queueconfig` at all; it maintains its own hardcoded `queueRegistry` with no programmatic link to the YAML config. The TODO on line 477 of the orchestrator example envisions bridging the two ("see also queueconfig.Store, which holds the per-queue data half"), but that bridge does not exist today. +`QueueConfig` today is a single-field entity (`Name string`). Its sole consumer is the gateway's `LandController`, which calls `queueconfig.Store.Get(ctx, queue)` to reject requests targeting unknown queues — a pure name-validation gate. The orchestrator does not import `queueconfig` at all; it maintains its own hardcoded `queueRegistry` with no programmatic link to the YAML config. The TODO on line 475 of the orchestrator example envisions bridging the two ("see also queueconfig.Store, which holds the per-queue data half"), but that bridge does not exist today. ### Three options for per-queue extension selection From 4e45d3168633145bc03a3905e7fdff0ce352d167 Mon Sep 17 00:00:00 2001 From: jamy Date: Thu, 16 Jul 2026 18:35:50 +0000 Subject: [PATCH 7/9] docs(submitqueue): rename fluent builder Build() to Create() Avoids confusion between the Builder type and its terminal method, per review feedback from @behindwalls. Co-Authored-By: Claude Opus 4.6 --- doc/rfc/submitqueue/modular-queue-wiring.md | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/doc/rfc/submitqueue/modular-queue-wiring.md b/doc/rfc/submitqueue/modular-queue-wiring.md index 23df0e94..6693ce96 100644 --- a/doc/rfc/submitqueue/modular-queue-wiring.md +++ b/doc/rfc/submitqueue/modular-queue-wiring.md @@ -423,7 +423,7 @@ fx.New( ### Fluent builder — complete example -The same orchestrator, expressed with the builder API. `Build()` calls `pipeline.Construct` internally — the engine is always the assembly mechanism. +The same orchestrator, expressed with the builder API. `Create()` calls `pipeline.Construct` internally — the engine is always the assembly mechanism. ```go package main @@ -470,7 +470,7 @@ func run(ctx context.Context) error { ). WithOption(pipeline.TopicNames(cfg.TopicNames)). WithOption(pipeline.Classifiers(backendClassifiers())). - Build() + Create() if err != nil { return err } if err := app.Start(ctx); err != nil { return err } @@ -503,7 +503,7 @@ app, err := submitqueue.New(). WithBuildRunner(noop.New()). WithConflictAnalyzer(noop.NewAnalyzer()), ). - Build() + Create() ``` **Integration test.** The builder produces a `lifecycle.Component` just like the engine, so tests start and stop the full pipeline without special harness code: @@ -523,7 +523,7 @@ func TestOrchestrator(t *testing.T) { WithScorer(constant.New(1.0)). WithConflictAnalyzer(noop.NewAnalyzer()), ). - Build() + Create() require.NoError(t, err) ctx, cancel := context.WithCancel(context.Background()) @@ -548,7 +548,7 @@ The `pipeline.Construct[D]` engine is the foundational API: typed, composable, a package submitqueue // Builder accumulates configuration for a SubmitQueue orchestrator app. -// It is a convenience layer — Build() populates a Deps struct, constructs +// It is a convenience layer — Create() populates a Deps struct, constructs // profiles, and calls pipeline.Construct under the hood. type Builder struct { logger *zap.SugaredLogger @@ -581,7 +581,7 @@ func (b *Builder) WithOption(o pipeline.Option) *Builder { b.opts = append(b.opts, o); return b } -func (b *Builder) Build() (*App, error) { +func (b *Builder) Create() (*App, error) { // 1. Validate required fields (storage, queues, at least one queue profile). // 2. Build a Profiles struct from b.perQueue (same as host profiles.go). // 3. Populate orchestrator.Deps from the accumulated state: @@ -631,7 +631,7 @@ func (q QueueBuilder) WithConflictAnalyzer(a conflict.Analyzer) QueueBuilder { - **Convenience, not replacement.** The builder calls `pipeline.Construct` — it does not bypass or duplicate the engine. Deployers who need full control (custom `Option`s, multi-service composition, fx integration) use the engine directly. - **Compile-time type safety.** Each `With*` method takes the concrete extension interface, not a string hint. A missing or mistyped extension is a compile error. - **`QueueBuilder` is a value type.** The fluent chain returns copies, not pointers, so partial builders are safe to reuse as templates (e.g. a `baseQueue` with defaults that each real queue overrides — see the template-reuse example above). -- **`Build()` validates eagerly.** Missing required fields (no storage, no queues, zero queue profiles) produce a clear error at build time, not a nil-pointer panic at runtime. +- **`Create()` validates eagerly.** Missing required fields (no storage, no queues, zero queue profiles) produce a clear error at build time, not a nil-pointer panic at runtime. - **No global state.** `New()` returns an isolated builder. Multiple orchestrator apps can coexist in the same process (useful for integration tests). ## Trade-off: profile hints vs. removing QueueConfig entirely From 4a2ab729f4756f7de529714803cd28ed5863fb8a Mon Sep 17 00:00:00 2001 From: jamy Date: Thu, 16 Jul 2026 18:38:21 +0000 Subject: [PATCH 8/9] docs(submitqueue): update modular queue wiring RFC description in index The previous description referenced adding profile hints to QueueConfig, which the RFC's recommended option (C) explicitly avoids. Updated to reflect the current design: a typed engine, service self-declaration, and host-owned profiles. Co-Authored-By: Claude Opus 4.6 --- doc/rfc/index.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/doc/rfc/index.md b/doc/rfc/index.md index 0c4bf92a..fbf53c1f 100644 --- a/doc/rfc/index.md +++ b/doc/rfc/index.md @@ -16,7 +16,7 @@ Design documents and technical proposals, grouped by scope. Shared/cross-cutting - [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 - [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 -- [Modular Queue Wiring](submitqueue/modular-queue-wiring.md) - Extract per-queue extension profiles, topic-registry builder, and controller registration into reusable domain packages; add data-driven profile hints to QueueConfig +- [Modular Queue Wiring](submitqueue/modular-queue-wiring.md) - Declare-don't-assemble engine (`pipeline.Construct`) that unifies topic registry, controller registration, DLQ pairing, and lifecycle ordering into one typed call; services self-declare via Deps struct + Stages slice, hosts own per-queue profiles and transport ## Stovepipe From 5206f02d35e3963c91c37ef26d1b3b0108030716 Mon Sep 17 00:00:00 2001 From: jamy Date: Fri, 17 Jul 2026 19:12:22 +0000 Subject: [PATCH 9/9] docs(submitqueue): drop With prefix from fluent builder methods MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Renames WithLogger→Logger, WithScope→Scope, WithStorage→Storage, WithMessageQueue→MessageQueue, WithQueue→Queue, WithOption→Option, WithChangeProvider→ChangeProvider, WithBuildRunner→BuildRunner, WithScorer→Scorer, WithConflictAnalyzer→ConflictAnalyzer. Also renames Queue() constructor to NewQueue() since Queue is now a Builder method name. Per review feedback from @behindwalls. Co-Authored-By: Claude Opus 4.6 --- doc/rfc/submitqueue/modular-queue-wiring.md | 108 ++++++++++---------- 1 file changed, 54 insertions(+), 54 deletions(-) diff --git a/doc/rfc/submitqueue/modular-queue-wiring.md b/doc/rfc/submitqueue/modular-queue-wiring.md index 6693ce96..0e7e828f 100644 --- a/doc/rfc/submitqueue/modular-queue-wiring.md +++ b/doc/rfc/submitqueue/modular-queue-wiring.md @@ -321,7 +321,7 @@ Two APIs serve different deployer needs. The core engine (`pipeline.Construct`) | Single orchestrator, handful of queues | Fluent builder | Reads top-to-bottom; no boilerplate | | Multiple services in one process | Core engine | Compose multiple `lifecycle.Component`s into a `lifecycle.Group` | | fx / custom DI integration | Core engine | `Construct` returns a `lifecycle.Component` that slots into `fx.Hook` | -| Custom `pipeline.Option`s (classifiers, extra components) | Either | Fluent builder exposes `WithOption()`; core engine takes variadic `Option`s directly | +| Custom `pipeline.Option`s (classifiers, extra components) | Either | Fluent builder exposes `Option()`; core engine takes variadic `Option`s directly | | Integration / e2e tests | Either | Both produce a `lifecycle.Component` with `Start`/`Stop` | ### Core engine — complete example @@ -450,26 +450,26 @@ func run(ctx context.Context) error { mq := mqmysql.New(cfg.QueueDB, logger, scope) app, err := submitqueue.New(). - WithLogger(logger). - WithScope(scope). - WithStorage(store). - WithMessageQueue(mq). - WithQueue( - submitqueue.Queue("go-code"). - WithChangeProvider(github.New(cfg.GitHub)). - WithBuildRunner(buildkite.New(cfg.CI)). - WithScorer(heuristic.New()). - WithConflictAnalyzer(tango.New(cfg.Tango)), + Logger(logger). + Scope(scope). + Storage(store). + MessageQueue(mq). + Queue( + submitqueue.NewQueue("go-code"). + ChangeProvider(github.New(cfg.GitHub)). + BuildRunner(buildkite.New(cfg.CI)). + Scorer(heuristic.New()). + ConflictAnalyzer(tango.New(cfg.Tango)), ). - WithQueue( - submitqueue.Queue("monorepo/exp"). - WithChangeProvider(github.New(cfg.GitHub)). - WithBuildRunner(local.New()). - WithScorer(heuristic.New()). - WithConflictAnalyzer(fileoverlap.New()), + Queue( + submitqueue.NewQueue("monorepo/exp"). + ChangeProvider(github.New(cfg.GitHub)). + BuildRunner(local.New()). + Scorer(heuristic.New()). + ConflictAnalyzer(fileoverlap.New()), ). - WithOption(pipeline.TopicNames(cfg.TopicNames)). - WithOption(pipeline.Classifiers(backendClassifiers())). + Option(pipeline.TopicNames(cfg.TopicNames)). + Option(pipeline.Classifiers(backendClassifiers())). Create() if err != nil { return err } @@ -483,25 +483,25 @@ func run(ctx context.Context) error { ```go // Common baseline: every queue uses GitHub and heuristic scoring. -base := submitqueue.Queue(""). - WithChangeProvider(github.New(cfg.GitHub)). - WithScorer(heuristic.New()) +base := submitqueue.NewQueue(""). + ChangeProvider(github.New(cfg.GitHub)). + Scorer(heuristic.New()) app, err := submitqueue.New(). - WithStorage(store). - WithMessageQueue(mq). + Storage(store). + MessageQueue(mq). // Override only what differs per queue: - WithQueue(base.Named("go-code"). - WithBuildRunner(buildkite.New(cfg.CI)). - WithConflictAnalyzer(tango.New(cfg.Tango)), + Queue(base.Named("go-code"). + BuildRunner(buildkite.New(cfg.CI)). + ConflictAnalyzer(tango.New(cfg.Tango)), ). - WithQueue(base.Named("monorepo/exp"). - WithBuildRunner(local.New()). - WithConflictAnalyzer(fileoverlap.New()), + Queue(base.Named("monorepo/exp"). + BuildRunner(local.New()). + ConflictAnalyzer(fileoverlap.New()), ). - WithQueue(base.Named("monorepo/test"). - WithBuildRunner(noop.New()). - WithConflictAnalyzer(noop.NewAnalyzer()), + Queue(base.Named("monorepo/test"). + BuildRunner(noop.New()). + ConflictAnalyzer(noop.NewAnalyzer()), ). Create() ``` @@ -514,14 +514,14 @@ func TestOrchestrator(t *testing.T) { mq := inmemory.NewQueues() app, err := submitqueue.New(). - WithStorage(store). - WithMessageQueue(mq). - WithQueue( - submitqueue.Queue("test-queue"). - WithChangeProvider(fake.NewChangeProvider()). - WithBuildRunner(fake.NewBuildRunner()). - WithScorer(constant.New(1.0)). - WithConflictAnalyzer(noop.NewAnalyzer()), + Storage(store). + MessageQueue(mq). + Queue( + submitqueue.NewQueue("test-queue"). + ChangeProvider(fake.NewChangeProvider()). + BuildRunner(fake.NewBuildRunner()). + Scorer(constant.New(1.0)). + ConflictAnalyzer(noop.NewAnalyzer()), ). Create() require.NoError(t, err) @@ -562,22 +562,22 @@ type Builder struct { func New() *Builder { return &Builder{perQueue: map[string]Profile{}} } -func (b *Builder) WithLogger(l *zap.SugaredLogger) *Builder { b.logger = l; return b } -func (b *Builder) WithScope(s tally.Scope) *Builder { b.scope = s; return b } +func (b *Builder) Logger(l *zap.SugaredLogger) *Builder { b.logger = l; return b } +func (b *Builder) Scope(s tally.Scope) *Builder { b.scope = s; return b } -func (b *Builder) WithStorage(s storage.Storage) *Builder { +func (b *Builder) Storage(s storage.Storage) *Builder { b.storage = s; return b } -func (b *Builder) WithMessageQueue(q messagequeue.Stores) *Builder { +func (b *Builder) MessageQueue(q messagequeue.Stores) *Builder { b.queues = q; return b } -func (b *Builder) WithQueue(qb QueueBuilder) *Builder { +func (b *Builder) Queue(qb QueueBuilder) *Builder { b.perQueue[qb.name] = qb.profile; return b } -func (b *Builder) WithOption(o pipeline.Option) *Builder { +func (b *Builder) Option(o pipeline.Option) *Builder { b.opts = append(b.opts, o); return b } @@ -601,27 +601,27 @@ type QueueBuilder struct { profile Profile } -func Queue(name string) QueueBuilder { return QueueBuilder{name: name} } +func NewQueue(name string) QueueBuilder { return QueueBuilder{name: name} } // Named returns a copy of this builder with a different queue name. -// Use with template reuse: base := Queue("").WithScorer(...); base.Named("q1") +// Use with template reuse: base := NewQueue("").Scorer(...); base.Named("q1") func (q QueueBuilder) Named(name string) QueueBuilder { q.name = name; return q } -func (q QueueBuilder) WithChangeProvider(cp changeprovider.ChangeProvider) QueueBuilder { +func (q QueueBuilder) ChangeProvider(cp changeprovider.ChangeProvider) QueueBuilder { q.profile.ChangeProvider = cp; return q } -func (q QueueBuilder) WithBuildRunner(br buildrunner.BuildRunner) QueueBuilder { +func (q QueueBuilder) BuildRunner(br buildrunner.BuildRunner) QueueBuilder { q.profile.BuildRunner = br; return q } -func (q QueueBuilder) WithScorer(s scorer.Scorer) QueueBuilder { +func (q QueueBuilder) Scorer(s scorer.Scorer) QueueBuilder { q.profile.Scorer = s; return q } -func (q QueueBuilder) WithConflictAnalyzer(a conflict.Analyzer) QueueBuilder { +func (q QueueBuilder) ConflictAnalyzer(a conflict.Analyzer) QueueBuilder { q.profile.Analyzer = a; return q } ``` @@ -629,7 +629,7 @@ func (q QueueBuilder) WithConflictAnalyzer(a conflict.Analyzer) QueueBuilder { ### Design constraints - **Convenience, not replacement.** The builder calls `pipeline.Construct` — it does not bypass or duplicate the engine. Deployers who need full control (custom `Option`s, multi-service composition, fx integration) use the engine directly. -- **Compile-time type safety.** Each `With*` method takes the concrete extension interface, not a string hint. A missing or mistyped extension is a compile error. +- **Compile-time type safety.** Each fluent method takes the concrete extension interface, not a string hint. A missing or mistyped extension is a compile error. - **`QueueBuilder` is a value type.** The fluent chain returns copies, not pointers, so partial builders are safe to reuse as templates (e.g. a `baseQueue` with defaults that each real queue overrides — see the template-reuse example above). - **`Create()` validates eagerly.** Missing required fields (no storage, no queues, zero queue profiles) produce a clear error at build time, not a nil-pointer panic at runtime. - **No global state.** `New()` returns an isolated builder. Multiple orchestrator apps can coexist in the same process (useful for integration tests).