Skip to main content
← Back to list
01Issue
FeatureShippedSwamp CLI
Assigneesstack72

Relationships

#947 Worker fleets phase 1: elastic queueing

Opened by stack72 · 7/3/2026· Shipped 7/3/2026

Summary

An empty or cold worker pool should queue placed steps instead of failing them. Today, a placed step with no matching connected worker fails immediately (unschedulable). Under fleets, "no workers connected right now" is the steady state between scale-up events — not a configuration mistake. This phase makes queueing the default behavior, makes queued demand visible and queryable as swamp data, and adds boot reconciliation so bookkeeping data never lies after an orchestrator crash.

This is Phase 1 of the worker-fleets design. Everything in phases 2–6 assumes elastic queueing is in place, so this must land first.

Design reference

Full design: worker-fleets.md at the repo root. This issue is self-contained for implementation — read the design doc for broader context, but all changes, invariants, and acceptance criteria are specified here.

Changes

1. Scheduler: collapse unschedulable into queue

File: src/domain/remote/scheduler.ts

ScheduleDecision currently has three variants: dispatch, queue, unschedulable. Delete the unschedulable variant and its reason-string construction. When zero connected workers match, return queue instead. Keep eligibleWorkers filtering as-is.

Extract the deleted reason-string construction into an exported describePlacement(placement) helper — it's reused for timeout errors and step_queued events.

Tests: scheduler_test.ts — empty pool and no-match cases now expect queue, not unschedulable.

2. Dispatch service: layered timeouts, wake-on-enroll

File: src/serve/dispatch_service.ts

  • #acquireWorker (lines ~368–394): remove the unschedulable throw (lines ~383–385). In all non-dispatch cases, wait for a pool change.
  • The deadline becomes layered (innermost wins): per-step queueTimeoutMs (from StepPlacement) → serve-level queueTimeoutMs (constructor option, already exists at line ~110 but is never passed) → DEFAULT_QUEUE_TIMEOUT_MS (change from 600_000 to 60_000 — conservative 60s default for phase 1; phase 6 raises it to 10 min once the queueing UX is proven).
  • 0 at any layer means wait forever — represent as a null deadline, never Date.now() + 0.
  • Timeout error message must restate the unmet placement requirement using the describePlacement() helper from the scheduler.
  • Add notifyWorkerEnrolled(worker) method → calls #wakePoolWaiters(), wired from the gateway's new onWorkerEnrolled callback.

Invariant: recordFirstWrite ordering must be preserved. The data plane awaits the lease's mark_writes before the first durable write persists. Any refactor of the wait/dispatch loop that reorders this is a correctness bug.

Invariant: sole-writer rule. All pending-dispatch transitions must serialize through the same #leaseTransition-style tail as lease and worker transitions. The datastore has no compare-and-swap; this serialization is the only concurrency control.

3. step_queued event pipeline

This needs to thread through four layers — skipping any layer means the event doesn't reach users:

  1. DispatchService: emit a new RpcStreamEvent kind queued (payload: the describePlacement string) on each entry to the wait state.
  2. method_execution_service.ts #executeRemotely (~371–383): translate the queued stream event to a new MethodExecutionEvent variant. Currently this bridge only forwards kind === "method_event" frames — add queued handling.
  3. execution_service.ts (~989–1010): wrap as a new WorkflowExecutionEvent kind step_queued (in src/domain/workflows/execution_events.ts).
  4. Renderers (src/presentation/renderers/workflow_run.ts and workflow_run_tree/): display "waiting for a worker matching …".

The serve codec (src/serve/serializer.ts) needs no structural change — serializeEvent is a generic JSON-safe clone — only a round-trip test in serializer_test.ts.

4. Wake-on-enroll

Files: src/serve/worker_gateway.ts, src/cli/commands/serve.ts

Add an onWorkerEnrolled gateway callback fired after a successful enroll (and after re-enroll within the grace window). Wire it to DispatchService.notifyWorkerEnrolled() → #wakePoolWaiters(), next to the existing onWorkerIdle / onGraceExpired wiring at serve.ts:592-593.

Today, only onWorkerIdle and onGraceExpired are wired. onWorkerDisconnected is declared but not wired. The missing signal is enrollment — a step queued against an empty pool must dispatch the moment the first matching worker dials home.

5. queueTimeout schema and threading

File: src/domain/workflows/step.ts (next to target/labels/platform, lines ~88–93)

New optional field queueTimeout — a number of seconds (not a duration string). This matches the manual_approval timeout precedent in step_task.ts and keeps duration-string parsing out of the domain layer. The domain must NOT import src/cli/duration_parser.ts — this is enforced by integration/ddd_layer_rules_test.ts.

Thread it by carrying queueTimeoutMs inside StepPlacement (src/domain/remote/scheduler.ts), so it rides the existing placement path:

Step.placement → StepExecutionContext (execution_service.ts ~984) → MethodContext (src/domain/models/model.ts) → #executeRemotely → RemoteStepRequest (src/domain/remote/remote_dispatch.ts) → the dispatch deadline.

The serve-level default is a --queue-timeout <duration> flag on swamp serve, parsed CLI-side with parseTimeout from src/cli/duration_parser.ts (already imported by serve.ts), passed as the queueTimeoutMs constructor option.

Include queueTimeout in the workflow step serialization round-trip.

6. Pending-dispatch built-in model

New file: src/domain/models/worker/pending_dispatch_model.ts

Mirror step_lease_model.ts — all records as named items under a single instance ("pending").

States: waiting → dispatched | timed_out | cancelled (+ orphaned, set only by boot reconciliation)

