broker -> executor architecture notes #

right now a spindle is one box: it takes a pipeline event, picks an engine, runs the job locally, all in one process. this is about splitting that into two roles, without rewriting the execution path (that's the whole constraint really).

• broker spindle: the one the repo points at. it takes pipeline events, decides which machine runs each job, and holds the backlog of jobs waiting for a spot. it doesn't run anything itself.
• executor spindle: a normal spindle that actually runs jobs, it just dials out to a broker and takes work from it instead of being pointed at directly.

the trick is that the broker pretends to be a normal engine. users still write:

engine: microvm

on a standalone spindle microvm is the real microvm engine. on a broker spindle microvm is the broker engine, which leases a seat from an executor that then runs the real microvm engine. nothing in the pipeline file changes, which is kinda the point.

the model is fly-ish / bidding: the broker keeps a rough view of the fleet and asks nodes for a seat, each executor holds the real local truth and can say no. so you get an orchestrator/supplier split without turning spindle into kubernetes, and local truth stays local (the broker never has to know what's actually going on inside a spindle).

shape #

flowchart TD
    user["pipeline event / user action"] --> brokerSpindle["broker spindle"]

    subgraph brokerSpindle["broker spindle"]
        pending["pending backlog + maxPending (no worker pool)"]
        brokerEngine["broker engine registered as microvm/nixery"]
        brokerState["fleet view + sessions + leases"]
    end

    brokerSpindle <--> stream["one websocket per executor (framed protobuf)"]

    stream <--> ex1["executor spindle hel-01"]
    stream <--> ex2["executor spindle hel-02"]
    stream <--> ex3["executor spindle homebox-01"]

    subgraph ex1["executor spindle"]
        executorShim1["executor shim"]
        microvm1["actual microvm engine"]
        nixery1["actual nixery engine"]
    end

    subgraph ex2["executor spindle"]
        executorShim2["executor shim"]
        microvm2["actual microvm engine"]
        nixery2["actual nixery engine"]
    end

what already exists that we lean on #

we have this right now:

server.processPipeline
  -> parse sh.tangled.pipeline
  -> pick engine from s.engs[w.Engine]
  -> eng.InitWorkflow(...)
  -> inject TANGLED_* env
  -> db.StatusPending(...)
  -> queue.Enqueue(...)

engine.StartWorkflows
  -> get secrets from vault
  -> open workflow logger
  -> if engine implements WorkflowSlotter, AcquireWorkflowSlot(...)
  -> db.StatusRunning(...)
  -> eng.SetupWorkflow(...)
  -> for each step: eng.RunStep(...)
  -> optional FinalizeWorkflow(...)
  -> db.StatusSuccess / StatusFailed / StatusTimeout
  -> eng.DestroyWorkflow(...)

the nice thing is the one hook we need is already there. AcquireWorkflowSlot is an optional interface (engine.WorkflowSlotter), called in StartWorkflows right before the job is marked running, with a defer slot.Release(). and both real engines already go through it:

• nixery slots with a SemaphoreSlotter (count of concurrent workflows).
• microvm slots with a ResourceScheduler in budget.go, which queues until the job's memory/vcpu/disk fits the node budget.

so "wait for a remote seat" is just a slow AcquireWorkflowSlot, and the time spent waiting is already shown to the user as pending. that's the whole idea: the broker is an engine whose AcquireWorkflowSlot does remote placement.

terminology #

broker spindle: the spindle the repo points at. receives pipeline events, owns
the pending backlog and candidate selection. does not run jobs.

broker: an Engine registered under the normal names ("microvm", "nixery"). its
AcquireWorkflowSlot leases a seat from an executor instead of running locally.

executor spindle: a normal spindle configured with a broker url. connects
outbound to the broker.

executor (shim): the bit beside the real engines that handles
reserve/commit/release/cancel over the session.

execution engine: the actual runtime: microvm, nixery, docker, etc. unchanged.

registration model #

standalone spindle (today):

engs["microvm"] = microvmEng
engs["nixery"] = nixeryEng

broker spindle:

engs["microvm"] = broker.NewEngine("microvm", fleetBroker)
engs["nixery"] = broker.NewEngine("nixery", fleetBroker)

executor spindle (real engines, plus a shim that dials the broker):

engs["microvm"] = microvmEng
engs["nixery"] = nixeryEng

executorShim := executor.New(brokerURL, engs, ...)
executorShim.Connect()

end-to-end workflow #

sequenceDiagram
    participant Knot as pipeline event
    participant BrokerSpindle as broker spindle
    participant Broker as broker engine
    participant Executor as executor spindle
    participant Engine as actual microvm/nixery engine

    Knot->>BrokerSpindle: sh.tangled.pipeline
    BrokerSpindle->>Broker: InitWorkflow(twf, tpl)
    Broker-->>BrokerSpindle: synthetic one-step workflow
    BrokerSpindle->>BrokerSpindle: StatusPending + hand to placement

    BrokerSpindle->>Broker: AcquireWorkflowSlot(wid, wf)
    Broker->>Broker: rank live executors from snapshots
    Broker->>Executor: ReserveSeat(attempt, target_engine, raw workflow)
    Executor->>Engine: InitWorkflow(twf, tpl)
    Executor->>Engine: TryAcquireWorkflowSlot(...)
    Engine-->>Executor: local slot (or reject now)
    Executor-->>Broker: accepted lease
    Broker-->>BrokerSpindle: remote seat

    BrokerSpindle->>BrokerSpindle: StatusRunning
    BrokerSpindle->>Broker: SetupWorkflow + RunStep("remote execution")
    Broker->>Executor: CommitLease(lease_id, secrets)
    Executor->>Engine: StartWorkflows with the pre-acquired slot
    Engine-->>Executor: terminal result
    Executor-->>Broker: status + log events relayed over session
    Broker-->>BrokerSpindle: nil / failed / timeout

    BrokerSpindle->>BrokerSpindle: re-insert executor status into eventstream
    BrokerSpindle->>Broker: DestroyWorkflow(...)
    Broker->>Executor: cleanup/cancel if needed

the broker's synthetic workflow is one step, remote execution. we don't mirror the executor's real steps in the broker lifecycle, there's no point.

broker engine lifecycle #

InitWorkflow #

don't parse the engine config here. just stash the raw stuff into wf.Data (which is already any, both real engines use it for their own state):

type brokerWorkflowState struct {
    TargetEngine string
    RawWorkflow  tangled.Pipeline_Workflow
    RawPipeline  tangled.Pipeline
    Lease        *RemoteLease
    Attempt      *RemoteAttempt
}

return a synthetic one-step workflow so processPipeline injects TANGLED_* env normally. note this means InitWorkflow runs twice, once here on the broker (synthetic, no config parsing) and once for real on the executor inside ReserveSeat. that's fine and intended.

AcquireWorkflowSlot #

this is where placement happens:

• read the target engine type out of wf.Data
• rank live executors from the fleet snapshots, ask the top k for a seat
• keep the winner's lease, release the losers

one thing to be honest about: the broker can't rank by exact resource fit. the memory/vcpu/disk of a microvm job is computed inside microvm's own InitWorkflow (from the image spec), which only runs on the executor. so the broker ranks on what it actually has: which executors are alive, the right arch, and a coarse "do they have free seats" from snapshots. the real yes/no is the executor's call, see ReserveSeat below.

placement blocks here: if no executor will take the job right now, AcquireWorkflowSlot keeps trying instead of returning, which the user sees as pending. see "concurrency" below for why that's fine on a broker.

SetupWorkflow / RunStep #

SetupWorkflow is basically a no-op info log. RunStep is commit-and-wait:

• send CommitLease with the secrets
• wait for the terminal result over the session

DestroyWorkflow #

unwind: release reserved leases, cancel running attempts, no-op if the job already finished.

concurrency #

the real concurrency limit is the executors, not the broker. a job can only run if some executor granted it a seat, and executors already cap their own seats (nixery's semaphore, microvm's resource budget). so the broker cannot have more jobs running than the fleet has capacity for, and any limit the broker puts on running jobs can only ever cap below true fleet capacity, which is just harmful. the broker brokers, it doesn't ration.

that has one consequence for the standalone code: a standalone spindle runs jobs in a bounded worker pool (queue.Queue), because it's doing real local work and must bound it. a broker spindle must not reuse that bounded pool, because placement waits inside AcquireWorkflowSlot, which runs inside a worker. with a fixed pool, a handful of jobs waiting for seats would hold every worker and starve other jobs that could actually be placed (especially in a mixed fleet, where "full" is per-arch / per-engine, not fleet-wide). so on a broker spindle, "run the job" is just spawn a goroutine, unbounded. each one is cheap: block in AcquireWorkflowSlot until a seat opens, then commit and wait for the result.

the only thing the broker does bound is how many jobs are waiting, which is just backpressure ("stop accepting past N pending", the role queue.maxSize plays today). the broker engine owns that number itself, a maxPending it checks in AcquireWorkflowSlot, nobody else needs to know about it. no separate scheduler component: the broker engine is a singleton, its AcquireWorkflowSlot is called concurrently by every waiting job, so the shared state (fleet view, pending count) just lives on the engine struct and the calls rendezvous on it.

fairness we deliberately punt on for v1: seat allocation is racy, every waiting job blocks in AcquireWorkflowSlot running its own bidding loop, and when a seat opens whoever grabs it wins. that's fine for a small fleet. when we do want per-user fairness (the thing queue.pickBest gives standalone spindles), it's a waiter list on the broker engine that the blocked calls order themselves through, still the same AcquireWorkflowSlot seam, no new subsystem. so it's a later refinement, not a redesign.

executor shim lifecycle #

lives next to the real engines, reads messages off the outbound session.

reserve #

check constraints, run actualEngine.InitWorkflow, try to grab a local slot.

it must not queue locally. if it can run now, accept. if it can't, reject immediately. the broker owns the backlog, the executor doesn't queue at all (it's basically "max 0 queued", it either has room right now or it says no). this is the one genuinely new bit of engine work: today both slotters block (the semaphore waits on ctx, the scheduler queues with aging), so we need a non-blocking TryAcquireWorkflowSlot on both SemaphoreSlotter and ResourceScheduler that tries once and returns ErrNoWorkflowSlots instead of waiting.

commit / release / cancel #

CommitLease marks the reservation committed and runs the real engine with the slot it already holds. ReleaseLease releases the slot (also fires on TTL expiry). CancelAttempt cancels the context and calls actualEngine.DestroyWorkflow.

graceful shutdown #

this is entirely executor-side, the broker doesn't (and shouldn't) know a drain is happening. when an executor is told to wind down (maintenance, redeploy, someone closing their laptop) it just: rejects new ReserveSeats (same path it uses when full), reports 0 free seats in its next snapshot so the broker stops even bidding it, lets its in-flight jobs run to completion, then closes the session and exits. the broker sees the session close and treats it as gone, which is all it ever needed to know. so a draining node is indistinguishable from a full one until the moment it leaves... which is exactly the point, less for the broker to track.

(this is the opposite of a node that dies mid-run, see leasing, where the work is genuinely lost and we fail the job. draining is cooperative, so we don't have to lose anything)

leasing #

a lease is a short-lived fencing token with idempotent states. if the broker dies before commit, the TTL cleans the reservation up. a brief session blip where the executor process is still alive doesn't kill a running job, the cursor offset (see node session api) replays whatever the broker missed.

what we don't try to recover is an executor that actually dies mid-run. the microvm and its work are gone with it, so once a committed job's executor dies, the broker fails the job. no waiting around, no half-state to reconcile.

stateDiagram-v2
    [*] --> Reserved: ReserveSeat accepted
    Reserved --> Released: broker chose another executor
    Reserved --> Expired: ttl elapsed before commit
    Reserved --> Committed: CommitLease
    Committed --> Running: executor starts workflow
    Running --> Succeeded: terminal result
    Running --> Failed: terminal result
    Running --> TimedOut: timeout/cancel context
    Running --> Cancelled: CancelAttempt
    Succeeded --> [*]
    Failed --> [*]
    TimedOut --> [*]
    Cancelled --> [*]
    Released --> [*]
    Expired --> [*]

the Committed edge matters for status (below): it's the line where the broker stops owning the user-facing status and the executor takes over.

status: who owns it #

status today is not atproto records. each spindle writes PipelineStatus rows into its own local events table and serves them over its /events SSE endpoint with a cursor, and the appview subscribes to that endpoint. so the question is purely "which box is the appview allowed to read from", and since executors connect outbound (for NAT, see below) the appview can't read a home-box executor directly. the broker has to stay the front door.

so we split authorship on the lease boundary, not the data path:

• the broker authors pending, and the one terminal it actually owns: couldn't place the job / every executor is dead -> failed. that's the only status the broker writes on its own.
• at CommitLease, authorship moves to the executor. it's a normal spindle, it already writes running/success/failed/timeout/cancelled into its own eventstream. it relays those events over the session it's already holding, the broker re-inserts them into its eventstream, and the appview (already subscribed to the broker) sees them with no changes.

so the "double bookkeeping" i was worried about isn't a bug, it's just the relay: the executor is the source of truth, the broker is a cache + relay so the appview can reach it. this reuses everything we already have, eventstream.Event, PipelineStatusNSID, the notifier, and the cursor seq offset already replays missed status (and log) events on reconnect. no new status schema, no appview changes at all ^^

logs #

same pattern as status, and it falls out of how /logs already works: the endpoint (streamLogsFromDisk) reads the spindle's local disk (LogDir), not S3. the S3 upload in StartWorkflows is just archival, it's not in the serve path at all.

so the whole thing is just the relay plus the existing endpoint:

• live tail and after: the executor relays log lines over the session, the broker writes them to its own LogDir at the same wid path, and the existing /logs websocket serves them with zero changes. the appview never knows the job ran elsewhere.
• S3 is configured on the executors, not the broker. the executor is the one running the job, so it owns its durable archive, same as a standalone spindle today. the broker doesn't need S3 configured at all: it already has the bytes on local disk from the relay, so it serves from disk.

the rule that falls out of all this: the broker is always the front door, the executor is the source of truth after commit, and everything flows over the one session we already keep open.

node session api #

executors dial the broker (outbound), so NAT just stops being a problem, home boxes and weird network segments work without anyone poking holes inward. it's also nice if people want to cluster spindles at home, or run them on networks where you can't deal with NAT cleanly, minor, but it's free qol.

the session is one websocket per executor carrying length-prefixed protobuf messages. we already have this framing in agentproto (it's how the spindle talks to the microvm guest over vsock) and we already terminate websockets, so there's no reason to pull in grpc, we just only have to do message parsing, which we already do right now so.

the executor keeps a sequential event log of broker-facing events. on reconnect it sends its latest offset (cursor) so the broker can replay anything it missed without breaking terminal-state idempotency. status and log events ride this same log, so the relay above survives a brief reconnect.

secrets #

zero secrets during ReserveSeat. they only go to the winning node inside CommitLease. on the executor they sit in a temporary in-memory vault wrapper so they never touch disk.

avoiding double resource accounting #

ReserveSeat already grabbed a real local slot. when the executor then runs the normal StartWorkflows, that code calls AcquireWorkflowSlot again, which would acquire a second slot. so we wrap the real engine in a reservedEngine shim whose AcquireWorkflowSlot hands back the already-held slot exactly once:

type reservedEngine struct {
    inner models.Engine
    slot  engine.WorkflowSlot
    once  sync.Once
}

func (e *reservedEngine) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow) (engine.WorkflowSlot, error) {
    var slot engine.WorkflowSlot
    e.once.Do(func() {
        slot = e.slot
        e.slot = nil
    })
    if slot == nil {
        return nil, fmt.Errorf("reserved slot already consumed")
    }
    return slot, nil
}

this shim is the only change on the executor's execution path, everything else is the plain standalone spindle.

candidate bidding #

bounded concurrent bidding: rank from snapshots, ask the top k, take the winner, release the rest.

sequenceDiagram
    participant Broker
    participant A as executor a
    participant B as executor b
    participant C as executor c

    Broker->>Broker: rank candidates from snapshots
    par ask top k
        Broker->>A: ReserveSeat
        Broker->>B: ReserveSeat
        Broker->>C: ReserveSeat
    end

    A-->>Broker: reject: disk pressure
    B-->>Broker: accept lease_b score 0.80
    C-->>Broker: accept lease_c score 0.62

    Broker->>B: CommitLease
    Broker->>C: ReleaseLease

resource tracking / snapshots #

hello messages carry static traits (arch, engines, capabilities). dynamic snapshots carry the changing state (free seats, memory/vcpu budget).

type NodeSnapshot struct {
    NodeID     string
    Seq        uint64
    ObservedAt time.Time
    Engines    map[string]EngineSnapshot
}

note there's no "draining" flag here on purpose, see graceful shutdown below. the broker only ever needs "0 free seats" or "session gone", which it already reacts to, so a draining node just looks like a full one until it exits.

the broker filters by session liveness and arch and ranks on coarse free capacity. it does not pretend to know exact per-job resource fit (it can't, see AcquireWorkflowSlot), the executor decides that in ReserveSeat. snapshots push on connect, periodically, and right after any state change (reserve, commit, terminal).

code changes #

we don't have to change core that much:

• spindle/fleet/broker: implements models.Engine and engine.WorkflowSlotter. its AcquireWorkflowSlot is the placement logic.
• spindle/fleet/executor: the session loop, the reservation table, the reservedEngine token, and relaying status/log events back to the broker.
• TryWorkflowSlotter: a new optional interface for non-blocking slot checks (TryAcquireWorkflowSlot), implemented on SemaphoreSlotter and ResourceScheduler.
• broker job execution is unbounded: a broker spindle spawns a goroutine per job instead of running through the bounded queue.Queue (see concurrency). the executor keeps queue.Queue unchanged.

pipeline parsing, the workflow lifecycle, db schema, and the raw execution loops stay untouched. if no executor accepts, the broker's AcquireWorkflowSlot keeps retrying, which the user just sees as pending, until it gives up and writes failed.