Skip to content

modular-queue-wiring.md

Latest commit

 

History

History
674 lines (542 loc) · 33.8 KB

modular-queue-wiring.md

File metadata and controls

674 lines (542 loc) · 33.8 KB
Raw

Modular Queue Wiring

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

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). The topic → subscription → DLQ subscription → DLQ controller linkage is maintained by copy-paste across 12 stages, where forgetting any half creates a silent failure.

The TODO on line 475 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

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

Principle

  • 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.
┌───────────────────────── 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  │
└─────────────────────────────────────────────────────────┘

Proposal

Step 1 · platform/lifecycle — the one interface a host ever sees

// 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
}

// 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 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.

Step 2 · platform/pipeline — the engine

// 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

    // 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)

    // DLQ, when non-nil, declares "this stage dead-letters". The engine then
    // derives the paired DLQ topic (<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)
}
// 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
}

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.

Step 3 · Service self-declaration — submitqueue/orchestrator/pipeline.go

A service's entire definition is three declarations — a struct (seams), a slice (topology), a constructor (controllers). No assembly code:

// ① 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
}

// ② 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
}

// ③ 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)}
}

Step 4 · Host-private profiles — service/.../profiles.go

Profiles stay entirely in the host. Nothing profile-shaped crosses the library boundary:

type Profile struct {
    BuildRunner    buildrunner.BuildRunner
    ChangeProvider changeprovider.Provider
    Scorer         scorer.Scorer
    Analyzer       conflict.Analyzer
}

type Profiles struct {
    byQueue        map[string]Profile
    defaultProfile Profile
}

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()},
    }
}

// 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) }

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

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.

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)
}

At delivery time — all the pieces meeting

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

Generalizes across all four services

                 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
────────────────────────────────────────────────────────────────────────────────────────────

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 the engine enforces

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)

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

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.Components into a lifecycle.Group
fx / custom DI integration Core engine Construct returns a lifecycle.Component that slots into fx.Hook
Custom pipeline.Options (classifiers, extra components) Either Fluent builder exposes Option(); core engine takes variadic Options directly
Integration / e2e tests Either Both produce a lifecycle.Component with Start/Stop

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.

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:

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:

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. Create() calls pipeline.Construct internally — the engine is always the assembly mechanism.

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().
        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)),
        ).
        Queue(
            submitqueue.NewQueue("monorepo/exp").
                ChangeProvider(github.New(cfg.GitHub)).
                BuildRunner(local.New()).
                Scorer(heuristic.New()).
                ConflictAnalyzer(fileoverlap.New()),
        ).
        Option(pipeline.TopicNames(cfg.TopicNames)).
        Option(pipeline.Classifiers(backendClassifiers())).
        Create()
    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:

// Common baseline: every queue uses GitHub and heuristic scoring.
base := submitqueue.NewQueue("").
    ChangeProvider(github.New(cfg.GitHub)).
    Scorer(heuristic.New())

app, err := submitqueue.New().
    Storage(store).
    MessageQueue(mq).
    // Override only what differs per queue:
    Queue(base.Named("go-code").
        BuildRunner(buildkite.New(cfg.CI)).
        ConflictAnalyzer(tango.New(cfg.Tango)),
    ).
    Queue(base.Named("monorepo/exp").
        BuildRunner(local.New()).
        ConflictAnalyzer(fileoverlap.New()),
    ).
    Queue(base.Named("monorepo/test").
        BuildRunner(noop.New()).
        ConflictAnalyzer(noop.NewAnalyzer()),
    ).
    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:

func TestOrchestrator(t *testing.T) {
    store := inmemory.NewStorage()
    mq := inmemory.NewQueues()

    app, err := submitqueue.New().
        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)

    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

// submitqueue/builder.go
package submitqueue

// Builder accumulates configuration for a SubmitQueue orchestrator app.
// 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
    scope    tally.Scope
    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) 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) Storage(s storage.Storage) *Builder {
    b.storage = s; return b
}

func (b *Builder) MessageQueue(q messagequeue.Stores) *Builder {
    b.queues = q; return b
}

func (b *Builder) Queue(qb QueueBuilder) *Builder {
    b.perQueue[qb.name] = qb.profile; return b
}

func (b *Builder) Option(o pipeline.Option) *Builder {
    b.opts = append(b.opts, o); return b
}

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:
    //        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
}

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 := NewQueue("").Scorer(...); base.Named("q1")
func (q QueueBuilder) Named(name string) QueueBuilder {
    q.name = name; return q
}

func (q QueueBuilder) ChangeProvider(cp changeprovider.ChangeProvider) QueueBuilder {
    q.profile.ChangeProvider = cp; return q
}

func (q QueueBuilder) BuildRunner(br buildrunner.BuildRunner) QueueBuilder {
    q.profile.BuildRunner = br; return q
}

func (q QueueBuilder) Scorer(s scorer.Scorer) QueueBuilder {
    q.profile.Scorer = s; return q
}

func (q QueueBuilder) ConflictAnalyzer(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 Options, multi-service composition, fx integration) use the engine directly.
  • 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).

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.

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 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

Option Description Pros Cons
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 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). pipeline.Construct is not DI: no runtime graph, no reflection, no topo-sort — a typed engine over declarative data. One-offs enter via Options. 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.

Migration path

The refactor can land incrementally:

  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 pure additions with zero behavioral change. Step 4 is the switch-over.