Sidecar boot restore: skip parked deployments, restore live ones bounded-parallel

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]);
+  });
+});

apps/sidecar/src/concurrency.ts

@@ -0,0 +1,38 @@
+export interface ConcurrencyFailure<T> {
+  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<T>(
+  items: readonly T[],
+  limit: number,
+  fn: (item: T) => Promise<void>,
+): Promise<ConcurrencyFailure<T>[]> {
+  const failures: ConcurrencyFailure<T>[] = [];
+  let nextIndex = 0;
+
+  async function worker(): Promise<void> {
+    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;
+}

apps/sidecar/src/workflow-deployment-record.test.ts

@@ -8,6 +8,7 @@ import {
   clearWorkflowDeploymentRestoreFailure,
   isWorkflowDeploymentRestoreQuarantined,
   markWorkflowDeploymentRecordParked,
+  partitionScannedDeployments,
   readWorkflowDeploymentRecord,
   recordWorkflowDeploymentRestoreFailure,
   scanWorkflowDeploymentRecords,
@@ -148,6 +149,33 @@ describe("scanWorkflowDeploymentRecords reaps pre-cutover records", () => {
   });
 });
 
+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);
+  });
+});
+
 describe("recordWorkflowDeploymentRestoreFailure", () => {
   test("starts a kind's counter at 1 and persists reason/timestamp", async () => {
     const dataDir = await makeDataDir();

apps/sidecar/src/workflow-deployment-record.ts

@@ -306,6 +306,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:

apps/sidecar/src/workflow-host-wiring/index.ts

@@ -57,11 +57,13 @@ import {
   deleteWorkflowDeploymentRecord,
   isWorkflowDeploymentRestoreQuarantined,
   markWorkflowDeploymentRecordParked,
+  partitionScannedDeployments,
   recordWorkflowDeploymentRestoreFailure,
   scanWorkflowDeploymentRecords,
   writeWorkflowDeploymentRecord,
   type WorkflowDeploymentRecord,
 } from "../workflow-deployment-record";
+import { runWithConcurrency } from "../concurrency";
 import {
   reapExpiredHibernationSnapshots as reapVaultSnapshots,
   restoreAgentIdentity,
@@ -138,6 +140,54 @@ 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;
+
+/**
+ * Ceiling on a single `restoreDeploymentFromRecord` attempt. A restore
+ * candidate can wedge indefinitely -- a stalled closure fetch, a spawn that
+ * never reports back -- and with a bounded worker pool one wedged record
+ * pins its worker (and, at the extreme, every worker) for the rest of
+ * boot. A restore that exceeds this deadline is treated as an ordinary
+ * transient failure: counted on the record's `restoreFailure` counter,
+ * logged, and skipped, so the boot moves on and the record gets another
+ * attempt next boot (or quarantines after `RESTORE_QUARANTINE_THRESHOLD`
+ * consecutive permanent failures).
+ */
+export const RESTORE_ATTEMPT_TIMEOUT_MS = 30_000;
+
+function withRestoreTimeout(
+  promise: Promise<void>,
+  deploymentId: string,
+): Promise<void> {
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(() => {
+      reject(
+        new Error(
+          `restore of ${deploymentId} exceeded ${RESTORE_ATTEMPT_TIMEOUT_MS}ms`,
+        ),
+      );
+    }, RESTORE_ATTEMPT_TIMEOUT_MS);
+    promise.then(
+      (value) => {
+        clearTimeout(timer);
+        resolve(value);
+      },
+      (error) => {
+        clearTimeout(timer);
+        reject(error);
+      },
+    );
+  });
+}
+
 /**
  * Await a supervisor's graceful `shutdown()`, escalating to a direct
  * SIGKILL of its child if `shutdown()` hasn't settled within
@@ -1727,74 +1777,83 @@ 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
-      // non-quarantined 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;
-      const alreadyQuarantinedCount = scanned.filter(({ record }) =>
+      // 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);
+      const alreadyQuarantinedCount = live.filter(({ record }) =>
         isWorkflowDeploymentRestoreQuarantined(record),
       ).length;
-      logger.info`Boot scan found ${scanned.length} deployment record(s): ${parkedCount} parked, ${alreadyQuarantinedCount} quarantined, ${scanned.length - parkedCount - alreadyQuarantinedCount} to restore (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 non-quarantined 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.
+      logger.info`Boot scan found ${scanned.length} deployment record(s): ${parked.length} parked (left asleep; will wake on the next message or routine fire), ${alreadyQuarantinedCount} quarantined, ${live.length - alreadyQuarantinedCount} 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. Each
+      // worker handles its own quarantine skip/record/clear bookkeeping and
+      // logging inline (rather than via `runWithConcurrency`'s returned
+      // failure list) so one record's outcome never depends on another's
+      // ordering, matching the isolation the old serial loop's per-iteration
+      // `try`/`catch` gave.
       let skippedQuarantinedCount = 0;
-      for (const { deploymentId, record } of scanned) {
-        // Skip WITHOUT attempting: a permanently unrestorable record that
-        // has already crossed RESTORE_QUARANTINE_THRESHOLD gets neither a
-        // spawn attempt nor a per-record warning this boot -- both are
-        // pointless for a deterministic failure that has already been
-        // reported that many times. It still counts toward the one
-        // summary line below, and the record itself is untouched (an
-        // operator reclaims it by undeploying the address).
-        if (isWorkflowDeploymentRestoreQuarantined(record)) {
-          skippedQuarantinedCount += 1;
-          continue;
-        }
-        try {
-          await restoreDeploymentFromRecord(dataDir, deploymentId, record);
-          if (record.restoreFailure !== undefined) {
-            await clearWorkflowDeploymentRestoreFailure(
+      await runWithConcurrency(
+        live,
+        RESTORE_CONCURRENCY,
+        async ({ deploymentId, record }) => {
+          // Skip WITHOUT attempting: a permanently unrestorable record that
+          // has already crossed RESTORE_QUARANTINE_THRESHOLD gets neither a
+          // spawn attempt nor a per-record warning this boot -- both are
+          // pointless for a deterministic failure that has already been
+          // reported that many times. It still counts toward the one
+          // summary line below, and the record itself is untouched (an
+          // operator reclaims it by undeploying the address).
+          if (isWorkflowDeploymentRestoreQuarantined(record)) {
+            skippedQuarantinedCount += 1;
+            return;
+          }
+          try {
+            await withRestoreTimeout(
+              restoreDeploymentFromRecord(dataDir, deploymentId, record),
+              deploymentId,
+            );
+            if (record.restoreFailure !== undefined) {
+              await clearWorkflowDeploymentRestoreFailure(
+                dataDir,
+                deploymentId,
+                record,
+              );
+            }
+          } catch (cause) {
+            const reason =
+              cause instanceof Error ? cause.message : String(cause);
+            const kind =
+              cause instanceof WorkflowRestoreFailure
+                ? cause.kind
+                : "transient";
+            const updated = await recordWorkflowDeploymentRestoreFailure(
               dataDir,
               deploymentId,
               record,
+              { kind, reason },
             );
+            if (isWorkflowDeploymentRestoreQuarantined(updated)) {
+              const attempts = updated.restoreFailure?.attempts ?? 0;
+              logger.warn`Workflow deployment ${deploymentId} failed to restore ${attempts} consecutive times and is now quarantined -- it will not be retried again until the address is undeployed. Last failure: ${reason}`;
+            } else {
+              logger.warn`Failed to restore workflow deployment ${deploymentId}: ${reason}`;
+            }
           }
-        } catch (cause) {
-          const reason = cause instanceof Error ? cause.message : String(cause);
-          const kind =
-            cause instanceof WorkflowRestoreFailure ? cause.kind : "transient";
-          const updated = await recordWorkflowDeploymentRestoreFailure(
-            dataDir,
-            deploymentId,
-            record,
-            { kind, reason },
-          );
-          if (isWorkflowDeploymentRestoreQuarantined(updated)) {
-            const attempts = updated.restoreFailure?.attempts ?? 0;
-            logger.warn`Workflow deployment ${deploymentId} failed to restore ${attempts} consecutive times and is now quarantined -- it will not be retried again until the address is undeployed. Last failure: ${reason}`;
-          } else {
-            logger.warn`Failed to restore workflow deployment ${deploymentId}: ${reason}`;
-          }
-        }
-      }
+        },
+      );
       if (skippedQuarantinedCount > 0) {
         logger.warn`Skipped ${skippedQuarantinedCount} quarantined workflow deployment record(s) (permanent restore failures, already reported); undeploy an address to clear its record`;
       }