feat(webapp): route the legacy v1 event store to v2

Runs on the legacy `clickhouse` event store now write and read `clickhouse_v2`, and new runs are stored as `clickhouse_v2`. Self-hosters on the legacy store stop seeing events recorded before the upgrade. Set `EVENT_REPOSITORY_V1_REROUTE_ENABLED=0` to keep the legacy store.

Mono-RevId: caf2efe61a9990d53f411ea869a58d3db3abfd35

Author: Mariefay

.server-changes/route-v1-event-store-to-v2.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: breaking
+---
+
+Self-hosted instances on the legacy `clickhouse` event store now use `clickhouse_v2`, so run logs and traces recorded before the upgrade no longer show. Set `EVENT_REPOSITORY_DEFAULT_STORE=clickhouse_v2`, or set `EVENT_REPOSITORY_V1_REROUTE_ENABLED=0` to stay on the legacy store.

apps/webapp/app/env.server.ts

@@ -2632,6 +2632,8 @@ const EnvironmentSchema = z
     EVENT_REPOSITORY_DEFAULT_STORE: z
       .enum(["postgres", "clickhouse", "clickhouse_v2"])
       .default("postgres"),
+    // Route the legacy "clickhouse" store (task_events_v1) to clickhouse_v2. Set to 0 to keep v1.
+    EVENT_REPOSITORY_V1_REROUTE_ENABLED: BoolEnvWithDefault(true),
     EVENT_REPOSITORY_DEBUG_LOGS_DISABLED: BoolEnv.default(false),
     EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED: BoolEnv.default(false),
     EVENTS_CLICKHOUSE_MAX_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(25_000),

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

