fix(webapp,run-store): close snapshot-store activation-time correctness gaps

Found by an adversarial review of the off-by-default snapshot store. None
affect the merged-inert state; all bite only once a Redis host is configured
and orgs are ramped.

- The v1 programmatic org flag route set the per-org dial but never stamped
  the one-way residency latch (only the v2 route did), so a run born after a
  v1 enable was resident while the census kept classifying the org
  never-enabled and skipped its transitions, freezing its Redis head. Stamp
  the latch and refresh the census, matching v2.
- At redis-only a non-throwing append outcome (forked or skippedNoKeyspace)
  reached the outcome handler and enqueued a Postgres repair or was dropped,
  but Postgres holds no snapshot there. Treat both as fatal at redis-only,
  matching the thrown-error path, so the caller retries rather than losing
  the transition.
- A failed post-save primary read of an org dial left it wedged on the
  deployment-wide position until the next save or a restart. Retry the
  primary briefly, then clear the pending flag so refreshes resume.
- Correct the redis-only boot warning: this build suppresses Postgres
  snapshot writes at redis-only, so the old "always writes Postgres" note
  was misleading.

Author: d-cs

apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.feature-flags.ts

@@ -9,9 +9,11 @@ import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.
 import { globalFlagsRegistry } from "~/v3/globalFlagsRegistry.server";
 import { snapshotStoreFlagSaveError } from "~/v3/snapshotStoreFlagGuard.server";
 import { invalidateSnapshotStoreOrgMode } from "~/v3/snapshotStoreMode.server";
+import { snapshotStoreOrgCensus } from "~/v3/snapshotStoreOrgCensus.server";
 import { selectMintBaselineSource, stampMintKindFlip } from "~/v3/runOpsMigration/mintFlipGrace";
 import {
   FEATURE_FLAG,
+  stampSnapshotStoreOrgEverEnabled,
   validatePartialFeatureFlags,
   withoutOrgForbiddenSnapshotKeys,
 } from "~/v3/featureFlags";
@@ -128,6 +130,12 @@ export async function action({ request, params }: ActionFunctionArgs) {
         env.RUN_OPS_MINT_FLIP_GRACE_MS
       );
 
