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

Relationships

#976 Worker fleets phase 3 PR1: drain state machine, lifecycle policies, signal handling

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

Summary

Phase 3 of worker fleets. Today, stopping a worker (SIGTERM, container shutdown, ASG scale-in) kills it immediately — any in-flight step is lost. After this PR, workers drain: finish in-flight work, reject new dispatches, disconnect cleanly, and exit with a code that tells the platform "done on purpose" vs "crashed."

Three pieces:

  1. Drain state machine — worker-side flag + `worker.drain` orchestrator notification
  2. Lifecycle policies — `--max-dispatches N` and `--idle-timeout ` as drain triggers
  3. Signal handling + exit-code contract — SIGTERM/SIGINT trigger drain; exit 0 for policy-complete, non-zero for failures

Design reference: `worker-fleets.md` at the repo root, section "Worker lifecycle".

Changes

1. Worker-side drain state machine

File: `src/worker/dispatch_handler.ts`

Replace the boolean `busy` guard (line 148) with a richer state:

```typescript let busy = false; let draining = false; ```

When `draining` is true and a new dispatch arrives: ```typescript if (draining) { throw new RpcError({ code: "worker_draining", message: "Worker is draining — no new dispatches accepted", }); } ```

The `worker_draining` code is distinct from `worker_busy` so the orchestrator can distinguish "temporarily full" from "permanently shutting down."

Expose a `drain()` function from `registerDispatchHandler` that sets `draining = true` and returns a promise that resolves when any in-flight dispatch completes (or immediately if idle):

```typescript export function registerDispatchHandler(options): { drain: () => Promise; } { let busy = false; let draining = false; let onDrainComplete: (() => void) | null = null;

// ... register handler ...

return { drain: () => { draining = true; if (!busy) return Promise.resolve(); return new Promise((resolve) => { onDrainComplete = resolve; }); }, }; } ```

In the handler's `finally` block, after `busy = false`, check `if (draining && onDrainComplete) onDrainComplete()`.

2. `worker.drain` control message

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

Add a new worker → orchestrator message: ```typescript export const RemoteMethod = { // ... existing ... drain: "worker.drain", } as const; ```

No payload needed — the message itself is the signal.

File: `src/serve/worker_gateway.ts`

Handle the `worker.drain` message:

  1. Set the worker entry's status to `"draining"`
  2. Persist via `set_status({ status: "draining" })` through `#recordTransition`
  3. Treat the subsequent disconnect as deliberate: remove from pool immediately with no grace window, no re-dispatch churn
  4. Wake pool waiters so queued steps re-evaluate (a draining worker is no longer a candidate)

Add `"draining"` to `WorkerSnapshot.status` and `WorkerEntry.status` types (alongside `idle`, `busy`, `unverified`).

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

Add `"draining"` to `SchedulableWorker.status`. Exclude from `eligibleWorkers` — same position as `unverified` (after the target check, so a targeted dispatch to a draining worker still works for diagnostic purposes):

```typescript if (worker.status === "unverified" || worker.status === "draining") { return false; } ```

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

Add `"draining"` to `WorkerStatusSchema`.

3. `worker_draining` rejection handling in dispatch service

File: `src/serve/dispatch_service.ts`

In `executeRemote`'s inner catch (alongside the `worker_busy` handler): ```typescript if (error instanceof RpcError && error.code === "worker_draining") { logger.warn( "Worker {worker} is draining; re-queueing step", { worker: workerName }, ); endQueue("mark_dispatched", { dispatchId: error.dispatchId ?? "unknown" }); continue; } ```

Treat it exactly like `worker_busy` — re-queue the step, never fail it. This covers the race where a dispatch is in flight toward a worker that starts draining.

4. Signal handling

File: `src/worker/connect.ts`

Add signal listeners in the `connectOnce` function (or in `runWorker`):

  • SIGTERM (Unix) / SIGINT (all platforms) / SIGBREAK (Windows): trigger drain
  • Gate registration by platform: `Deno.addSignalListener("SIGTERM", ...)` only on non-Windows
  • A second signal force-exits with non-zero code

The drain flow on signal:

  1. Set a flag to prevent the reconnect loop from retrying
  2. Call the dispatch handler's `drain()` — waits for in-flight work to finish
  3. Send `worker.drain` to the orchestrator over the control channel
  4. Close the WebSocket cleanly
  5. Return from `runWorker` (which triggers exit)

File: `src/worker/connect.ts` — `RunWorkerOptions`

Add new options: ```typescript maxDispatches?: number; idleTimeoutMs?: number; ```

5. `--max-dispatches` and `--idle-timeout` flags

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

Add flags: ``` --max-dispatches <n:number> Drain and exit 0 after N dispatches complete --idle-timeout duration:string Drain and exit 0 after being continuously idle for this duration ```

Add env-var fallbacks (matching the phase 2 pattern): ``` SWAMP_WORKER_MAX_DISPATCHES → --max-dispatches SWAMP_WORKER_IDLE_TIMEOUT → --idle-timeout ```

Parse `--idle-timeout` CLI-side with `parseTimeout` (same as `--queue-timeout`).

Thread into `runWorker` options.

6. Lifecycle policy implementation in connect loop

File: `src/worker/connect.ts`

`--max-dispatches N`: After the Nth dispatch completes (counted in the dispatch handler), trigger drain. The handler's `drain()` returns immediately since the Nth dispatch just finished and no new one is in flight.

