feat(webapp,run-engine): take up the store and sweeper connection contract

The store now receives a pre-built client instead of building its own, so
cluster mode reaches the hot path and not only the sweep. The webapp owns both
sockets and closes both, with the hot-path client last so an append in flight
can still land.

The sweep receives the dedicated client, the confirm window that was parsed and
unused, and the deadline and abort signal it now accepts. So the pass budget is
enforced inside a pass rather than only bounding the lock, and a truncated pass
reports itself as partial.

The engine also arms an abort a minute past the budget. The deadline is the
sweep's own stopping rule, checked at batch boundaries; this is the backstop for
a pass that has stopped reaching one, so the signal is not decorative.

Author: d-cs

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

@@ -64,6 +64,7 @@ function buildClient(name: string): RedisClient {
 
 type Instance = {
   sweepClient: RedisClient;
+  hotPathClient: RedisClient;
   redisSnapshotStore: RedisSnapshotStore;
   decorate: (store: RunStore) => RunStore;
 };
@@ -79,16 +80,17 @@ const instance = singleton<Instance | undefined>("snapshotStoreInstance", () =>
   // transition append. It also backs the sweep's exclusion lock.
   const sweepClient = buildClient("sweep");
 
+  const hotPathClient = buildClient("store");
+
   const redisSnapshotStore = new RedisSnapshotStore({
-    // The store still builds its own single-node client, so cluster mode reaches the sweep but not
-    // the hot path. Hand it a pre-built client once its options accept one.
-    redisOptions: redisOptions(),
+    client: hotPathClient,
     completedTtlMs: env.RUN_ENGINE_SNAPSHOT_STORE_COMPLETED_TTL_MS,
     metrics: metrics.store,
   });
 
   return {
     sweepClient,
+    hotPathClient,
     redisSnapshotStore,
     decorate: (store: RunStore) =>
       new TaskRunExecutionSnapshotStore(store, {
@@ -145,6 +147,8 @@ export async function quitSnapshotStoreClients(): Promise<void> {
     await quit().catch(() => undefined);
   }
   await instance.sweepClient.quit().catch(() => undefined);
-  // Last: an append in flight must still land.
+  // Last: an append in flight must still land. The store's own quit() returns early on a
+  // caller-supplied client, so closing the socket is ours.
   await instance.redisSnapshotStore.quit().catch(() => undefined);
+  await instance.hotPathClient.quit().catch(() => undefined);
 }

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

@@ -23,27 +23,20 @@ export function registerSnapshotStoreWiring(): boolean {
   });
 
   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: {} }),
-    },
+    // Its own connection, so a scan of every master can never stall a transition append.
+    client: sweepClient,
     // 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,
+    confirmOrphanAfterMs: env.RUN_ENGINE_SNAPSHOT_STORE_CONFIRM_ORPHAN_AFTER_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()) }),
+      sweep: async ({ deadline, signal }) => ({ ...(await sweeper.sweep({ deadline, signal })) }),
       lockTtlMs: env.RUN_ENGINE_SNAPSHOT_STORE_GC_SWEEP_BUDGET_MS + 3_600_000,
     })
   );

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

@@ -2961,6 +2961,9 @@ export class RunEngine {
 
     const budgetMs = this.options.snapshotStore?.sweepBudgetMs ?? 10_800_000;
     const controller = new AbortController();
+    // The deadline is the sweep's own stopping rule; this is the backstop for a pass that has
+    // stopped reaching a batch boundary, so the signal is not merely decorative.
+    const abortAt = globalThis.setTimeout(() => controller.abort(), budgetMs + 60_000);
     let outcome = "failed";
     let counts: Record<string, number | boolean> | undefined;
 
@@ -2974,6 +2977,7 @@ export class RunEngine {
     } catch (error) {
       this.logger.error("sweepSnapshotOrphans threw", { error });
     } finally {
+      globalThis.clearTimeout(abortAt);
       this.snapshotSweepPassCounter?.add(1, { outcome });
       for (const [field, value] of Object.entries(counts ?? {})) {
         if (typeof value === "number") {