feat(webapp): gate v2 minting on ClickHouse publication readiness; scope reads on v2-may-exist

#1 publication-readiness interlock: a v2 run minted before task_run_v2 is in the
ClickHouse replication publication is permanently absent from ClickHouse
(Postgres only decodes changes for transactions that begin after ALTER
PUBLICATION ADD TABLE, which the replication leader runs at its own startup, not
via a migration), and the run list/metrics/tags are ClickHouse-only. Add a
cached, periodically-refreshed status (runTableV2Status.server.ts) and gate
minting (triggerTask, triggerFailedTask) through canMintV2Run = org cut over AND
table published. Minting fails safe to legacy until the publication carries the
table and self-heals once it does, removing the manual pg_publication_tables
enable step.

#2 native-rollback read scope: cross-table read scoping (idempotency dedup,
ApiRetrieveRun hierarchy) keyed on the native master switch alone, so disabling
native realtime after v2 runs exist re-scoped reads to legacy and hid existing
v2 runs (an idempotency dedup miss means duplicate execution). Scope on
v2RunsMayExist (native on OR task_run_v2 has rows) instead; it is monotonic, so
the read scope cannot regress once v2 runs exist.

Author: d-cs

apps/webapp/app/presenters/v3/ApiRetrieveRunPresenter.server.ts