Fields: placement requirement (target/labels/platform), workflow/job/step names, model type + method, queuedAt, endedAt.

Methods: enqueue, mark_dispatched, timeout, cancel, orphan — following the lease model's naming style.

Record identity: A dispatchId doesn't exist at queue time (it's minted inside #dispatchOnce), so mint a queueId per wait episode — each time executeRemote enters the wait state, including re-entries after a worker_busy desync or a lost no-write dispatch. Record the dispatchId on mark_dispatched. One record per episode (a re-queued step gets a fresh record and re-emits step_queued).

Declare bounded garbageCollection (match the other bookkeeping models, likely 20). Register in the barrel src/domain/models/models.ts and add to its built-in types list.

DispatchService writes these transitions through the same serialized transition-tail pattern (sole-writer rule).

7. Queue introspection

  • Serve protocol: worker.queue.list request/response frame (admin read). Mirror worker.list in src/serve/protocol.ts and its schema + dispatch case in src/serve/connection.ts.
  • libswamp: workerQueueList operation reading the pending-dispatch data from the datastore (source of truth — not the dispatch service's in-memory waiters). Template: src/libswamp/worker/list.ts. Export via src/libswamp/mod.ts — CLI must NOT import internal paths.
  • CLI: swamp worker queue command under the existing worker group. Dual-mode pattern: log + json output. Show waiting entries (requirement, step, age). Template: worker_list.ts + render function in src/presentation/output/worker_output.ts.

8. Boot reconciliation

File: src/cli/commands/serve.ts (~572–607, before Deno.serve accepts traffic)

At serve startup, sweep the three bookkeeping models:

  • active step-leases → expired (reason: orchestrator restart)
  • waiting pending-dispatches → orphaned
  • Workers with connected-looking status → disconnected

Query with the datastore-read pattern from src/libswamp/worker/list.ts. Transition through the model-method runner (createWorkerModelRunDeps, skipAllReports: true). Log a one-line summary of what was swept.

Invariants to preserve

These are correctness-critical — violating any of them is a bug:

  1. Sole-writer rule. All worker/token/lease/pending-dispatch transitions are serialized inside the orchestrator process. The datastore has no compare-and-swap.
  2. recordFirstWrite ordering. The data plane awaits the lease's mark_writes before the first durable write persists.
  3. Placement never falls back silently. A placed step with no dispatcher or no match must queue loudly — never run loopback.
  4. Protocol discipline. Any change to enroll/dispatch/frame shapes bumps REMOTE_PROTOCOL_VERSION; the denylist in environment_snapshot.ts is pinned to it.
  5. Workers stay repo-less. No new worker-side state, config, or secrets.
  6. forEach sharing. ForEachExpansionService expanded instances share the template's Step object and its placement — don't break this.
  7. Domain layer boundary. The domain must NOT import CLI-layer modules (duration_parser.ts). Enforced by integration/ddd_layer_rules_test.ts.

Tests required

  • Scheduler unit tests: empty pool → queue; no-match → queue; describePlacement output.
  • Dispatch service tests: wake-on-enroll (enroll after executeRemote is already waiting); layered timeout resolution (step > serve > default); timeout message contains placement description; abort-while-queued cancels cleanly; pending-dispatch transitions including dispatch/timeout races.
  • Serializer round-trip: step_queued event survives serialize → deserialize.
  • Pending-dispatch model: state machine transitions; garbageCollection declared.
  • Boot reconciliation: seed stale records (active lease, waiting pending-dispatch, connected-looking worker), boot, assert all transitioned correctly.
  • Integration test: placed workflow starts with zero workers → worker connects → run completes. This is the golden path — extend the existing startOrchestrator() + runWorker() pattern in integration/remote_execution_test.ts.

Important test conventions:

  • Tests live next to source: foo.ts → foo_test.ts.
  • Named as Deno.test("functionName: describes behavior", ...).
  • CLI command tests need await initializeLogging({}) and import "../../domain/models/models.ts" (see src/cli/commands/data_get_test.ts).
  • Cross-component changes (protocol frames, schema fields, model shapes) need integration tests under integration/.
  • Tests must pass on Linux/macOS/Windows. Use @std/path helpers and assertPathEquals for paths.

Acceptance criteria

  1. A placed workflow run against an empty pool waits, emits step_queued, shows in swamp worker queue, and completes when a matching worker enrolls.
  2. The same run times out with a requirement-naming error when nothing enrolls within the timeout.
  3. swamp serve restart marks orphaned records and the summary line appears in logs.
  4. All existing remote-execution tests continue to pass (the unschedulable test cases change to expect queue).
  5. deno check, deno lint, deno fmt, deno run test, deno run compile all pass.

Repo conventions

  • New .ts files need the AGPLv3 header — run deno run license-headers.
  • TypeScript strict, no any, named exports only.
  • CLI commands and renderers import libswamp only via src/libswamp/mod.ts.
  • Every new command supports both log and json output modes.
  • Use the ddd skill when shaping new domain objects.
  • Use the github-pr skill for commits/PRs.
  • Keep blast radius small — one phase per PR series, no adjacent refactors.
  • Update worker-fleets.md and design/remote-execution.md when implementation diverges from the plan.
02Bog Flow
✓OPEN✓TRIAGED✓IN PROGRESS✓SHIPPED+ 1 MOREASSIGNED+ 5 MOREREVIEW+ 4 MOREPR_MERGED+ 1 MORENOTIFICATION_SKIPPED

Shipped

7/3/2026, 10:56:36 PM

Click a lifecycle step above to view its details.

03Sludge Pulse
stack72 assigned stack727/3/2026, 9:03:22 PM

Sign in to post a ripple.