`--idle-timeout `: A timer that starts when the worker becomes idle (dispatch completes or initial enrollment with no immediate dispatch). Reset on every dispatch start. When it fires, trigger drain. The timer must be cancelled on drain from other triggers (signal, max-dispatches) to avoid double-drain.

7. Exit-code contract

File: `src/worker/connect.ts` → `src/cli/commands/worker_connect.ts`

`runWorker` currently returns `void`. Change it to return a result indicating why it stopped:

```typescript interface WorkerExitResult { reason: "drained" | "max-dispatches" | "idle-timeout" | "signal" | "shutdown" | "error"; } ```

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

Map the result to exit codes:

  • `drained`, `max-dispatches`, `idle-timeout`, `signal` (first signal) → exit 0 (policy-complete)
  • `error`, permanent enrollment failure → exit 1 (failure)

This is what makes k8s Jobs record success (`restartPolicy: Never` + exit 0), compose `restart: on-failure` not restart after a clean drain, and ASG health checks distinguish "done" from "broken."

8. Deliberate disconnect handling in gateway

File: `src/serve/worker_gateway.ts`

In `#handleSocketClosed`, check if the worker's status is `"draining"`:

  • If `draining`: skip the grace window entirely. Remove from pool immediately. Don't re-arm timers. Don't fire `onGraceExpired`.
  • If not draining: existing grace window behavior (unchanged).

This prevents a draining worker from occupying a pool slot during the grace window and avoids churn from re-dispatch attempts.

9. Presentation changes

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

Add color for `draining` status in worker list (e.g., yellow, matching `unverified`).

File: `src/presentation/output/worker_status_output.ts` (or wherever `renderWorkerStatus` lives)

Add status events for drain lifecycle: `draining`, `drain-complete`.

Invariants to preserve

  1. Sole-writer rule. The `worker.drain` → `set_status("draining")` transition must flow through `#recordTransition`.
  2. `worker_draining` is re-queue, not fail. Same semantics as `worker_busy` — the step is never failed due to a draining worker.
  3. Drain is not a grace-window event. A draining worker's disconnect is deliberate. No grace timer, no re-dispatch.
  4. Signal handling is platform-gated. SIGTERM only on non-Windows. SIGINT on all platforms. SIGBREAK on Windows only.
  5. Exit code 0 = policy-complete. Any trigger that the operator configured (signal, max-dispatches, idle-timeout) exits 0. Only unexpected failures exit non-zero.
  6. Protocol discipline. Adding `worker.drain` to `RemoteMethod` changes the worker→orchestrator protocol. Bump `REMOTE_PROTOCOL_VERSION`.
  7. Idle-timeout means truly idle. Zero active dispatches for the full duration. Reset on dispatch start, not dispatch end.

Tests required

Dispatch handler tests (`dispatch_handler_test.ts`):

  • `draining` rejects with `worker_draining`, not `worker_busy`
  • In-flight dispatch completes normally during drain
  • `drain()` resolves immediately when idle
  • `drain()` resolves after in-flight dispatch completes

Gateway tests (`worker_gateway_test.ts`):

  • `worker.drain` message sets status to `draining`
  • Draining worker's disconnect skips grace window
  • Pool waiters are woken on drain

Scheduler tests (`scheduler_test.ts`):

  • `draining` excluded from `eligibleWorkers` (non-targeted)
  • Targeted placement still reaches `draining` worker

Dispatch service tests (`dispatch_service_test.ts`):

  • `worker_draining` re-queues the step (same as `worker_busy`)

Connect/lifecycle tests:

  • `--max-dispatches 1`: one dispatch completes, worker drains, exits 0
  • `--idle-timeout`: timer fires after idleness, worker drains, exits 0
  • `--idle-timeout` resets on dispatch start
  • SIGTERM triggers drain, in-flight completes, exit 0
  • Second SIGTERM force-exits non-zero

Integration test (`integration/remote_execution_test.ts`):

  • Worker connects, receives dispatch, mid-dispatch signal → step completes → clean disconnect → orchestrator pool consistent

Acceptance criteria

  1. `--max-dispatches 1` runs exactly one step and exits 0
  2. SIGTERM during a step finishes the step, takes no new work, exits 0
  3. A draining worker never receives a dispatch (scheduler excludes it)
  4. `--idle-timeout 5s` fires only after true idleness (not after a dispatch completes with 4s remaining)
  5. `swamp worker list` shows `draining` status during the drain window
  6. Exit code 0 for all policy-complete shutdowns; non-zero for failures
  7. All existing tests pass
  8. `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
  • Unit tests live next to source: `foo.ts` → `foo_test.ts`
  • Signal handling tests must be platform-gated (`ignore: Deno.build.os === "windows"` for SIGTERM tests)
  • Use the `ddd` skill when shaping domain changes
  • Use the `github-pr` skill for commits/PRs
  • Bump `REMOTE_PROTOCOL_VERSION` for the new `worker.drain` message
02Bog Flow
✓OPEN✓TRIAGED✓IN PROGRESS✓SHIPPED+ 1 MOREASSIGNED+ 5 MOREREVIEW+ 4 MOREPR_MERGED+ 1 MORENOTIFICATION_SKIPPED

Shipped

7/5/2026, 7:14:01 PM

Click a lifecycle step above to view its details.

03Sludge Pulse
stack72 assigned stack727/5/2026, 3:52:49 PM

Sign in to post a ripple.