@@ -24,6 +24,7 @@ import {
 import { generatePresignedUrl } from "~/v3/objectStore.server";
 import { runStore } from "~/v3/runStore.server";
 import { hydrateParentAndRoot, hydrateChildRuns } from "~/v3/runHierarchy.server";
+import { v2RunsMayExist } from "~/v3/runTableV2Status.server";
 import { env as serverEnv } from "~/env.server";
 import { tracer } from "~/v3/tracer.server";
 import { startSpanWithEnv } from "~/v3/tracing.server";
@@ -148,16 +149,19 @@ export class ApiRetrieveRunPresenter {
       // legacy run's v2 children), which arise in the mixed window, would come
       // back null/empty. Resolve parent/root by id (RunStore routes by format)
       // and children by a both-table predicate.
-      // Scope the cross-table reads on whether ANY v2 run can exist in this
-      // deployment (the native master switch), NOT the org's current flag: a
-      // run's table is fixed by its id format, and an org that was on v2 then
-      // flipped off still HAS v2 runs (and v2 children) that runTableV2.server.ts
-      // documents as staying readable. pgRow is routed here by id format, so it
-      // can be a v2 run for a now-non-v2 org; scoping to "legacy" off the per-org
-      // flag would then silently drop its v2 children/parent. Until native is
-      // enabled no v2 run exists yet (minting requires it), so "legacy" is safe
-      // and skips the empty task_run_v2 query. The reads also run in parallel.
-      const tables = serverEnv.REALTIME_BACKEND_NATIVE_ENABLED === "1" ? "both" : "legacy";
+      // Scope the cross-table reads on whether a v2 run could exist at all, NOT
+      // the org's current flag: a run's table is fixed by its id format, and an
+      // org that was on v2 then flipped off still HAS v2 runs (and v2 children)
+      // that stay readable. pgRow is routed here by id format, so it can be a v2
+      // run for a now-non-v2 org; scoping to "legacy" would then silently drop
+      // its v2 children/parent. v2RunsMayExist is monotonic (native on now, OR
+      // task_run_v2 already has rows), so turning the native master switch off
+      // does not re-scope to legacy and hide existing v2 runs. While no v2 run
+      // has ever existed it stays "legacy" and skips the empty task_run_v2 query.
+      // The reads also run in parallel.
+      const tables = v2RunsMayExist(serverEnv.REALTIME_BACKEND_NATIVE_ENABLED === "1")
+        ? "both"
+        : "legacy";
       const [{ parentTaskRun, rootTaskRun }, childRuns] = await Promise.all([
         hydrateParentAndRoot(
           { parentTaskRunId: pgRow.parentTaskRunId, rootTaskRunId: pgRow.rootTaskRunId },

apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts

@@ -12,6 +12,7 @@ import { claimOrAwait } from "~/v3/mollifier/idempotencyClaim.server";
 import { makeResolveMollifierFlag } from "~/v3/mollifier/mollifierGate.server";
 import { runStore } from "~/v3/runStore.server";
 import { shouldUseV2RunTable } from "~/v3/runTableV2.server";
+import { v2RunsMayExist } from "~/v3/runTableV2Status.server";
 import type { TraceEventConcern, TriggerTaskRequest } from "../types";
 
 // In-memory per-org mollifier-enabled check, shared with `evaluateGate`
@@ -313,16 +314,16 @@ export class IdempotencyKeyConcern {
       nativeRealtimeEnabled: env.REALTIME_BACKEND_NATIVE_ENABLED === "1",
     });
 
-    // Scope the idempotency dedup read on whether ANY v2 run can exist in this
-    // deployment (the native master switch), NOT on whether this org currently
-    // mints v2. A run's table is fixed by its id format, so an org that was on
-    // v2 then flipped off still holds v2 runs an idempotency key can match
-    // (runTableV2.server.ts documents they stay readable); gating the read on
-    // orgUsesV2 would miss them and let a duplicate through. Until native is
-    // enabled no v2 run exists yet (minting requires it), so "legacy" is safe and
-    // skips the empty task_run_v2 query on the trigger hot path; once native is
-    // on, read both.
-    const anyV2RunsPossible = env.REALTIME_BACKEND_NATIVE_ENABLED === "1";
+    // Scope the idempotency dedup read on whether a v2 run could exist at all,
+    // NOT on whether this org currently mints v2. A run's table is fixed by its
+    // id format, so an org that was on v2 then flipped off still holds v2 runs an
+    // idempotency key can match; gating the read on orgUsesV2 would miss them and
+    // let a duplicate through. v2RunsMayExist is monotonic (native on now, OR
+    // task_run_v2 already has rows), so turning the native master switch off
+    // after v2 runs exist does NOT re-scope the read back to legacy and hide
+    // them. While no v2 run has ever existed it stays "legacy" and skips the
+    // empty task_run_v2 query on the trigger hot path.
+    const anyV2RunsPossible = v2RunsMayExist(env.REALTIME_BACKEND_NATIVE_ENABLED === "1");
 
     const existingRun = idempotencyKey
       ? await runStore.findRun(

apps/webapp/app/runEngine/services/triggerFailedTask.server.ts

@@ -10,7 +10,7 @@ import { PerformTaskRunAlertsService } from "~/v3/services/alerts/performTaskRun
 import { DefaultQueueManager } from "../concerns/queues.server";
 import type { TriggerTaskRequest } from "../types";
 import { runStore } from "~/v3/runStore.server";
-import { shouldUseV2RunTable } from "~/v3/runTableV2.server";
+import { canMintV2Run } from "~/v3/runTableV2Status.server";
 import { env } from "~/env.server";
 
 export type TriggerFailedTaskRequest = {
@@ -76,7 +76,7 @@ export class TriggerFailedTaskService {
     // batch, create an ongoing cross-table edge on the failure path. Mirrors the
     // mint gate in triggerTask.server.ts.
     const failedRunFriendlyId = (
-      shouldUseV2RunTable(request.environment.organization.featureFlags, {
+      canMintV2Run(request.environment.organization.featureFlags, {
         nativeRealtimeEnabled: env.REALTIME_BACKEND_NATIVE_ENABLED === "1",
       })
         ? RunId.generateKsuid()
@@ -292,10 +292,9 @@ export class TriggerFailedTaskService {
         where: { id: opts.organizationId },
         select: { featureFlags: true },
       });
-      useV2RunTable = shouldUseV2RunTable(
-        (org?.featureFlags as Record<string, unknown>) ?? null,
-        { nativeRealtimeEnabled: env.REALTIME_BACKEND_NATIVE_ENABLED === "1" }
-      );
+      useV2RunTable = canMintV2Run((org?.featureFlags as Record<string, unknown>) ?? null, {
+        nativeRealtimeEnabled: env.REALTIME_BACKEND_NATIVE_ENABLED === "1",
+      });
     } catch {
       // Leave useV2RunTable=false (legacy id).
     }

apps/webapp/app/runEngine/services/triggerTask.server.ts

@@ -25,7 +25,7 @@ import { logger } from "~/services/logger.server";
 import { parseDelay } from "~/utils/delays";
 import { handleMetadataPacket } from "~/utils/packets";
 import { startSpan } from "~/v3/tracing.server";
-import { shouldUseV2RunTable } from "~/v3/runTableV2.server";
+import { canMintV2Run } from "~/v3/runTableV2Status.server";
 import type {
   TriggerTaskServiceOptions,
   TriggerTaskServiceResult,
@@ -159,7 +159,7 @@ export class RunEngineTriggerTaskService {
           // trigger hot path. Downstream routing is by id format only.
           const runFriendlyId =
             options?.runFriendlyId ??
-            (shouldUseV2RunTable(environment.organization.featureFlags, {
+            (canMintV2Run(environment.organization.featureFlags, {
               nativeRealtimeEnabled: env.REALTIME_BACKEND_NATIVE_ENABLED === "1",
             })
               ? RunId.generateKsuid()

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

@@ -0,0 +1,104 @@
+import { prisma } from "~/db.server";
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+import { singleton } from "~/utils/singleton";
+import { shouldUseV2RunTable, type ShouldUseV2RunTableOptions } from "~/v3/runTableV2.server";
+
+/**
+ * Cached, periodically-refreshed facts about the `task_run_v2` table, read OFF
+ * the trigger hot path (no per-request DB query) to gate v2 minting and
+ * cross-table read scoping.
+ */
+type RunTableV2Status = {
+  /**
+   * Is `task_run_v2` in the ClickHouse logical-replication publication?
+   *
+   * Postgres only decodes a table's changes for transactions that BEGIN after
+   * the decoder sees `ALTER PUBLICATION ... ADD TABLE`, and that ADD TABLE is run
+   * lazily by the replication leader on its own startup, NOT by a migration. So a
+   * v2 run minted before the table is published is permanently absent from
+   * ClickHouse with no backfill, and the run list / metrics / tags / bulk actions
+   * are ClickHouse-only. Mint v2 ONLY when this is true; otherwise mint legacy
+   * (fail-safe), self-healing once the leader publishes the table.
+   */
+  published: boolean;
+  /**
+   * Has any v2 run ever existed (monotonic in practice)? Cross-table READ scoping
+   * uses this (OR the native master switch) rather than the master switch alone,
+   * so disabling native realtime cannot re-scope reads back to legacy and hide
+   * already-minted v2 runs from idempotency dedup and hierarchy reads.
+   */
+  hasRows: boolean;
+};
+
+const REFRESH_INTERVAL_MS = 30_000;
+
+const status = singleton("runTableV2Status", initialize);
+
+function initialize(): RunTableV2Status {
+  const state: RunTableV2Status = { published: false, hasRows: false };
+
+  // The publication only exists when runs replication is configured. Without it
+  // no v2 run can be captured by ClickHouse, so leave published=false: minting
+  // stays on legacy regardless of org flags.
+  if (!env.RUN_REPLICATION_CLICKHOUSE_URL) {
+    return state;
+  }
+
+  const refresh = async () => {
+    try {
+      const published = await prisma.$queryRaw<Array<{ present: boolean }>>`
+        SELECT EXISTS (
+          SELECT 1 FROM pg_publication_tables
+          WHERE pubname = ${env.RUN_REPLICATION_PUBLICATION_NAME}
+            AND tablename = 'task_run_v2'
+        ) AS present`;
+      state.published = published[0]?.present ?? false;
+
+      // hasRows is monotonic; once true, stop probing.
+      if (!state.hasRows) {
+        const hasRows = await prisma.$queryRaw<Array<{ present: boolean }>>`
+          SELECT EXISTS (SELECT 1 FROM task_run_v2 LIMIT 1) AS present`;
+        state.hasRows = hasRows[0]?.present ?? false;
+      }
+    } catch (error) {
+      logger.warn("runTableV2Status refresh failed; keeping last-known status", {
+        error: error instanceof Error ? error.message : String(error),
+      });
+    }
+  };
+
+  void refresh();
+  const timer = setInterval(() => void refresh(), REFRESH_INTERVAL_MS);
+  timer.unref?.();
+
+  return state;
+}
+
+/** `task_run_v2` is in the ClickHouse replication publication (cached, off the hot path). */
+export function isV2RunTablePublished(): boolean {
+  return status.published;
+}
+
+/**
+ * Whether a v2 run could be relevant to a cross-table READ: native realtime is on
+ * (v2 is being minted now) OR `task_run_v2` already holds rows. Scope cross-table
+ * reads on this, not the native master switch alone, so turning native off cannot
+ * hide already-minted v2 runs.
+ */
+export function v2RunsMayExist(nativeRealtimeEnabled: boolean): boolean {
+  return nativeRealtimeEnabled || status.hasRows;
+}
+
+/**
+ * Mint gate: mint a v2 (KSUID) run only when the org is cut over to v2 AND
+ * `task_run_v2` is in the ClickHouse publication, so a v2 run can never be
+ * silently lost from ClickHouse by being minted before the replication leader
+ * publishes the table. Fails safe to legacy until then; self-heals once published.
+ */
+export function canMintV2Run(
+  orgFeatureFlags: unknown,
+  options: ShouldUseV2RunTableOptions
+): boolean {
+  return shouldUseV2RunTable(orgFeatureFlags, options) && isV2RunTablePublished();
+}

apps/webapp/test/runTableV2Status.test.ts

@@ -0,0 +1,53 @@
+import { describe, expect, it } from "vitest";
+import { canMintV2Run, v2RunsMayExist } from "~/v3/runTableV2Status.server";
+
+// The module caches its status in a globalThis singleton ("runTableV2Status").
+// In the unit-test env runs replication is unconfigured, so it initializes to
+// { published:false, hasRows:false } with no background poller. Mutate that
+// cached object to exercise the gates deterministically.
+function setStatus(published: boolean, hasRows: boolean) {
+  const singletons = (globalThis as any).__trigger_singletons;
+  // Force module init (the singleton is created on first getter call/import).
+  v2RunsMayExist(false);
+  singletons.runTableV2Status.published = published;
+  singletons.runTableV2Status.hasRows = hasRows;
+}
+
+const CUTOVER_FLAGS = { realtimeBackend: "native", runTableV2: true };
+
+describe("canMintV2Run (mint gate: org cut over AND task_run_v2 published)", () => {
+  it("mints v2 only when the org is cut over AND the table is published", () => {
+    setStatus(true, true);
+    expect(canMintV2Run(CUTOVER_FLAGS, { nativeRealtimeEnabled: true })).toBe(true);
+  });
+
+  it("fails safe to legacy when the org is cut over but the table is NOT published", () => {
+    setStatus(false, true);
+    expect(canMintV2Run(CUTOVER_FLAGS, { nativeRealtimeEnabled: true })).toBe(false);
+  });
+
+  it("stays legacy when the org is not cut over, even if published", () => {
+    setStatus(true, true);
+    expect(
+      canMintV2Run({ realtimeBackend: "electric", runTableV2: false }, { nativeRealtimeEnabled: true })
+    ).toBe(false);
+    expect(canMintV2Run(CUTOVER_FLAGS, { nativeRealtimeEnabled: false })).toBe(false);
+  });
+});
+
+describe("v2RunsMayExist (read scope: native on OR table has rows)", () => {
+  it("is true when native realtime is on (v2 being minted now)", () => {
+    setStatus(false, false);
+    expect(v2RunsMayExist(true)).toBe(true);
+  });
+
+  it("is true when task_run_v2 already has rows even with native OFF (rollback safety)", () => {
+    setStatus(false, true);
+    expect(v2RunsMayExist(false)).toBe(true);
+  });
+
+  it("is false only when native is off AND no v2 run has ever existed", () => {
+    setStatus(false, false);
+    expect(v2RunsMayExist(false)).toBe(false);
+  });
+});