@@ -210,7 +210,11 @@ export class RunPresenter {
     // Read before the first chunk so the first tail covers the background load.
     const firstChunkReadAt = Date.now();
 
-    const firstChunkPromise = canLiveTail(run.taskEventStore, env.TRACE_VIEW_EMERGENCY_SPAN_CAP)
+    const firstChunkPromise = canLiveTail(
+      run.taskEventStore,
+      env.TRACE_VIEW_EMERGENCY_SPAN_CAP,
+      env.EVENT_REPOSITORY_V1_REROUTE_ENABLED
+    )
       ? repository.getTraceChunk(
           getTaskEventStoreTableForRun(run),
           environment.id,

apps/webapp/app/presenters/v3/liveTailGate.ts

@@ -1,8 +1,17 @@
+import { effectiveEventStore } from "~/v3/eventRepository/effectiveEventStore";
+
 // Progressive loading only runs with the live tail (without it every stream ping would
 // reset and re-download the tree), and only the v2 store has the write time it pages by.
 // The emergency span cap turns both off, so new loads take the summary path that honours it.
-export function canLiveTail(taskEventStore: string, emergencySpanCap: number | undefined): boolean {
-  return taskEventStore === "clickhouse_v2" && emergencySpanCap === undefined;
+export function canLiveTail(
+  taskEventStore: string,
+  emergencySpanCap: number | undefined,
+  v1RerouteEnabled: boolean
+): boolean {
+  return (
+    effectiveEventStore(taskEventStore, v1RerouteEnabled) === "clickhouse_v2" &&
+    emergencySpanCap === undefined
+  );
 }
 
 export function hasWriteTimes(events: { insertedAt?: string }[]): boolean {

apps/webapp/app/services/clickhouse/clickhouseFactory.server.ts

@@ -10,6 +10,7 @@ import { clampToEmergencySpanCap } from "~/v3/eventRepository/emergencySpanCap.s
 import { singleton } from "~/utils/singleton";
 import type { OrganizationDataStoresRegistry } from "~/services/dataStores/organizationDataStoresRegistry.server";
 import { type IEventRepository } from "~/v3/eventRepository/eventRepository.types";
+import { effectiveEventStore } from "~/v3/eventRepository/effectiveEventStore";
 
 // Declared before the singletons below, which initialize clients at module load.
 const LOGS_LIST_REQUEST_TIMEOUT_MARGIN_SECONDS = 5;
@@ -696,7 +697,10 @@ export class ClickhouseFactory {
   /** Event repositories keyed by hostname hash (stateful, must be reused). */
   private readonly _eventRepositoryCache = new Map<string, ClickhouseEventRepository>();
 
-  constructor(private readonly _registry: OrganizationDataStoresRegistry) {}
+  constructor(
+    private readonly _registry: OrganizationDataStoresRegistry,
+    private readonly _v1RerouteEnabled: boolean = env.EVENT_REPOSITORY_V1_REROUTE_ENABLED
+  ) {}
 
   async isReady(): Promise<boolean> {
     if (!this._registry.isLoaded) {
@@ -774,9 +778,11 @@ export class ClickhouseFactory {
   }
 
   getEventRepositoryForOrganizationSync(
-    store: string,
+    requestedStore: string,
     organizationId: string
   ): { key: string; repository: IEventRepository } {
+    // Normalized before the cache key, so "clickhouse" and "clickhouse_v2" share one repository
+    const store = effectiveEventStore(requestedStore, this._v1RerouteEnabled);
     const dataStore = this._registry.get(organizationId, "CLICKHOUSE");
 
     if (!dataStore) {

apps/webapp/app/v3/eventRepository/effectiveEventStore.ts

@@ -0,0 +1,11 @@
+/**
+ * The store a run's events are actually written to and read from. With the reroute on, the legacy
+ * "clickhouse" store (task_events_v1) resolves to "clickhouse_v2"; every other value is unchanged.
+ */
+export function effectiveEventStore(store: string, v1RerouteEnabled: boolean): string {
+  if (v1RerouteEnabled && store === "clickhouse") {
+    return "clickhouse_v2";
+  }
+
+  return store;
+}

apps/webapp/app/v3/eventRepository/index.server.ts

@@ -5,9 +5,9 @@ import { prisma } from "~/db.server";
 import { runStore } from "../runStore.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
 import { logger } from "~/services/logger.server";
-import { FEATURE_FLAG } from "../featureFlags";
 import { flag } from "../featureFlags.server";
 import { getTaskEventStore } from "../taskEventStore.server";
+import { resolveEventStore } from "./resolveEventStore.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 
 export const EVENT_STORE_TYPES = {
@@ -55,13 +55,13 @@ export async function getConfiguredEventRepository(
     throw new Error("Organization not found when configuring event repository");
   }
 
-  // resolveTaskEventRepositoryFlag checks:
-  // 1. organization.featureFlags (highest priority)
-  // 2. global feature flags (via flags() function)
-  // 3. env.EVENT_REPOSITORY_DEFAULT_STORE (fallback)
-  const taskEventStore = await resolveTaskEventRepositoryFlag(
-    (organization.featureFlags as Record<string, unknown> | null) ?? undefined
-  );
+  // Org override, then the global flag, then EVENT_REPOSITORY_DEFAULT_STORE
+  const taskEventStore = await resolveEventStore({
+    featureFlags: (organization.featureFlags as Record<string, unknown> | null) ?? undefined,
+    flag,
+    defaultStore: env.EVENT_REPOSITORY_DEFAULT_STORE,
+    v1RerouteEnabled: env.EVENT_REPOSITORY_V1_REROUTE_ENABLED,
+  });
 
   if (taskEventStore === EVENT_STORE_TYPES.CLICKHOUSE_V2) {
     const { repository: resolvedRepository } =
@@ -83,7 +83,14 @@ export async function getEventRepository(
   featureFlags: Record<string, unknown> | undefined,
   parentStore: string | undefined
 ): Promise<{ repository: IEventRepository; store: string }> {
-  const taskEventStore = parentStore ?? (await resolveTaskEventRepositoryFlag(featureFlags));
+  // New runs store the effective value, so the legacy v1 store stops spreading through old trees
+  const taskEventStore = await resolveEventStore({
+    parentStore,
+    featureFlags,
+    flag,
+    defaultStore: env.EVENT_REPOSITORY_DEFAULT_STORE,
+    v1RerouteEnabled: env.EVENT_REPOSITORY_V1_REROUTE_ENABLED,
+  });
 
   // Non-ClickHouse stores (e.g. the "taskEvent" DB default for Postgres-backed
   // runs, or the legacy "postgres" value) resolve to the Prisma event repo.
@@ -110,26 +117,6 @@ export async function getEventRepository(
   }
 }
 
-async function resolveTaskEventRepositoryFlag(
-  featureFlags: Record<string, unknown> | undefined
-): Promise<"clickhouse" | "clickhouse_v2" | "postgres"> {
-  const flagResult = await flag({
-    key: FEATURE_FLAG.taskEventRepository,
-    defaultValue: env.EVENT_REPOSITORY_DEFAULT_STORE,
-    overrides: featureFlags,
-  });
-
-  if (flagResult === "clickhouse_v2") {
-    return "clickhouse_v2";
-  }
-
-  if (flagResult === "clickhouse") {
-    return "clickhouse";
-  }
-
-  return flagResult;
-}
-
 export async function recordRunDebugLog(
   runId: string,
   message: string,

apps/webapp/app/v3/eventRepository/resolveEventStore.server.ts

@@ -0,0 +1,32 @@
+import { FEATURE_FLAG } from "../featureFlags";
+import type { makeFlag } from "../featureFlags.server";
+import { effectiveEventStore } from "./effectiveEventStore";
+
+type ResolveEventStoreOptions = {
+  /** Set for child runs, which inherit their parent's store. */
+  parentStore?: string;
+  /** The organization's flag overrides. */
+  featureFlags: Record<string, unknown> | undefined;
+  flag: ReturnType<typeof makeFlag>;
+  defaultStore: "postgres" | "clickhouse" | "clickhouse_v2";
+  v1RerouteEnabled: boolean;
+};
+
+/** The effective event store for a new run, or for an org's configured store when there's no parent. */
+export async function resolveEventStore({
+  parentStore,
+  featureFlags,
+  flag,
+  defaultStore,
+  v1RerouteEnabled,
+}: ResolveEventStoreOptions): Promise<string> {
+  const store =
+    parentStore ??
+    (await flag({
+      key: FEATURE_FLAG.taskEventRepository,
+      defaultValue: defaultStore,
+      overrides: featureFlags,
+    }));
+
+  return effectiveEventStore(store, v1RerouteEnabled);
+}

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

@@ -57,7 +57,7 @@ type OTLPExporterConfig = {
   meter?: Meter;
 };
 
-class OTLPExporter {
+export class OTLPExporter {
   private _tracer: Tracer;
   private readonly _clickhouseFactory: ClickhouseFactory;
   private readonly _verbose: boolean;

apps/webapp/test/clickhouseFactory.test.ts

@@ -6,6 +6,7 @@ import { postgresTest } from "@internal/testcontainers";
 import { OrganizationDataStoresRegistry } from "~/services/dataStores/organizationDataStoresRegistry.server";
 import { ClickhouseConnectionSchema } from "~/services/clickhouse/clickhouseSecretSchemas.server";
 import { ClickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server";
+import type { ClickhouseEventRepository } from "~/v3/eventRepository/clickhouseEventRepository.server";
 
 vi.setConfig({ testTimeout: 60_000 });
 
@@ -124,4 +125,48 @@ describe("ClickHouse Factory", () => {
       expect(after).toBe(defaultClient);
     }
   );
+  postgresTest(
+    "routes the v1 store to the shared v2 repository when the reroute is on",
+    async ({ prisma }) => {
+      const registry = new OrganizationDataStoresRegistry(prisma, prisma);
+
+      await registry.addDataStore({
+        key: "reroute-store",
+        kind: "CLICKHOUSE",
+        organizationIds: ["org-reroute"],
+        config: ClickhouseConnectionSchema.parse({ url: TEST_URL }),
+      });
+
+      await registry.loadFromDatabase();
+
+      const factory = new ClickhouseFactory(registry, true);
+      const v1 = factory.getEventRepositoryForOrganizationSync("clickhouse", "org-reroute");
+      const v2 = factory.getEventRepositoryForOrganizationSync("clickhouse_v2", "org-reroute");
+
+      expect(v1.key).toBe(v2.key);
+      expect(v1.repository).toBe(v2.repository);
+      expect((v1.repository as ClickhouseEventRepository).version).toBe("v2");
+    }
+  );
+
+  postgresTest("keeps the v1 repository when the reroute is off", async ({ prisma }) => {
+    const registry = new OrganizationDataStoresRegistry(prisma, prisma);
+
+    await registry.addDataStore({
+      key: "no-reroute-store",
+      kind: "CLICKHOUSE",
+      organizationIds: ["org-no-reroute"],
+      config: ClickhouseConnectionSchema.parse({ url: TEST_URL }),
+    });
+
+    await registry.loadFromDatabase();
+
+    const factory = new ClickhouseFactory(registry, false);
+    const v1 = factory.getEventRepositoryForOrganizationSync("clickhouse", "org-no-reroute");
+    const v2 = factory.getEventRepositoryForOrganizationSync("clickhouse_v2", "org-no-reroute");
+
+    expect(v1.key).not.toBe(v2.key);
+    expect((v1.repository as ClickhouseEventRepository).version).toBe("v1");
+    expect((v2.repository as ClickhouseEventRepository).version).toBe("v2");
+  });
 });