feat(webapp,run-engine): bind the append-failure hook to the existing repair job

A third module imports both sides, so the run store never has to import the
engine and the engine never has to know about the sweep's collaborators. It is
invoked from entry.server.tsx beside the other boot registrations, and returns
early when no snapshot-store Redis is configured.

enqueueSnapshotRepair shares the stall watchdog's job id and its availableAt.
Sharing the id is what stops the two compensators enqueueing two repairs for one
run; sharing the delay is what stops the winner of a race changing when the
repair runs. It now returns whether a job was actually inserted, which is what
makes "exactly one repair job" directly assertable rather than inferred.

The sweeper is given the undecorated router. Its rule 2 asks Postgres whether a
run row exists, and it must never be able to ask Redis whether Redis is an
orphan.

The sweep does not yet accept a deadline, so the budget is not enforced inside a
pass. The fenced lock is what keeps two passes apart, and its TTL covers the
expected duration.

Author: d-cs

apps/webapp/app/entry.server.tsx

@@ -17,6 +17,8 @@ import { LocaleContextProvider } from "./components/primitives/LocaleProvider";
 import type { OperatingSystemPlatform } from "./components/primitives/OperatingSystemProvider";
 import { OperatingSystemContextProvider } from "./components/primitives/OperatingSystemProvider";
 import { assertRunOpsSplitSentinel, Prisma } from "./db.server";
+import { assertSnapshotStoreBootFromEnv } from "./v3/snapshotStoreBoot.server";
+import { registerSnapshotStoreWiring } from "./v3/snapshotStoreWiring.server";
 import { env } from "./env.server";
 import { eventLoopMonitor, eventLoopUtilizationMonitor } from "./eventLoopMonitor.server";
 import { logger } from "./services/logger.server";
@@ -325,6 +327,18 @@ singleton("AssertRunOpsSplitSentinel", () => {
   return true;
 });
 
+singleton("SnapshotStoreWiring", registerSnapshotStoreWiring);
+
+// Ordered after the wiring above: the boot check asserts the repair binding is set, and the
+// binding is what the wiring installs.
+singleton("AssertSnapshotStoreBoot", () => {
+  assertSnapshotStoreBootFromEnv().catch((error) => {
+    logger.error("Snapshot store boot check failed; refusing to start", { error });
+    process.exit(1);
+  });
+  return true;
+});
+
 singleton("RunEngineEventBusHandlers", registerRunEngineEventBusHandlers);
 singleton("SetupBatchQueueCallbacks", setupBatchQueueCallbacks);
 // Attach the realtime run-changed publish delegations to the engine event bus.

apps/webapp/app/v3/runEngine.server.ts

@@ -12,6 +12,7 @@ import { runEnginePendingVersionLookup } from "./runEnginePendingVersionLookup.s
 import { pickRunOpsStoreForCompletion } from "./runOpsMigration/crossSeamGuard.server";
 import { runEngineControlPlaneResolver } from "./runOpsMigration/runEngineControlPlaneResolver.server";
 import { runStore } from "./runStore.server";
+import { getSnapshotSweepRunner } from "./snapshotStoreBindings.server";
 import { meter, tracer } from "./tracer.server";
 
 export const engine = singleton("RunEngine", createRunEngine);
@@ -241,6 +242,18 @@ function createRunEngine() {
         randomize: true,
       },
     },
