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

Relationships

#952 Worker fleets phase 1 PR3: boot reconciliation and integration test

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

Summary

PR3 of worker-fleets phase 1 (#947). PR1 landed core queueing (#1746, #1747), PR2 landed queue visibility (#949). This PR adds resilience: boot reconciliation so bookkeeping data never lies after an orchestrator crash, and the golden-path integration test proving the full queue→dispatch flow end-to-end.

Depends on PR1 and PR2, both shipped.

Design reference

Full design: worker-fleets.md at the repo root (sections "Failure and recovery" → "Boot reconciliation" and the Phase 1 acceptance criteria). This issue is self-contained for implementation.

Changes

1. Boot reconciliation sweep at serve startup

File: src/cli/commands/serve.ts (in the remote-execution wiring block, ~lines 572–607, before Deno.serve accepts traffic)

At serve startup, sweep three bookkeeping models for records that are stale from a prior crash:

Step-leases (swamp/step-lease):

  • Query: modelType == "swamp/step-lease" && attributes.state == "active"
  • Transition each to expired with inputs { leaseId, error: "orchestrator restart" }
  • Use the step-lease model's expire method

Pending-dispatches (swamp/pending-dispatch):

  • Query: modelType == "swamp/pending-dispatch" && attributes.state == "waiting"
  • Transition each to orphaned with inputs { queueId, endedAt: new Date().toISOString() }
  • Use the pending-dispatch model's orphan method

Workers (swamp/worker):

  • Query: modelType == "swamp/worker" && attributes.status != "disconnected"
  • Transition each to disconnected via the worker model's set_status method with inputs { status: "disconnected" }

Implementation pattern:

  • Query with the datastore-read pattern from src/libswamp/worker/list.ts (use DataQueryService.query with loadAttributes: true)
  • Transition through createWorkerModelRunDeps with skipAllReports: true
  • Log a one-line summary: "Boot reconciliation: swept N leases, M pending dispatches, K workers"
  • If no records need sweeping, log nothing (don't clutter a clean boot)

Timing: Must complete before Deno.serve starts accepting traffic. Workers that reconnect during the sweep will re-enroll normally after it finishes — the sweep transitions their stale record, and re-enrollment creates a fresh one.

Error handling: Individual transition failures should warn but not abort startup. A single bad record shouldn't prevent the orchestrator from serving.

2. Golden-path integration test: queue → enroll → dispatch → complete

File: integration/remote_execution_test.ts

Add a new test that exercises the full elastic queueing flow:

Deno.test({
  name: "remote execution: elastic queueing — empty pool queues, worker enrolls, step dispatches",
  sanitizeOps: false,
  sanitizeResources: false,
  fn: async () => {
    await withTempDir(async (dir) => {
      const orchestrator = await startOrchestrator(dir);
      const workerStop = new AbortController();
      let workerDone: Promise<void> | null = null;
      try {
        const token = await orchestrator.mintToken("queue-worker");

        // Start the dispatch BEFORE any worker connects — it should queue
        const pending = orchestrator.dispatchService.executeRemote(
          remoteStepRequest({
            placement: { labels: { tier: "it" }, queueTimeoutMs: 10_000 },
          }),
        );

        // Give it time to enter the queue
        await new Promise((r) => setTimeout(r, 200));

        // Verify: no dispatch happened yet, step is queued
        assertEquals(orchestrator.gateway.workers().length, 0);

        // Connect a worker — step should dispatch immediately (wake-on-enroll)
        workerDone = runWorker({
          url: orchestrator.serverUrl,
          token,
          labels: { tier: "it" },
          swampVersion: "test",
          cacheDir: join(dir, "worker-cache"),
          signal: workerStop.signal,
        });

        // The pending dispatch should complete
        const result = await pending;
        assertEquals(result.outputs.length, 2);

        // Verify the worker is now idle (dispatch finished)
        await waitFor(
          () => orchestrator.gateway.worker("queue-worker")?.status === "idle",
          "worker to return to idle",
        );
      } finally {
        workerStop.abort();
        if (workerDone) await workerDone.catch(() => {});
        await orchestrator.shutdown();
      }
    });
  },
});

3. Boot reconciliation integration test

File: integration/remote_execution_test.ts

Add a test that seeds stale records and verifies boot reconciliation:

Deno.test({
  name: "remote execution: boot reconciliation marks stale records",
  sanitizeOps: false,
  sanitizeResources: false,
  fn: async () => {
    await withTempDir(async (dir) => {
      // First boot: create some active state
      let orchestrator = await startOrchestrator(dir);
      const token = await orchestrator.mintToken("recon-worker");
      // ... connect worker, start a dispatch, then hard-kill without cleanup ...
      await orchestrator.shutdown(); // simulates crash — leaves stale records

      // Second boot: reconciliation should sweep
      orchestrator = await startOrchestrator(dir);
      try {
        // Query the bookkeeping data — stale records should be swept
        const leases = await orchestrator.repoContext.dataQueryService.query(
          'modelType == "swamp/step-lease" && attributes.state == "active"',
          { loadAttributes: true },
        );
        assertEquals(leases.length, 0); // all swept to expired

        const pending = await orchestrator.repoContext.dataQueryService.query(
          'modelType == "swamp/pending-dispatch" && attributes.state == "waiting"',
          { loadAttributes: true },
        );
        assertEquals(pending.length, 0); // all swept to orphaned
      } finally {
        await orchestrator.shutdown();
      }
    });
  },
});

The exact setup will depend on how to create stale records — either by running a dispatch and force-killing without cleanup, or by directly writing stale records via the model method runner before the second boot.

4. Update design docs

File: worker-fleets.md

  • Mark phase 1 as implemented (all three PRs shipped)
  • Note any deviations from the plan

File: design/remote-execution.md

  • Add boot reconciliation to the "Worker state is swamp data" section (it sweeps the three bookkeeping models on startup)

Invariants to preserve

  1. Boot reconciliation must complete before accepting traffic. Workers reconnecting during the sweep must wait for it to finish — their re-enrollment will succeed after the sweep transitions their stale record.
  2. Sole-writer rule. The sweep runs transitions through the model-method runner (serialized through the transition tail), not by direct datastore writes.
  3. Individual failures don't abort startup. A corrupted record in one model shouldn't prevent the orchestrator from starting.
  4. The sweep is idempotent. Running it against an already-clean datastore is a no-op (no records match the stale-state predicates).

Tests required

  • Boot reconciliation unit test: seed stale records (active lease, waiting pending-dispatch, connected-looking worker), boot, assert all transitioned to terminal states
  • Clean boot test: boot with no stale records, assert no errors and no transitions fired
  • Integration test: golden-path queue→enroll→dispatch flow (the most important test — proves the whole system works end-to-end)
  • Integration test: boot reconciliation with a second startOrchestrator call against the same repo dir

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"
  • Cross-component changes need integration tests under integration/

Acceptance criteria

  1. swamp serve restart marks stale leases as expired, stale pending-dispatches as orphaned, and stale workers as disconnected
  2. A one-line summary appears in logs only when records were swept
  3. The golden-path integration test passes: empty pool → step queues → worker enrolls → step dispatches → result returned
  4. All existing tests continue to pass
  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
  • Use the github-pr skill for commits/PRs
  • Keep blast radius small — no adjacent refactors
  • Update worker-fleets.md when implementation diverges from the plan
02Bog Flow
✓OPEN✓TRIAGED✓IN PROGRESS✓SHIPPED+ 1 MOREASSIGNED+ 3 MOREFINDINGS+ 4 MOREPR_MERGED+ 1 MORENOTIFICATION_SKIPPED

Shipped

7/4/2026, 1:24:46 PM

Click a lifecycle step above to view its details.

03Sludge Pulse
stack72 assigned stack727/4/2026, 12:11:29 PM

Sign in to post a ripple.