+      // One-way per-org residency latch, exactly as the v2 route does. Without it a run born after
+      // this enable is resident but the census keeps classifying the org definitely-never-enabled, so
+      // its transitions are skipped and its Redis head freezes. ORed against the locked existing value
+      // so a save back to off never clears it.
+      stampSnapshotStoreOrgEverEnabled(existingRaw, mergedFlags);
+
       return tx.organization.update({
         where: {
           id: organizationId,
@@ -150,6 +158,9 @@ export async function action({ request, params }: ActionFunctionArgs) {
     // Org feature flags are embedded in every env of the org; drop all its cached env rows.
     controlPlaneResolver.invalidateOrganization(organizationId);
     invalidateSnapshotStoreOrgMode(organizationId);
+    // Refresh the census in THIS process at once, as the v2 route does, so a just-enabled org stops
+    // reading as definitely-never-enabled here immediately. Other pods lag at most the reload interval.
+    void snapshotStoreOrgCensus.refresh();
 
     const updatedFlagsResult = updatedOrganization.featureFlags
       ? validatePartialFeatureFlags(updatedOrganization.featureFlags as Record<string, unknown>)

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

@@ -95,7 +95,7 @@ export async function assertSnapshotStoreBoot(deps: SnapshotStoreBootDeps): Prom
 
   if (deps.mode === "redis-only") {
     deps.warn(
-      "Snapshot store dial is redis-only but this build always writes Postgres snapshots. Double-writing is safe, but a Redis fault at this position fails run creation.",
+      "Snapshot store dial is redis-only: Postgres snapshot writes are suppressed, so Redis is the sole store for run snapshots. A Redis fault at this position fails run creation and cannot be repaired from Postgres, and rolling the dial down does not recover runs born here.",
       { mode: deps.mode }
     );
   }

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

@@ -186,6 +186,12 @@ const WARM_TIMEOUT_MS = 500;
 const DEFAULT_CACHE_MAX = 10_000;
 const DEFAULT_CACHE_TTL_MS = 30_000;
 
+// A failed post-save primary read must not wedge the org on the global fallback forever: retry the
+// PRIMARY a few times (the replica stays blocked meanwhile so it cannot restore the superseded value),
+// then give up and let normal refreshes resume. A bounded stale window beats a permanent one.
+const DEFAULT_PRIMARY_INVALIDATE_RETRY_DELAYS_MS = [100, 250, 500, 1000];
+const sleep = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));
+
 type OrgModeClient = {
   organization: {
     findFirst(args: {
@@ -195,12 +201,17 @@ type OrgModeClient = {
   };
 };
 
-export function createOrgModeSource(clients?: {
-  primary: OrgModeClient;
-  replica: OrgModeClient;
-}): OrgModeSource {
+export function createOrgModeSource(
+  clients?: {
+    primary: OrgModeClient;
+    replica: OrgModeClient;
+  },
+  opts?: { primaryInvalidateRetryDelaysMs?: number[] }
+): OrgModeSource {
   const primaryClient = (clients?.primary ?? prisma) as OrgModeClient;
   const replicaClient = (clients?.replica ?? $replica) as OrgModeClient;
+  const primaryInvalidateRetryDelaysMs =
+    opts?.primaryInvalidateRetryDelaysMs ?? DEFAULT_PRIMARY_INVALIDATE_RETRY_DELAYS_MS;
   // Defaults inline as well as in the schema: this must not throw when a caller supplies a partial
   // env, and an LRU with neither bound set is a constructor error.
   const cache = new LRUCache<string, CachedOrg>({
@@ -233,22 +244,7 @@ export function createOrgModeSource(clients?: {
       generations.set(organizationId, generation);
       cache.delete(organizationId);
       primaryPending.set(organizationId, generation);
-      void load(organizationId, primaryClient, generation).then((outcome) => {
-        // Only if no NEWER save has claimed it since. Deleting unconditionally is what let an older
-        // read reopen the window for a newer one.
-        if (primaryPending.get(organizationId) !== generation) {
-          return;
-        }
-        // And only if the primary actually ANSWERED. `load` swallows its own errors, so a rejected
-        // primary read used to clear the flag with nothing cached, reopening the window to a lagging
-        // replica that would then restore the pre-save value for a full cache lifetime. Staying
-        // pending keeps replica refreshes out until a primary read succeeds; the resolver falls back
-        // to the deployment-wide position meanwhile, which is the safe answer.
-        if (outcome === "failed") {
-          return;
-        }
-        primaryPending.delete(organizationId);
-      });
+      void loadWithPrimaryRetry(organizationId, generation);
     },
     refresh: (organizationId) => {
       // A save is mid-read for this organisation. Its answer is authoritative and a replica cannot
@@ -292,6 +288,36 @@ export function createOrgModeSource(clients?: {
     },
   };
 
+  // The save's own primary read, with bounded retry. primaryPending stays set for `generation` across
+  // retries so a lagging replica cannot restore the superseded value; a success clears it, and so does
+  // exhausting the retries, so a persistent primary fault falls back to normal refreshes rather than
+  // wedging the org on the global position until the next save or a restart.
+  async function loadWithPrimaryRetry(organizationId: string, generation: number): Promise<void> {
+    for (let attempt = 0; ; attempt++) {
+      const outcome = await load(organizationId, primaryClient, generation);
+      // A newer save (or its read) has superseded this one; it now owns the flag.
+      if (primaryPending.get(organizationId) !== generation) {
+        return;
+      }
+      if (outcome !== "failed") {
+        primaryPending.delete(organizationId);
+        return;
+      }
+      if (attempt >= primaryInvalidateRetryDelaysMs.length) {
+        // Retries exhausted: clear the flag so replica refreshes resume. The next save or reload
+        // corrects any briefly-restored stale value; a permanent wedge would not.
+        if (primaryPending.get(organizationId) === generation) {
+          primaryPending.delete(organizationId);
+        }
+        return;
+      }
+      await sleep(primaryInvalidateRetryDelaysMs[attempt]);
+      if (primaryPending.get(organizationId) !== generation) {
+        return;
+      }
+    }
+  }
+
   function load(
     organizationId: string,
     client: OrgModeClient,

apps/webapp/test/adminOrgFeatureFlagsV1Route.test.ts

@@ -0,0 +1,129 @@
+// The v1 PAT route enables an org's snapshot dial with merge semantics. It must stamp the one-way
+// per-org residency latch (snapshotStoreOrgEverEnabled) exactly as the v2 route does; without it the
+// census keeps the org classified definitely-never-enabled and its resident runs' transitions are
+// skipped. These drive the real exported action against a real Postgres and assert the stored blob.
+// The guard is stubbed to a pass so the test isolates the stamp wiring (the guard has its own tests);
+// only peripheral module boundaries are substituted, the database is the genuine article.
+import type { PrismaClient } from "@trigger.dev/database";
+import { postgresTest } from "@internal/testcontainers";
+import { describe, expect, vi } from "vitest";
+import { FEATURE_FLAG } from "~/v3/featureFlags";
+
+vi.setConfig({ testTimeout: 60_000 });
+
+const db = vi.hoisted(() => ({ client: null as unknown as PrismaClient }));
+
+vi.mock("~/services/personalAccessToken.server", () => ({
+  requireAdminApiRequest: async () => ({}),
+}));
+
+vi.mock("~/db.server", () => ({
+  get prisma() {
+    return db.client;
+  },
+  get $replica() {
+    return db.client;
+  },
+}));
+
+// Bypass the host/arming-latch gate so the test exercises the stamp path directly. The guard's own
+// behaviour (reject without a host or before the global latch) is covered in its own tests.
+vi.mock("~/v3/snapshotStoreFlagGuard.server", () => ({
+  snapshotStoreFlagSaveError: () => undefined,
+}));
+
+vi.mock("~/v3/runOpsMigration/controlPlaneResolver.server", () => ({
+  controlPlaneResolver: { invalidateOrganization: () => {} },
+}));
+
+// Only used to seed the mint-flip baseline; irrelevant to the residency latch under test.
+vi.mock("~/v3/featureFlags.server", () => ({
+  flags: async () => ({}),
+}));
+
+import { action } from "~/routes/admin.api.v1.orgs.$organizationId.feature-flags";
+
+const MODE = FEATURE_FLAG.snapshotStoreOrgMode;
+const LATCH = FEATURE_FLAG.snapshotStoreOrgEverEnabled;
+
+let orgSeq = 0;
+
+async function seedOrg(prisma: PrismaClient, featureFlags?: Record<string, unknown>) {
+  db.client = prisma;
+  const id = `org_v1route_${orgSeq++}`;
+  await prisma.organization.create({
+    data: {
+      id,
+      title: "V1 route test org",
+      slug: `v1-route-${id}`,
+      ...(featureFlags ? { featureFlags } : {}),
+    },
+  });
+  return id;
+}
+
+async function post(organizationId: string, body: unknown) {
+  const request = new Request(
+    `https://localhost:3030/admin/api/v1/orgs/${organizationId}/feature-flags`,
+    {
+      method: "POST",
+      body: JSON.stringify(body),
+      headers: { "content-type": "application/json" },
+    }
+  );
+  return (await (action as any)({
+    request,
+    params: { organizationId },
+    context: {},
+  })) as Response;
+}
+
+async function readFlags(prisma: PrismaClient, id: string) {
+  const row = await prisma.organization.findFirst({
+    where: { id },
+    select: { featureFlags: true },
+  });
+  return (row?.featureFlags ?? null) as Record<string, unknown> | null;
+}
+
+describe("admin v1 org feature-flags route stamps the per-org residency latch", () => {
+  postgresTest("stamps the latch when the dial is enabled past off", async ({ prisma }) => {
+    const id = await seedOrg(prisma);
+
+    const response = await post(id, { [MODE]: "redis-read" });
+
+    expect(response.status).toBe(200);
+    const flags = await readFlags(prisma, id);
+    expect(flags?.[MODE]).toBe("redis-read");
+    expect(flags?.[LATCH]).toBe(true);
+  });
+
+  postgresTest(
+    "keeps the latch when a later save sets the dial back to off",
+    async ({ prisma }) => {
+      const id = await seedOrg(prisma);
+
+      await post(id, { [MODE]: "redis-read" });
+      const response = await post(id, { [MODE]: "off" });
+
+      expect(response.status).toBe(200);
+      const flags = await readFlags(prisma, id);
+      expect(flags?.[MODE]).toBe("off");
+      // One-way: the latch survives the roll-back so a run still resident keeps mirroring.
+      expect(flags?.[LATCH]).toBe(true);
+    }
+  );
+
+  postgresTest(
+    "does not stamp the latch for an off save with no prior latch",
+    async ({ prisma }) => {
+      const id = await seedOrg(prisma);
+
+      const response = await post(id, { [MODE]: "off" });
+
+      expect(response.status).toBe(200);
+      const flags = await readFlags(prisma, id);
+      expect(flags?.[LATCH]).toBeUndefined();
+    }
+  );
+});

apps/webapp/test/snapshotStoreMode.test.ts

@@ -386,31 +386,74 @@ describe("warming the organisation dial before a birth", () => {
     // Still unknown, so the synchronous resolve falls back exactly as it did before.
     expect(source.get("org_1")).toBeUndefined();
   });
-  it("keeps replica refreshes out after a FAILED primary read", async () => {
-    // load() swallows its own errors, so a rejected primary read used to clear primaryPending with
-    // nothing cached. A refresh could then read a lagging replica carrying the current generation,
-    // which the generation guard cannot discard, restoring the pre-save value for a cache lifetime.
-    const primary = deferred();
-    const replica = deferred();
-    const source = createOrgModeSource({
-      primary: clientFor(() => primary.promise),
-      replica: clientFor(() => replica.promise),
-    });
+  it("keeps replica refreshes out while a failed primary read is still retrying", async () => {
+    // While the primary is failing and retries are pending, a lagging replica must not restore the
+    // pre-save value: primaryPending stays set for the generation, so refresh does not read the replica.
+    let primaryCalls = 0;
+    const source = createOrgModeSource(
+      {
+        primary: clientFor(() => {
+          primaryCalls += 1;
+          return Promise.reject(new Error("primary unavailable"));
+        }),
+        replica: clientFor(() =>
+          Promise.resolve({ featureFlags: { snapshotStoreOrgMode: "off" } })
+        ),
+      },
+      // Long delays: the retry window is still open for the duration of this test.
+      { primaryInvalidateRetryDelaysMs: [10_000, 10_000] }
+    );
 
     source.invalidate("org_1");
+    await vi.waitFor(() => expect(primaryCalls).toBeGreaterThanOrEqual(1));
 
-    // The primary read FAILS.
-    primary.resolve(Promise.reject(new Error("primary unavailable")) as never);
-    await new Promise((r) => setTimeout(r, 0));
+    // A refresh arriving mid-retry must not start a replica read; the resolver falls back to global.
+    source.refresh("org_1");
     await new Promise((r) => setTimeout(r, 0));
+    expect(source.get("org_1")).toBeUndefined();
+  });
 
-    // A refresh arriving now must not start a replica read, because no primary answer ever landed.
+  it("retries the primary and recovers the saved value after a transient failure", async () => {
+    let calls = 0;
+    const source = createOrgModeSource(
+      {
+        primary: clientFor(() => {
+          calls += 1;
+          return calls === 1
+            ? Promise.reject(new Error("primary unavailable"))
+            : Promise.resolve({ featureFlags: { snapshotStoreOrgMode: "dual-write" } });
+        }),
+        replica: clientFor(() => Promise.resolve({ featureFlags: {} })),
+      },
+      { primaryInvalidateRetryDelaysMs: [5, 5, 5] }
+    );
+
+    source.invalidate("org_1");
+    // A concurrent refresh during the retry must not read the replica out from under the retry.
     source.refresh("org_1");
-    replica.resolve({ featureFlags: { snapshotStoreOrgMode: "off" } });
-    await new Promise((r) => setTimeout(r, 0));
 
-    // Nothing cached: the resolver falls back to the deployment-wide position, which is the safe
-    // answer, rather than serving a stale replica value as though it were the saved one.
-    expect(source.get("org_1")).toBeUndefined();
+    await vi.waitFor(() => expect(source.get("org_1")).toBe("dual-write"));
+    expect(calls).toBeGreaterThanOrEqual(2);
+  });
+
+  it("unwedges after exhausting primary retries so a later refresh repopulates", async () => {
+    const source = createOrgModeSource(
+      {
+        primary: clientFor(() => Promise.reject(new Error("primary unavailable"))),
+        replica: clientFor(() =>
+          Promise.resolve({ featureFlags: { snapshotStoreOrgMode: "redis-read" } })
+        ),
+      },
+      { primaryInvalidateRetryDelaysMs: [1, 1] }
+    );
+
+    source.invalidate("org_1");
+
+    // Once the retries exhaust, primaryPending clears and a refresh can repopulate from the replica,
+    // rather than the org staying wedged on the global position forever.
+    await vi.waitFor(() => {
+      source.refresh("org_1");
+      expect(source.get("org_1")).toBe("redis-read");
+    });
   });
 });