From 4ec3b1b4bda0648d945b1a8f48c3e8b732ce434b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 21 Aug 2026 18:41:50 -0700 Subject: [PATCH 1/3] Add tests for bounded-parallel boot restore and parked-record partitioning Covers the pieces the CL-6282 boot-restore cutover needs: a generic bounded-concurrency runner with per-item failure isolation, and a partition of a boot scan into live vs. hibernate-parked records. --- apps/sidecar/src/concurrency.test.ts | 52 +++++++++++++++++++ .../src/workflow-deployment-record.test.ts | 28 ++++++++++ 2 files changed, 80 insertions(+) create mode 100644 apps/sidecar/src/concurrency.test.ts diff --git a/apps/sidecar/src/concurrency.test.ts b/apps/sidecar/src/concurrency.test.ts new file mode 100644 index 000000000..bdb105328 --- /dev/null +++ b/apps/sidecar/src/concurrency.test.ts @@ -0,0 +1,52 @@ +import { describe, test, expect } from "bun:test"; + +import { runWithConcurrency } from "./concurrency"; + +describe("runWithConcurrency", () => { + test("runs every item and never exceeds the concurrency limit", async () => { + let inFlight = 0; + let maxInFlight = 0; + const seen: number[] = []; + + await runWithConcurrency([1, 2, 3, 4, 5, 6, 7], 3, async (item) => { + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + await new Promise((resolve) => setTimeout(resolve, 5)); + seen.push(item); + inFlight -= 1; + }); + + expect(seen.sort((a, b) => a - b)).toEqual([1, 2, 3, 4, 5, 6, 7]); + expect(maxInFlight).toBeLessThanOrEqual(3); + }); + + test("isolates a failing item: the rest still complete and the failure is reported per item", async () => { + const completed: number[] = []; + const failures = await runWithConcurrency([1, 2, 3], 2, async (item) => { + if (item === 2) throw new Error("boom"); + completed.push(item); + }); + + expect(completed.sort()).toEqual([1, 3]); + expect(failures).toHaveLength(1); + expect(failures[0]?.item).toBe(2); + expect((failures[0]?.error as Error).message).toBe("boom"); + }); + + test("an empty list resolves immediately with no failures", async () => { + const failures = await runWithConcurrency( + [] as number[], + 4, + async () => {}, + ); + expect(failures).toEqual([]); + }); + + test("a limit larger than the item count still runs everything exactly once", async () => { + const seen: number[] = []; + await runWithConcurrency([1, 2], 10, async (item) => { + seen.push(item); + }); + expect(seen.sort()).toEqual([1, 2]); + }); +}); diff --git a/apps/sidecar/src/workflow-deployment-record.test.ts b/apps/sidecar/src/workflow-deployment-record.test.ts index a7e0deb64..571a12f97 100644 --- a/apps/sidecar/src/workflow-deployment-record.test.ts +++ b/apps/sidecar/src/workflow-deployment-record.test.ts @@ -5,6 +5,7 @@ import path from "node:path"; import { markWorkflowDeploymentRecordParked, + partitionScannedDeployments, readWorkflowDeploymentRecord, scanWorkflowDeploymentRecords, writeWorkflowDeploymentRecord, @@ -143,3 +144,30 @@ describe("scanWorkflowDeploymentRecords reaps pre-cutover records", () => { await fs.access(filePath); }); }); + +describe("partitionScannedDeployments", () => { + test("splits scanned records into live (no parkedAt) and parked (parkedAt set)", async () => { + const dataDir = await makeDataDir(); + await writeWorkflowDeploymentRecord(dataDir, "dep_live", baseRecord); + await writeWorkflowDeploymentRecord(dataDir, "dep_parked", baseRecord); + await markWorkflowDeploymentRecordParked(dataDir, "dep_parked"); + const scanned = await scanWorkflowDeploymentRecords(dataDir); + + const { live, parked } = partitionScannedDeployments(scanned); + + expect(live.map((s) => s.deploymentId)).toEqual(["dep_live"]); + expect(parked.map((s) => s.deploymentId)).toEqual(["dep_parked"]); + }); + + test("an all-live scan yields nothing to skip", () => { + const scanned = [ + { deploymentId: "dep_a", record: baseRecord }, + { deploymentId: "dep_b", record: baseRecord }, + ]; + + const { live, parked } = partitionScannedDeployments(scanned); + + expect(live).toHaveLength(2); + expect(parked).toHaveLength(0); + }); +}); From c83f3821da129c069a103948ac270bcf1bf92777 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 21 Aug 2026 18:42:01 -0700 Subject: [PATCH 2/3] Sidecar boot restore: skip parked deployments, restore live ones bounded-parallel The boot scan used to restore every persisted deployment record, serially, regardless of whether the hub had put it to sleep on purpose. On a workbench with any real usage this meant restoring dozens of hibernated deployments before the sidecar could even connect back to the hub -- work with no payoff, since a hibernated deployment resumes correctly on the next message or routine fire that addresses it (the same wake path that already handles a cold deployment). `parkedAt` (stamped by the hibernate teardown) is a durable, local signal a boot scan can already read without any hub round-trip, so this cuts over to the CL-6282 design the code was already flagging: skip parked records entirely, and restore only the live ones, bounded-parallel (cap of 8) instead of one at a time, with the same per-record failure isolation the serial loop had. Also prunes a deployment record whose stored address derives a different slug than its own directory -- that mismatch is deterministic and permanent, so leaving the record on disk only means warning about it forever. --- apps/sidecar/src/concurrency.ts | 38 +++++++++ .../sidecar/src/workflow-deployment-record.ts | 21 +++++ .../sidecar/src/workflow-host-wiring/index.ts | 81 +++++++++++-------- 3 files changed, 108 insertions(+), 32 deletions(-) create mode 100644 apps/sidecar/src/concurrency.ts diff --git a/apps/sidecar/src/concurrency.ts b/apps/sidecar/src/concurrency.ts new file mode 100644 index 000000000..88b5ddc79 --- /dev/null +++ b/apps/sidecar/src/concurrency.ts @@ -0,0 +1,38 @@ +export interface ConcurrencyFailure { + item: T; + error: unknown; +} + +/** + * Runs `fn` over every item with at most `limit` in flight at once. Each + * item's failure is caught and returned rather than thrown, so one bad + * item never stops the rest of the batch from running -- the same + * per-item isolation a serial `for` loop with a `try`/`catch` gives, just + * bounded-parallel instead of one-at-a-time. + */ +export async function runWithConcurrency( + items: readonly T[], + limit: number, + fn: (item: T) => Promise, +): Promise[]> { + const failures: ConcurrencyFailure[] = []; + let nextIndex = 0; + + async function worker(): Promise { + while (nextIndex < items.length) { + const index = nextIndex; + nextIndex += 1; + const item = items[index] as T; + try { + await fn(item); + } catch (error) { + failures.push({ item, error }); + } + } + } + + const workerCount = Math.max(1, Math.min(limit, items.length)); + await Promise.all(Array.from({ length: workerCount }, () => worker())); + + return failures; +} diff --git a/apps/sidecar/src/workflow-deployment-record.ts b/apps/sidecar/src/workflow-deployment-record.ts index 50f3b65b5..190aa2227 100644 --- a/apps/sidecar/src/workflow-deployment-record.ts +++ b/apps/sidecar/src/workflow-deployment-record.ts @@ -188,6 +188,27 @@ export interface ScannedWorkflowDeployment { record: WorkflowDeploymentRecord; } +/** + * Splits a boot scan into what still needs restoring and what a prior + * hibernate already put to sleep. `parkedAt` is the durable, locally + * decidable signal for "genuinely live" (CL-6282): a deployment the hub + * deliberately hibernated resumes on the next message or routine fire + * that addresses it, not by being pre-spawned at boot. + */ +export function partitionScannedDeployments( + scanned: readonly ScannedWorkflowDeployment[], +): { + live: ScannedWorkflowDeployment[]; + parked: ScannedWorkflowDeployment[]; +} { + const live: ScannedWorkflowDeployment[] = []; + const parked: ScannedWorkflowDeployment[] = []; + for (const entry of scanned) { + (entry.record.parkedAt === undefined ? live : parked).push(entry); + } + return { live, parked }; +} + /** * Enumerate the persisted deployment records under `workflow-runs/` so a * boot-time restore can re-establish each deployment. Soft-fails per record: diff --git a/apps/sidecar/src/workflow-host-wiring/index.ts b/apps/sidecar/src/workflow-host-wiring/index.ts index 8fff6fcc5..5c0025ab4 100644 --- a/apps/sidecar/src/workflow-host-wiring/index.ts +++ b/apps/sidecar/src/workflow-host-wiring/index.ts @@ -53,10 +53,12 @@ import type { import { deleteWorkflowDeploymentRecord, markWorkflowDeploymentRecordParked, + partitionScannedDeployments, scanWorkflowDeploymentRecords, writeWorkflowDeploymentRecord, type WorkflowDeploymentRecord, } from "../workflow-deployment-record"; +import { runWithConcurrency } from "../concurrency"; import { computeWireDefinitionHash, validateWorkflowProjection, @@ -128,6 +130,16 @@ export const CHILD_KILL_ESCALATION_MS = 3000; */ export const TEARDOWN_DRAIN_DEADLINE_MS = 5000; +/** + * How many workflow deployment records `restoreWorkflowDeployments` restores + * at once. Bounded, not unbounded, so a host with many live records at boot + * cannot storm the OS with concurrent `Bun.spawn` calls all at once; 8 is + * comfortably below typical per-process fd/thread pressure from a handful of + * child processes while still cutting a boot with dozens of records from + * minutes of serial restore to a few bounded rounds. + */ +export const RESTORE_CONCURRENCY = 8; + /** * Await a supervisor's graceful `shutdown()`, escalating to a direct * SIGKILL of its child if `shutdown()` hasn't settled within @@ -1389,11 +1401,14 @@ export function createSidecarDeployRouter(deps: { record: WorkflowDeploymentRecord, ): Promise { // Integrity: the stored address must re-derive to its own directory - // name. A mismatch means a corrupt or misplaced record; skip it - // rather than restore a deployment under the wrong slug. + // name. A mismatch is deterministic and permanent -- the same address + // always derives the same slug, so a record that disagrees with its + // own directory will disagree on every future boot too. Prune it now + // rather than restore under the wrong slug and warn forever. const derived = deriveDeploymentId(record.agentAddress); if (derived !== deploymentId) { - logger.warn`skipping workflow deployment restore: ${record.agentAddress} derives slug ${derived}, not its directory ${deploymentId}`; + await deleteWorkflowDeploymentRecord(dataDir, deploymentId); + logger.warn`pruned workflow deployment record ${deploymentId}: ${record.agentAddress} derives slug ${derived}, not its directory`; return; } @@ -1625,36 +1640,38 @@ export function createSidecarDeployRouter(deps: { } const scanned = await scanWorkflowDeploymentRecords(dataDir); - // Report parked-vs-live counts from the `parkedAt` marker - // (`markWorkflowDeploymentRecordParked`, written on the hibernate - // teardown) before restoring anything. This is REPORT-ONLY: every - // record below still restores LIVE regardless of this count, exactly - // as before this marker existed. Making the boot scan actually skip - // (or defer) the parked ones is CL-6282 -- a separate change, once its - // design pass lands -- and needs a signal this scan cannot see on its - // own before that cutover is safe: whether the hub still wants a - // parked deployment running is a hub-side decision, not something a - // sidecar with no hub connection yet can determine at boot. - const parkedCount = scanned.filter( - ({ record }) => record.parkedAt !== undefined, - ).length; - logger.info`Boot scan found ${scanned.length} deployment record(s): ${parkedCount} parked, ${scanned.length - parkedCount} live; restoring all of them (parked-aware boot restore is CL-6282)`; - // Restore serially, not in parallel: deterministic boot-log ordering, - // one isolable warning per failed record, and no concurrent - // child-spawn / transport-register storm. Restore runs before - // `hubLink.connect()`, so there are no concurrent deploys to contend - // with. Each record's failure is caught so one bad deployment cannot - // strand the rest. Every persisted record restores LIVE: a deployment - // that was previously torn down as a state-preserving hibernate is - // relaunched by the hub's own reap-and-relaunch flow, not by this - // boot scan guessing at staleness. - for (const { deploymentId, record } of scanned) { - try { + // CL-6282: `parkedAt` (written by `markWorkflowDeploymentRecordParked` + // on a state-preserving hibernate teardown) is the durable, locally + // decidable answer to "is this deployment genuinely live" -- a + // deployment the hub deliberately put to sleep does not need to be + // running to be correct; it resumes the moment a message or routine + // fire addresses it again (the same wake path `ensureAwake`/`deployAtHead` + // already use for an idle-slept deployment that never crashed at all). + // Restoring it anyway at every boot is exactly the "restart dead + // stuff" cost this cutover removes. A record with no `parkedAt` + // (never hibernated, or predates the field) restores as before -- + // absent is "assume live", matching `readWorkflowDeploymentRecord`'s + // own backward-compatible read. + const { live, parked } = partitionScannedDeployments(scanned); + logger.info`Boot scan found ${scanned.length} deployment record(s): ${parked.length} parked (left asleep; will wake on the next message or routine fire), ${live.length} to restore`; + // Bounded-parallel, not fully parallel: `RESTORE_CONCURRENCY` caps how + // many workflow-process children spawn at once so a boot with many + // live deployments cannot storm the host, while still restoring far + // faster than one-at-a-time. Restore runs before `hubLink.connect()`, + // so there are no concurrent hub-driven deploys to contend with. + // `runWithConcurrency` isolates each record's failure the same way the + // old serial loop's per-iteration `try`/`catch` did -- one bad record + // logs a warning and never strands the rest. + const failures = await runWithConcurrency( + live, + RESTORE_CONCURRENCY, + async ({ deploymentId, record }) => { await restoreDeploymentFromRecord(dataDir, deploymentId, record); - } catch (cause) { - const reason = cause instanceof Error ? cause.message : String(cause); - logger.warn`Failed to restore workflow deployment ${deploymentId}: ${reason}`; - } + }, + ); + for (const { item, error } of failures) { + const reason = error instanceof Error ? error.message : String(error); + logger.warn`Failed to restore workflow deployment ${item.deploymentId}: ${reason}`; } }, activeAddresses(): string[] { From b8c0e1d65aa332222768f1209660a175e5ce2908 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Fri, 21 Aug 2026 23:04:34 -0700 Subject: [PATCH 3/3] Fix prettier formatting from merge conflict resolution --- apps/sidecar/src/workflow-host-wiring/index.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/sidecar/src/workflow-host-wiring/index.ts b/apps/sidecar/src/workflow-host-wiring/index.ts index bc454328e..766e8f768 100644 --- a/apps/sidecar/src/workflow-host-wiring/index.ts +++ b/apps/sidecar/src/workflow-host-wiring/index.ts @@ -1836,7 +1836,9 @@ export function createSidecarDeployRouter(deps: { const reason = cause instanceof Error ? cause.message : String(cause); const kind = - cause instanceof WorkflowRestoreFailure ? cause.kind : "transient"; + cause instanceof WorkflowRestoreFailure + ? cause.kind + : "transient"; const updated = await recordWorkflowDeploymentRestoreFailure( dataDir, deploymentId,