fix(run-engine,webapp): harden waitpoint blip-resilience guards after review

Make the write-ahead guards for waitpoint completion safe across every queue stage.

- The publish guard re-publishes a run only when it is genuinely absent from the queue (checking the sorted set and the run's concurrency claim), so it can no longer duplicate a run already dispatched to a worker or strip a live run's concurrency. Its decide-then-publish runs under the same run lock as the original enqueue, so the guard and the resume publisher can never both publish across a dispatch.
- Interrupted-resume recovery now clears leaked blocking edges and re-installs the run's heartbeat, so a resume that committed but died before finishing can no longer strand the run or leave it without a stall watchdog.
- The publish guard is scoped to the waitpoint resume path.
- The blip-retry flag is read synchronously from an in-memory poller warmed at startup, so the completion path does no database work and adds no latency when the flag is off.

Author: d-cs

apps/webapp/app/entry.server.tsx

@@ -12,6 +12,7 @@ import { initMollifierStaleSweepWorker } from "~/v3/mollifierStaleSweepWorker.se
 import { initBillingLimitWorker } from "~/v3/billingLimitWorker.server";
 import { initLogsSearchProjectorWorker } from "~/v3/logsSearchProjectorWorker.server";
 import { initQueueMetricsConsumer, initQueueMetricsEmitter } from "~/v3/queueMetrics.server";
+import { startRunStoreInfraRetryFlagPoller } from "~/v3/runStoreInfraRetryFlag.server";
 import { bootstrap } from "./bootstrap";
 import { LocaleContextProvider } from "./components/primitives/LocaleProvider";
 import type { OperatingSystemPlatform } from "./components/primitives/OperatingSystemProvider";
@@ -282,6 +283,9 @@ initLogsSearchProjectorWorker();
 initQueueMetricsEmitter();
 initQueueMetricsConsumer();
 
+// Warm the blip-retry flag poller at startup so the operation-path getter stays purely synchronous.
+startRunStoreInfraRetryFlagPoller();
+
 bootstrap().catch((error) => {
   logError(error);
 });

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

@@ -65,20 +65,12 @@ export function makeFlag(_prisma: PrismaClientOrTransaction = prisma) {
 }
 
 const cachedFlagStore = new Map<string, { value: unknown; expiresAt: number }>();
-// In-flight refreshes, keyed like cachedFlagStore, so concurrent callers on a cold/expired entry
-// share ONE database read instead of stampeding it (every ttl window, or repeatedly while it fails).
-const cachedFlagInFlight = new Map<string, Promise<unknown>>();
 
 /**
  * flag() behind a short process-level TTL cache, for global flags read on hot
  * paths (e.g. the root loader) where a database round-trip per request is too
  * expensive. Flips propagate within ttlMs per process. Overrides are rejected
  * by the type: a scoped resolution must never be reused across scopes.
- *
- * Failure-safe: if the underlying read throws (e.g. the control-plane DB is
- * blipping), it never throws to the caller — it serves the last cached value if
- * there is one, otherwise `defaultValue`. A read on hot paths must not turn a DB
- * blip into a 5xx, and a caller that gates behavior on this must degrade safely.
  */
 export async function cachedFlag<T extends FeatureFlagKey>(
   opts: Omit<FlagsOptions<T>, "overrides"> & {
@@ -93,27 +85,9 @@ export async function cachedFlag<T extends FeatureFlagKey>(
     return hit.value as z.infer<(typeof FeatureFlagCatalog)[T]>;
   }
 
-  const existing = cachedFlagInFlight.get(cacheKey);
-  if (existing) {
-    return existing as Promise<z.infer<(typeof FeatureFlagCatalog)[T]>>;
-  }
-
-  const refresh = (async () => {
-    try {
-      const value = await flag(opts);
-      cachedFlagStore.set(cacheKey, { value, expiresAt: Date.now() + ttlMs });
-      return value;
-    } catch {
-      // Serve the stale value if we have one, otherwise the default. Do not cache the failure, so
-      // the next call retries the refresh (single-flighted).
-      return hit ? (hit.value as z.infer<(typeof FeatureFlagCatalog)[T]>) : opts.defaultValue;
-    } finally {
-      cachedFlagInFlight.delete(cacheKey);
-    }
-  })();
-
-  cachedFlagInFlight.set(cacheKey, refresh);
-  return refresh as Promise<z.infer<(typeof FeatureFlagCatalog)[T]>>;
+  const value = await flag(opts);
+  cachedFlagStore.set(cacheKey, { value, expiresAt: Date.now() + ttlMs });
+  return value;
 }
 
 export function makeSetFlag(_prisma: PrismaClientOrTransaction = prisma) {

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

@@ -3,48 +3,69 @@ import { logger } from "~/services/logger.server";
 import { flag } from "./featureFlags.server";
 
 // The run-store blip-retry gate, resolved from the global `runStoreInfraRetryEnabled` flag on a
-// background timer and cached in memory. The store consults this synchronously per operation, so:
-//   - it does NO database read on the hot path — a connection blip can never turn the gate itself
-//     into a failure or disable resilience mid-blip (the last value is held in memory), and
-//   - a failing flag read never stampedes the control-plane pool (one timer, not one query per op).
-// A failed refresh keeps the last known value; the only cold window is a brand-new process whose
-// first refresh has not completed, which resolves to the safe `false`.
+// background timer and held in memory. The store consults it SYNCHRONOUSLY per operation, so:
+//   - the operation path does ZERO database work and never awaits — a blip can neither add latency
+//     nor turn the gate itself into a failure or disable resilience mid-blip (the last value is held),
+//   - a failing refresh keeps the last known value, and
+//   - a single background timer (not one query per op) never stampedes the control-plane pool.
+// The poller is warmed at process startup (see startRunStoreInfraRetryFlagPoller). The only cold
+// window is a brand-new process before its first read completes, which safely reads `false`.
 
 const REFRESH_INTERVAL_MS = 30_000;
 
 let current = false;
-let initPromise: Promise<void> | null = null;
+let started = false;
+let warmup: Promise<void> | null = null;
 
-async function refresh(): Promise<void> {
+async function refresh(): Promise<boolean> {
   try {
     current = await flag({ key: FEATURE_FLAG.runStoreInfraRetryEnabled, defaultValue: false });
+    return true;
   } catch (error) {
     // Keep the last known value: a flag-store blip must not flip resilience on or off.
     logger.debug("runStoreInfraRetry flag refresh failed; keeping last value", {
       error: error instanceof Error ? error.message : String(error),
     });
+    return false;
   }
 }
 
-function ensureStarted(): Promise<void> {
-  if (initPromise) {
-    return initPromise;
+async function initialRead(): Promise<void> {
+  // Bounded retry of the FIRST read so a brief blip at startup doesn't leave resilience off until the
+  // next tick. Never on the operation path. If the store is unreachable the whole window, `current`
+  // stays at the safe default (false) and self-heals on a later tick.
+  for (let attempt = 0; attempt < 5; attempt++) {
+    if (await refresh()) {
+      return;
+    }
+    await new Promise((resolve) => setTimeout(resolve, 200));
   }
-  // One initial read, shared by all early callers, so a globally-enabled flag is effective from the
-  // FIRST operation of a fresh process (not only after the first background tick). Later calls await
-  // this already-resolved promise and return the in-memory value with no further database work.
-  initPromise = refresh();
+}
+
+/**
+ * Start the background poller once (idempotent). Call at process startup to warm the gate before
+ * traffic. It kicks off the initial read and the refresh timer WITHOUT blocking; the returned promise
+ * resolves when the first read settles, so a caller MAY await readiness at boot, but the operation
+ * path never does.
+ */
+export function startRunStoreInfraRetryFlagPoller(): Promise<void> {
+  if (started) {
+    return warmup ?? Promise.resolve();
+  }
+  started = true;
+  warmup = initialRead();
   const timer = setInterval(() => void refresh(), REFRESH_INTERVAL_MS);
   timer.unref?.(); // don't keep the process alive for the poll timer
-  return initPromise;
+  return warmup;
 }
 
 /**
- * The run-store blip-retry gate. The first call awaits a single initial read (so op #1 sees the real
- * flag); every later call resolves from the in-memory value with no database work — so the gate does
- * no per-op DB read, survives a blip (last value held), and never stampedes the control-plane pool.
+ * The run-store blip-retry gate: a PURELY SYNCHRONOUS, database-free read of the in-memory value the
+ * background poller maintains. It never awaits and does no database work, so it adds no latency and
+ * cannot itself fail or hang during a blip. It only ensures the poller is running (non-blocking) as a
+ * fallback for entrypoints that skip the explicit startup warm.
  */
-export async function isRunStoreInfraRetryEnabled(): Promise<boolean> {
-  await ensureStarted();
+export function isRunStoreInfraRetryEnabled(): boolean {
+  startRunStoreInfraRetryFlagPoller();
   return current;
 }

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

@@ -107,7 +107,7 @@ import pMap from "p-map";
 export class RunEngine {
   private runLockRedis: Redis;
   private runLock: RunLocker;
-  private worker: EngineWorker;
+  worker: EngineWorker;
   private ttlWorker: Worker<ReturnType<typeof createTtlWorkerCatalog>>;
   private logger: Logger;
   private tracer: Tracer;

internal-packages/run-engine/src/engine/systems/enqueueSystem.ts

@@ -38,6 +38,7 @@ export class EnqueueSystem {
     workerId,
     runnerId,
     skipRunLock,
+    armPublishGuard = false,
     includeTtl = false,
     anchorEligibilityAtQueuePosition = false,
     enableFastPath = false,
@@ -61,6 +62,17 @@ export class EnqueueSystem {
     workerId?: string;
     runnerId?: string;
     skipRunLock?: boolean;
+    /**
+     * Arm the write-ahead publish guard before the snapshot write (see below). Only the waitpoint
+     * SUSPENDED->QUEUED resume sets this: it publishes with default options (so the guard's option-less
+     * replay is faithful) AND the run was SUSPENDED with its concurrency already released, which the
+     * guard's in-flight check relies on to tell a lost publish from a live run. The delayed and
+     * pending-version enqueues do NOT set it (they need `includeTtl`/`anchorEligibilityAtQueuePosition`
+     * the guard can't reconstruct); the checkpoint re-queue does NOT either (it releases concurrency
+     * after enqueue, so a lost publish there would leave a claim the in-flight check can't distinguish).
+     * Default false. Still gated by the runtime blip-retry flag.
+     */
+    armPublishGuard?: boolean;
     /**
      * When true, arm the run's TTL on the queued message. Set by every path that is the run's
      * first real entry into the queue: trigger, a delayed run coming due, and the pending-version
@@ -95,8 +107,10 @@ export class EnqueueSystem {
       // Write-ahead publish guard: the snapshot write (Postgres) and the queue publish (Redis) are
       // not atomic, so a lost publish would leave a QUEUED run absent from the queue. When enabled,
       // pre-mint the snapshot id, arm the guard keyed by it BEFORE the snapshot write, and ack it
-      // only after the publish succeeds; the guard replays the publish idempotently otherwise.
-      const armGuard = await this.$.isBlipRetryEnabled();
+      // only after the publish succeeds; the guard replays the publish idempotently otherwise. Scoped
+      // to the resume re-enqueues (armPublishGuard) so it never has to reconstruct publish options it
+      // wasn't given; the flag check is skipped entirely when not armed, so other paths do no extra work.
+      const armGuard = armPublishGuard ? await this.$.isBlipRetryEnabled() : false;
       const snapshotId = armGuard ? SnapshotId.generate().id : undefined;
       if (armGuard && snapshotId) {
         await this.#scheduleRunPublishedGuard(run.id, snapshotId);
@@ -134,35 +148,53 @@ export class EnqueueSystem {
         enableFastPath,
       });
 
-      if (armGuard) {
-        await this.#ackRunPublishedGuard(run.id);
+      if (armGuard && snapshotId) {
+        await this.#ackRunPublishedGuard(run.id, snapshotId);
       }
 
       return newSnapshot;
     });
   }
 
-  #runPublishedGuardId(runId: string): string {
-    return `ensureRunPublished:${runId}`;
+  // Keyed by BOTH run id and snapshot id: a run goes through many QUEUED transitions over its life,
+  // so a run-only key would let a stale guard from an earlier transition block enqueueOnce for the
+  // next one, leaving the newer transition unprotected.
+  #runPublishedGuardId(runId: string, snapshotId: string): string {
+    return `ensureRunPublished:${runId}:${snapshotId}`;
   }
 
   async #scheduleRunPublishedGuard(runId: string, snapshotId: string): Promise<void> {
     await this.$.worker.enqueueOnce({
-      id: this.#runPublishedGuardId(runId),
+      id: this.#runPublishedGuardId(runId, snapshotId),
       job: "ensureRunPublished",
       payload: { runId, snapshotId },
       availableAt: new Date(Date.now() + this.$.guardDelayMs),
     });
   }
 
-  async #ackRunPublishedGuard(runId: string): Promise<void> {
-    await this.$.worker.ack(this.#runPublishedGuardId(runId));
+  async #ackRunPublishedGuard(runId: string, snapshotId: string): Promise<void> {
+    await this.$.worker.ack(this.#runPublishedGuardId(runId, snapshotId));
   }
 
   /**
-   * Redelivery handler for the publish guard: re-publishes a run whose snapshot committed but whose
-   * queue publish was lost. Idempotent — publishRun/enqueueMessage dedupes on run id — and a no-op
-   * once the snapshot is superseded (already dequeued/executing) so it never re-queues a live run.
+   * Redelivery handler for the publish guard: re-publishes a resume-enqueued run whose snapshot
+   * committed but whose queue publish was lost. It re-publishes ONLY when the run is genuinely absent
+   * from the queue, which two checks establish together:
+   *   1. `snapshotId` is still the latest snapshot AND is QUEUED. Once a consumer dequeues the run the
+   *      snapshot moves off QUEUED, so a lost ack after a successful publish is a no-op here.
+   *   2. `messageInFlight` is false. While the snapshot still reads QUEUED the message may already be
+   *      live in the queue: either waiting in the sorted set, or dispatched to a worker queue (where it
+   *      holds a concurrency claim) but not yet consumed. Re-publishing in that window would BOTH add a
+   *      duplicate queue entry AND strip the live run's concurrency claim (the enqueue Lua SREMs it), so
+   *      we must skip. The guard is scoped to resume enqueues, and a suspend releases all concurrency,
+   *      so a genuinely-lost resume-publish holds no claim and correctly re-publishes.
+   * We deliberately do NOT gate on the message key existing: it is written at trigger and lives for the
+   * whole run lifecycle (deleted only on ack/TTL-expiry), so it is present for every resume.
+   *
+   * The whole decide-then-publish runs under the SAME run lock as `enqueueRun` (keyed by runId), so the
+   * original resume publisher and this guard cannot both observe "absent" and publish across a
+   * queue-dispatch transition: whichever runs second acquires the lock only after the first has finished
+   * publishing, sees the message in-flight, and skips.
    */
   public async ensureRunPublished({
     runId,
@@ -171,21 +203,33 @@ export class EnqueueSystem {
     runId: string;
     snapshotId: string;
   }): Promise<void> {
-    const run = await this.$.runStore.findRun({ id: runId }, this.$.prisma);
-    if (!run) {
-      return;
-    }
-    const latest = await getLatestExecutionSnapshot(this.$.prisma, runId, this.$.runStore);
-    if (latest.id !== snapshotId || latest.executionStatus !== "QUEUED") {
-      // Superseded (already dequeued/executing) or a different transition owns the run now.
-      return;
-    }
-    const env = await this.$.controlPlaneResolver.resolveEnv(run.runtimeEnvironmentId);
-    if (!env) {
-      this.$.logger.error("ensureRunPublished: environment not found", { runId });
-      return;
-    }
-    await this.publishRun({ run, env });
+    await this.$.runLock.lock("ensureRunPublished", [runId], async () => {
+      const run = await this.$.runStore.findRun({ id: runId }, this.$.prisma);
+      if (!run) {
+        return;
+      }
+      const latest = await getLatestExecutionSnapshot(this.$.prisma, runId, this.$.runStore);
+      if (latest.id !== snapshotId || latest.executionStatus !== "QUEUED") {
+        // Superseded (already dequeued/executing) or a different transition owns the run now.
+        return;
+      }
+      const env = await this.$.controlPlaneResolver.resolveEnv(run.runtimeEnvironmentId);
+      if (!env) {
+        this.$.logger.error("ensureRunPublished: environment not found", { runId });
+        return;
+      }
+      const inFlight = await this.$.runQueue.messageInFlight(
+        env,
+        run.queue,
+        run.id,
+        run.concurrencyKey ?? undefined
+      );
+      if (inFlight) {
+        // Already waiting or dispatched; re-publishing would duplicate it and strip its concurrency.
+        return;
+      }
+      await this.publishRun({ run, env });
+    });
   }
 
   /**