+    snapshotStore: {
+      runSweep: async (opts) => {
+        const run = getSnapshotSweepRunner();
+        if (!run) {
+          return { outcome: "unbound" };
+        }
+        return run(opts);
+      },
+      sweepSchedule: env.RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_SCHEDULE,
+      sweepJitterInMs: env.RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_JITTER_IN_MS,
+      sweepBudgetMs: env.RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_BUDGET_MS,
+    },
     // Debounce configuration
     debounce: {
       maxDebounceDurationMs: env.RUN_ENGINE_MAXIMUM_DEBOUNCE_DURATION_MS,

apps/webapp/app/v3/snapshotStoreWiring.server.ts

@@ -0,0 +1,53 @@
+import { SnapshotOrphanSweeper } from "@internal/run-store";
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+import { engine } from "./runEngine.server";
+import { runStoreWithoutSnapshotDecorator } from "./runStore.server";
+import { buildSnapshotSweepRunner } from "./snapshotSweepRunner.server";
+import { setSnapshotRepairEnqueuer, setSnapshotSweepRunner } from "./snapshotStoreBindings.server";
+import { getSnapshotSweepClient } from "./snapshotStoreInstance.server";
+
+/**
+ * The third module: it imports both sides, so neither the run store nor the engine has to import
+ * the other. Invoked from entry.server.tsx, beside the other boot registrations.
+ */
+export function registerSnapshotStoreWiring(): boolean {
+  const sweepClient = getSnapshotSweepClient();
+
+  if (!sweepClient) {
+    return false;
+  }
+
+  setSnapshotRepairEnqueuer(async (args) => {
+    await engine.enqueueSnapshotRepair(args);
+  });
+
+  const sweeper = new SnapshotOrphanSweeper({
+    redisOptions: {
+      keyPrefix: "engine:",
+      host: env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST ?? undefined,
+      port: env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_PORT ?? undefined,
+      username: env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_USERNAME ?? undefined,
+      password: env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_PASSWORD ?? undefined,
+      ...(env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+    },
+    // The undecorated router: rule 2 asks Postgres whether a run row exists, and must never be
+    // able to ask Redis whether Redis is an orphan.
+    runStore: runStoreWithoutSnapshotDecorator,
+    completedTtlMs: env.RUN_ENGINE_SNAPSHOT_STORE_COMPLETED_TTL_MS,
+    orphanAgeMs: env.RUN_ENGINE_SNAPSHOT_STORE_ORPHAN_AGE_MS,
+  });
+
+  setSnapshotSweepRunner(
+    buildSnapshotSweepRunner({
+      client: sweepClient,
+      // The sweep does not yet accept a deadline, so the budget is not enforced inside a pass. The
+      // fenced lock is what keeps two passes apart; its TTL covers the expected pass duration.
+      sweep: async () => ({ ...(await sweeper.sweep()) }),
+      lockTtlMs: env.RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_BUDGET_MS + 3_600_000,
+    })
+  );
+
+  logger.info("snapshot store wiring registered");
+  return true;
+}

internal-packages/run-engine/src/engine/index.ts

@@ -2497,6 +2497,23 @@ export class RunEngine {
     };
   }
 
+  /**
+   * The append-failure compensator. Shares the stall watchdog's job id AND its availableAt, so the
+   * two cannot enqueue two repairs for one run and neither can win a race that changes the delay.
+   */
+  async enqueueSnapshotRepair(payload: {
+    runId: string;
+    snapshotId: string;
+    executionStatus: string;
+  }): Promise<boolean> {
+    return this.worker.enqueueOnce({
+      id: `repair-in-progress-run:${payload.runId}`,
+      job: "repairSnapshot",
+      payload,
+      availableAt: new Date(Date.now() + this.repairSnapshotTimeoutMs),
+    });
+  }
+
   async #repairRun(runId: string, dryRun: boolean) {
     const snapshot = await getLatestExecutionSnapshot(this.prisma, runId, this.runStore);
 

internal-packages/run-engine/src/engine/tests/helpers/snapshotRepairEngine.ts

@@ -0,0 +1,24 @@
+// A minimal engine for asserting the append-failure repair binding. No decorator and no Redis
+// snapshot store: the property under test is the job id's dedupe, which lives on the engine.
+import { trace } from "@internal/tracing";
+
+export function engineOptionsForSnapshotRepair(prisma: unknown, redisOptions: unknown) {
+  return {
+    prisma,
+    worker: { redis: redisOptions, workers: 1, tasksPerWorker: 10, pollIntervalMs: 100 },
+    queue: {
+      redis: redisOptions,
+      masterQueueConsumersDisabled: true,
+      processWorkerQueueDebounceMs: 50,
+    },
+    runLock: { redis: redisOptions },
+    machines: {
+      defaultMachine: "small-1x" as const,
+      machines: {
+        "small-1x": { name: "small-1x" as const, cpu: 0.5, memory: 0.5, centsPerMs: 0.0001 },
+      },
+      baseCostInCents: 0.0001,
+    },
+    tracer: trace.getTracer("test", "0.0.0"),
+  };
+}

internal-packages/run-engine/src/engine/tests/snapshotRepairBinding.test.ts

@@ -0,0 +1,48 @@
+import { containerTest } from "@internal/testcontainers";
+import { expect } from "vitest";
+import { RunEngine } from "../index.js";
+import { engineOptionsForSnapshotRepair } from "./helpers/snapshotRepairEngine.js";
+
+containerTest(
+  "a second repair for the same run enqueues no second job",
+  async ({ prisma, redisOptions }) => {
+    const engine = new RunEngine(engineOptionsForSnapshotRepair(prisma, redisOptions) as never);
+
+    try {
+      const payload = {
+        runId: "run_repair_dedupe",
+        snapshotId: "snap_1",
+        executionStatus: "EXECUTING",
+      };
+
+      // The stall watchdog uses this same job id, so the two compensators must collapse to one.
+      expect(await engine.enqueueSnapshotRepair(payload)).toBe(true);
+      expect(await engine.enqueueSnapshotRepair({ ...payload, snapshotId: "snap_2" })).toBe(false);
+    } finally {
+      await engine.quit();
+    }
+  }
+);
+
+containerTest("a different run gets its own repair job", async ({ prisma, redisOptions }) => {
+  const engine = new RunEngine(engineOptionsForSnapshotRepair(prisma, redisOptions) as never);
+
+  try {
+    expect(
+      await engine.enqueueSnapshotRepair({
+        runId: "run_a",
+        snapshotId: "snap_a",
+        executionStatus: "EXECUTING",
+      })
+    ).toBe(true);
+    expect(
+      await engine.enqueueSnapshotRepair({
+        runId: "run_b",
+        snapshotId: "snap_b",
+        executionStatus: "EXECUTING",
+      })
+    ).toBe(true);
+  } finally {
+    await engine.quit();
+  }
+});