# 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: ```yaml 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 ```mermaid 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: ```text 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 ```text 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): ```go engs["microvm"] = microvmEng engs["nixery"] = nixeryEng ``` broker spindle: ```go engs["microvm"] = broker.NewEngine("microvm", fleetBroker) engs["nixery"] = broker.NewEngine("nixery", fleetBroker) ``` executor spindle (real engines, plus a shim that dials the broker): ```go engs["microvm"] = microvmEng engs["nixery"] = nixeryEng executorShim := executor.New(brokerURL, engs, ...) executorShim.Connect() ``` ## end-to-end workflow ```mermaid 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): ```go 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 `ReserveSeat`s (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. ```mermaid 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: ```go 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. ```mermaid 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). ```go 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`.