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

Relationships

#989 Worker fleets phase 4b: capacity slots and concurrent dispatch

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

Summary

Phase 4b of worker fleets. Phase 4a shipped the dispatch runner (child process per dispatch). This PR adds concurrent dispatch: a worker advertises N slots, the scheduler packs least-loaded, and multiple runners execute in parallel with per-dispatch environment isolation.

Today, each worker runs one dispatch at a time (`busy` boolean guard). After this PR, `--concurrency 8` lets a single large machine run 8 steps concurrently — same isolation semantics as 8 separate workers but on one box.

Design reference: `worker-fleets.md` section "Phase 4 — Capacity: slots and dispatch runners" → subsection (b).

Changes

1. `--concurrency` flag on `worker connect`

File: `src/cli/commands/worker_connect.ts`

Add flag: ``` --concurrency <n:string> Number of concurrent dispatch slots ("auto" = CPU count, min 1). Default: 1 ```

Env var: `SWAMP_WORKER_CONCURRENCY`

Parse: integer ≥ 1, or the string `"auto"` which resolves to `navigator.hardwareConcurrency ?? 1` (Deno's CPU count API). Pass as `concurrency: number` to `runWorker`.

Thread into `RunWorkerOptions` → `connectOnce` → `registerDispatchHandler`.

2. Send capacity at enrollment

File: `src/domain/remote/protocol.ts`

The `EnrollParamsSchema` already has `resourceLimits: z.record(z.string(), z.unknown()).optional()` (line 184) — it's dead code, never sent. Wire it:

```typescript resourceLimits: { capacity: concurrency } ```

No schema change needed — the field exists. Bump `REMOTE_PROTOCOL_VERSION` (3→4) because the field's semantics are now load-bearing.

3. Worker-side: active-runner count replaces busy boolean

File: `src/worker/dispatch_handler.ts`

Replace: ```typescript let busy = false; ```

With: ```typescript let activeRunners = 0; const capacity: number; // from options ```

Dispatch guard: ```typescript if (draining) throw RpcError("worker_draining"); if (activeRunners >= capacity) throw RpcError("worker_busy"); activeRunners++; try { ... } finally { activeRunners--; ... } ```

`worker_busy` remains the over-capacity desync answer — same error code, same re-queue behavior on the orchestrator side.

Drain semantics: `drain()` now resolves when `activeRunners === 0` (not when one dispatch finishes). The idle-timeout (phase 3) also means "zero active runners."

4. Gateway: capacity + activeDispatchIds

File: `src/serve/worker_gateway.ts`

`WorkerSnapshot` and `WorkerEntry`: Replace: ```typescript status: "idle" | "busy" | "unverified" | "draining"; dispatchId: string | null; ```

With: ```typescript status: "idle" | "busy" | "unverified" | "draining"; capacity: number; activeDispatchIds: string[]; ```

Derive `status` for display:

  • `activeDispatchIds.length === 0` → `"idle"`
  • `activeDispatchIds.length > 0 && activeDispatchIds.length < capacity` → `"busy"` (partially loaded)
  • `activeDispatchIds.length >= capacity` → `"busy"` (fully loaded)
  • `unverified` and `draining` stay as explicit states

Enrollment: Read `capacity` from `resourceLimits.capacity` in the enroll params (default 1). Persist to the worker model.

Dispatch: Push `dispatchId` onto `activeDispatchIds` at dispatch start. Remove at dispatch end.

`onWorkerIdle`: Fire on any slot freeing (when `activeDispatchIds.length` decreases), not just when the worker becomes fully idle. This wakes queued steps so they can fill the freed slot.

Dispatch `finally` block: Currently skips the idle transition when `status === "draining"`. With slots, also skip when other dispatches are still active (no need to persist "idle" if the worker is still busy with other slots).

5. Worker model schema

File: `src/domain/models/worker/worker_model.ts`

Add `capacity: number` and `activeDispatchIds: string[]` to the worker state schema. These supersede the single `currentDispatchId`. Version bump.

Keep `currentDispatchId` as optional for read-time migration (same pattern as the enrollment token's `boundMachineId`).

6. Scheduler: slot-aware eligibility + least-loaded tiebreak

File: `src/domain/remote/scheduler.ts`

`SchedulableWorker`: Add `capacity: number` and `activeCount: number`.

`eligibleWorkers`: Unchanged — eligibility is about labels/platform/target/status, not load.

`scheduleStep`: Replace: ```typescript const idle = eligible.filter(w => w.status === "idle") .sort((a, b) => a.name.localeCompare(b.name)); ```

With: ```typescript const schedulable = eligible.filter(w => w.activeCount < w.capacity) .sort((a, b) => { const freeA = a.capacity - a.activeCount; const freeB = b.capacity - b.activeCount; if (freeA !== freeB) return freeB - freeA; // most free slots first return a.name.localeCompare(b.name); // deterministic tiebreak }); ```

A worker with 6/8 slots free beats one with 2/8 slots free. Equal free slots breaks on name for determinism.

7. Dispatch service: per-worker reservation count

File: `src/serve/dispatch_service.ts`

Replace: ```typescript readonly #reserved = new Set(); ```

With: ```typescript readonly #reserved = new Map<string, number>(); // workerName → reservation count ```

`#poolSnapshot`: Overlay reservation counts onto the snapshot's `activeCount`: ```typescript const reservations = this.#reserved.get(worker.name) ?? 0; return { ...worker, activeCount: worker.activeCount + reservations }; ```

On acquire: Increment the reservation count, not add to a set. On dispatch/release: Decrement.

8. Idle-timeout with slots

File: `src/worker/connect.ts`

Idle-timeout (phase 3) currently means "no dispatch in progress." With slots, it means "zero active runners." The idle timer should:

  • Start when `activeRunners` drops to 0
  • Reset when any dispatch starts (not just the first)
  • Not fire while any runner is active

9. Presentation

File: `src/presentation/output/worker_output.ts`

Update `swamp worker list` to show capacity and load:

  • Add `CAPACITY` and `LOAD` columns (e.g., `8` and `3/8`)
  • Or show as `SLOTS: 3/8`

10. Protocol version bump

Bump `REMOTE_PROTOCOL_VERSION` from 3 to 4. The `resourceLimits.capacity` field is now semantically load-bearing.

Invariants to preserve

  1. Capacity 1 is byte-for-byte today's behavior. Default `--concurrency 1` produces the same scheduling, same dispatch, same `worker_busy` rejection as before.
  2. `worker_busy` means "all slots full." Same error code, same re-queue behavior on the orchestrator.
  3. Each runner gets its own environment. The dispatch runner extraction (4a) already guarantees this — each child gets its spawn env. No change needed.
  4. Crash isolation per runner. One runner crashing doesn't affect siblings. Already guaranteed by 4a.
  5. Drain waits for ALL runners. `drain()` resolves when `activeRunners === 0`, not when one finishes.
  6. Sole-writer rule. Worker model transitions (`set_status`, `activeDispatchIds` updates) flow through `#recordTransition`.

Tests required

Scheduler tests:

  • Least-loaded tiebreak (most free slots wins)
  • Equal free slots breaks on name
  • Capacity 1 behavior unchanged
  • Worker at capacity is not schedulable

Dispatch handler tests:

  • Concurrent dispatches up to capacity succeed
  • Dispatch at capacity+1 rejected with `worker_busy`
  • Drain waits for all active runners to finish

Gateway tests:

  • `activeDispatchIds` tracks concurrent dispatches
  • `onWorkerIdle` fires on each slot freed
  • Capacity persisted from enrollment `resourceLimits`

Dispatch service tests:

  • Per-worker reservation count overlays correctly
  • Two queued steps don't both reserve the last slot

Integration test:

  • `forEach` fan-out onto one capacity-4 worker — assert temporal overlap (each step records start/end time, assert ranges overlap)
  • Conflicting environment snapshots isolated per dispatch (two concurrent steps with different shipped env vars each see only their own)

Acceptance criteria

  1. A capacity-8 worker runs 8 steps concurrently with correct per-step environments
  2. Killing one runner fails only that step; sibling runners complete normally
  3. Capacity-1 behavior is identical to pre-4b (serial, `worker_busy`)
  4. `swamp worker list` shows capacity and current load
  5. All existing tests pass
  6. `deno check`, `deno lint`, `deno fmt`, `deno run test`, `deno run compile` all pass

Repo conventions

  • TypeScript strict, no `any`, named exports only
  • Unit tests next to source
  • Bump `REMOTE_PROTOCOL_VERSION` (3→4)
  • Use the `github-pr` skill for commits/PRs
02Bog Flow
✓OPEN✓TRIAGED✓IN PROGRESS✓SHIPPED+ 1 MOREASSIGNED+ 8 MOREREVIEW+ 4 MOREPR_MERGED+ 1 MORENOTIFICATION_SKIPPED

Shipped

7/6/2026, 8:05:50 PM

Click a lifecycle step above to view its details.

03Sludge Pulse
stack72 assigned stack727/6/2026, 3:27:52 PM

Sign in to post a ripple.