Route sessions replication via ClickhouseFactory, gate replication services on factory readiness, and tighten review-comment fixes

- SessionsReplicationService now takes ClickhouseFactory; #flushBatch buckets per
  org and routes each bucket to the org's ClickHouse (mirrors runs replication).
- sessionsReplicationInstance gates service.start() on clickhouseFactory.isReady()
  and owns SIGTERM/SIGINT shutdown — collapsed entry.server.tsx's 35-line block to
  a single bootstrapping reference. Prevents misrouting org-scoped writes to the
  default cluster while the registry is still loading.
- Factory: added sessions_replication ClientType + default singleton (parallels
  runs replication), converted the events client to a singleton (was building two
  pools to the same URL for v1/v2 repos), added missing startTimeMaxAgeMs to the
  clickhouse_v2 case.
- Converted non-hot-path callers (runEngineHandlers, services, route handlers)
  from sync resolveEventRepositoryForStore to async getEventRepositoryForStore so
  they await registry readiness too. Sync now reserved for OTEL exporter and the
  replication services, which gate startup on isReady.
- resolveEventRepositoryForStore is now non-exported with a JSDoc note explaining
  when sync-bypass-with-isReady-gating is justified.
- Minor review fixes: tryCatch from @trigger.dev/core/utils subpath, safeParse +
  400 response in admin.data-stores add/update branches, otlpExporter await moved
  past the content-type check so the unsupported branch short-circuits.
- TestReplicationClickhouseFactory routes both replication and sessions_replication
  to the test client; sessions replication tests updated.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: matt-aitken

apps/webapp/app/entry.server.tsx

@@ -24,44 +24,12 @@ import {
   registerRunEngineEventBusHandlers,
   setupBatchQueueCallbacks,
 } from "./v3/runEngineHandlers.server";
+// Touch the sessions replication singleton at entry so it boots deterministically
+// on webapp startup. The singleton's initializer wires start (gated on
+// `clickhouseFactory.isReady()`) and SIGTERM/SIGINT shutdown — mirrors
+// runsReplicationInstance.
 import { sessionsReplicationInstance } from "./services/sessionsReplicationInstance.server";
-import { signalsEmitter } from "./services/signals.server";
-
-// Start the sessions replication service (subscribes to the logical replication
-// slot, runs leader election, flushes to ClickHouse). Done at entry level so it
-// runs deterministically on webapp boot rather than lazily via a singleton
-// reference elsewhere in the module graph.
-if (sessionsReplicationInstance && env.SESSION_REPLICATION_ENABLED === "1") {
-  // Capture a non-nullable reference so the shutdown closure below
-  // doesn't need to re-null-check (TS narrowing doesn't follow through
-  // an inner function scope).
-  const replicator = sessionsReplicationInstance;
-  replicator
-    .start()
-    .then(() => {
-      console.log("🗃️ Sessions replication service started");
-    })
-    .catch((error) => {
-      console.error("🗃️ Sessions replication service failed to start", {
-        error,
-      });
-    });
-
-  // Wrap the async shutdown in a sync handler that catches rejections —
-  // SIGTERM/SIGINT fire during process teardown, and an unhandled
-  // promise rejection from `_replicationClient.stop()` there would
-  // bubble up past the process exit. Matches the pattern in
-  // dynamicFlushScheduler.server.ts.
-  const shutdownSessionsReplication = () => {
-    replicator.shutdown().catch((error) => {
-      console.error("🗃️ Sessions replication service shutdown error", {
-        error,
-      });
-    });
-  };
-  signalsEmitter.on("SIGTERM", shutdownSessionsReplication);
-  signalsEmitter.on("SIGINT", shutdownSessionsReplication);
-}
+void sessionsReplicationInstance;
 
 const ABORT_DELAY = 30000;
 

apps/webapp/app/routes/admin.data-stores.tsx

@@ -28,7 +28,7 @@ import { prisma } from "~/db.server";
 import { requireUser } from "~/services/session.server";
 import { ClickhouseConnectionSchema } from "~/services/clickhouse/clickhouseSecretSchemas.server";
 import { organizationDataStoresRegistry } from "~/services/dataStores/organizationDataStoresRegistryInstance.server";
-import { tryCatch } from "@trigger.dev/core";
+import { tryCatch } from "@trigger.dev/core/utils";
 
 // ---------------------------------------------------------------------------
 // Loader
@@ -93,14 +93,20 @@ export async function action({ request }: ActionFunctionArgs) {
         .map((s) => s.trim())
         .filter(Boolean);
 
-      const config = ClickhouseConnectionSchema.parse({ url: connectionUrl });
+      const parsedConfig = ClickhouseConnectionSchema.safeParse({ url: connectionUrl });
+      if (!parsedConfig.success) {
+        return typedjson(
+          { error: parsedConfig.error.issues.map((i) => i.message).join(", ") },
+          { status: 400 }
+        );
+      }
 
       const [error, _] = await tryCatch(
         organizationDataStoresRegistry.addDataStore({
           key,
           kind: "CLICKHOUSE",
           organizationIds,
-          config,
+          config: parsedConfig.data,
         })
       );
 
@@ -117,9 +123,17 @@ export async function action({ request }: ActionFunctionArgs) {
         .map((s) => s.trim())
         .filter(Boolean);
 
-      const config = connectionUrl
-        ? ClickhouseConnectionSchema.parse({ url: connectionUrl })
-        : undefined;
+      let config: ReturnType<typeof ClickhouseConnectionSchema.parse> | undefined;
+      if (connectionUrl) {
+        const parsedConfig = ClickhouseConnectionSchema.safeParse({ url: connectionUrl });
+        if (!parsedConfig.success) {
+          return typedjson(
+            { error: parsedConfig.error.issues.map((i) => i.message).join(", ") },
+            { status: 400 }
+          );
+        }
+        config = parsedConfig.data;
+      }
 
       const [error, _] = await tryCatch(
         organizationDataStoresRegistry.updateDataStore({

apps/webapp/app/routes/api.v1.runs.$runId.events.ts

@@ -6,7 +6,7 @@ import {
   createLoaderApiRoute,
 } from "~/services/routeBuilders/apiBuilder.server";
 import { ApiRetrieveRunPresenter } from "~/presenters/v3/ApiRetrieveRunPresenter.server";
-import { resolveEventRepositoryForStore } from "~/v3/eventRepository/index.server";
+import { getEventRepositoryForStore } from "~/v3/eventRepository/index.server";
 
 const ParamsSchema = z.object({
   runId: z.string(), // This is the run friendly ID
@@ -38,7 +38,7 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ resource: run, authentication }) => {
-    const eventRepository = resolveEventRepositoryForStore(
+    const eventRepository = await getEventRepositoryForStore(
       run.taskEventStore,
       authentication.environment.organization.id
     );

apps/webapp/app/routes/api.v1.runs.$runId.spans.$spanId.ts

@@ -7,7 +7,7 @@ import {
   anyResource,
   createLoaderApiRoute,
 } from "~/services/routeBuilders/apiBuilder.server";
-import { resolveEventRepositoryForStore } from "~/v3/eventRepository/index.server";
+import { getEventRepositoryForStore } from "~/v3/eventRepository/index.server";
 import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";
 
 const ParamsSchema = z.object({
@@ -45,7 +45,7 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ params, resource: run, authentication }) => {
-    const eventRepository = resolveEventRepositoryForStore(
+    const eventRepository = await getEventRepositoryForStore(
       run.taskEventStore,
       authentication.environment.organization.id
     );

apps/webapp/app/routes/api.v1.runs.$runId.trace.ts

@@ -6,7 +6,7 @@ import {
   anyResource,
   createLoaderApiRoute,
 } from "~/services/routeBuilders/apiBuilder.server";
-import { resolveEventRepositoryForStore } from "~/v3/eventRepository/index.server";
+import { getEventRepositoryForStore } from "~/v3/eventRepository/index.server";
 import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";
 
 const ParamsSchema = z.object({
@@ -43,7 +43,7 @@ export const loader = createLoaderApiRoute(
     },
   },
   async ({ resource: run, authentication }) => {
-    const eventRepository = resolveEventRepositoryForStore(
+    const eventRepository = await getEventRepositoryForStore(
       run.taskEventStore,
       authentication.environment.organization.id
     );

apps/webapp/app/routes/otel.v1.metrics.ts

@@ -7,10 +7,10 @@ import { otlpExporter } from "~/v3/otlpExporter.server";
 
 export async function action({ request }: ActionFunctionArgs) {
   try {
-    const exporter = await otlpExporter;
     const contentType = request.headers.get("content-type")?.toLowerCase() ?? "";
 
     if (contentType.startsWith("application/json")) {
+      const exporter = await otlpExporter;
       const body = await request.json();
 
       const exportResponse = await exporter.exportMetrics(
@@ -19,6 +19,7 @@ export async function action({ request }: ActionFunctionArgs) {
 
       return json(exportResponse, { status: 200 });
     } else if (contentType.startsWith("application/x-protobuf")) {
+      const exporter = await otlpExporter;
       const buffer = await request.arrayBuffer();
 
       const exportRequest = ExportMetricsServiceRequest.decode(new Uint8Array(buffer));

apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts

@@ -8,7 +8,7 @@ import { Readable } from "stream";
 import { formatDurationMilliseconds } from "@trigger.dev/core/v3/utils/durations";
 import { getTaskEventStoreTableForRun } from "~/v3/taskEventStore.server";
 import { TaskEventKind } from "@trigger.dev/database";
-import { resolveEventRepositoryForStore } from "~/v3/eventRepository/index.server";
+import { getEventRepositoryForStore } from "~/v3/eventRepository/index.server";
 
 export async function loader({ params, request }: LoaderFunctionArgs) {
   const user = await requireUser(request);
@@ -33,7 +33,10 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
     return new Response("Not found", { status: 404 });
   }
 
-  const eventRepository = resolveEventRepositoryForStore(run.taskEventStore, run.organizationId);
+  const eventRepository = await getEventRepositoryForStore(
+    run.taskEventStore,
+    run.organizationId
+  );
 
   const runEvents = await eventRepository.getRunEvents(
     getTaskEventStoreTableForRun(run),

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

@@ -153,6 +153,63 @@ function initializeRunsReplicationClickhouseClient(): ClickHouse {
   });
 }
 
+/** Session replication to ClickHouse (`SESSION_REPLICATION_CLICKHOUSE_URL`); not exported. */
+const defaultSessionsReplicationClickhouseClient = singleton(
+  "sessionsReplicationClickhouseClient",
+  initializeSessionsReplicationClickhouseClient
+);
+
+function initializeSessionsReplicationClickhouseClient(): ClickHouse {
+  if (!env.SESSION_REPLICATION_CLICKHOUSE_URL) {
+    // Sessions replication worker gates on this URL; factory may still resolve "sessions_replication" for tests.
+    return defaultClickhouseClient;
+  }
+
+  const url = new URL(env.SESSION_REPLICATION_CLICKHOUSE_URL);
+  url.searchParams.delete("secure");
+
+  return new ClickHouse({
+    url: url.toString(),
+    name: "sessions-replication",
+    keepAlive: {
+      enabled: env.SESSION_REPLICATION_KEEP_ALIVE_ENABLED === "1",
+      idleSocketTtl: env.SESSION_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
+    },
+    logLevel: env.SESSION_REPLICATION_CLICKHOUSE_LOG_LEVEL,
+    compression: { request: true },
+    maxOpenConnections: env.SESSION_REPLICATION_MAX_OPEN_CONNECTIONS,
+  });
+}
+
+/** Task events (`EVENTS_CLICKHOUSE_URL`); not exported — accessed via factory. */
+const defaultEventsClickhouseClient = singleton(
+  "eventsClickhouseClient",
+  initializeEventsClickhouseClient
+);
+
+function initializeEventsClickhouseClient(): ClickHouse {
+  if (!env.EVENTS_CLICKHOUSE_URL) {
+    throw new Error("EVENTS_CLICKHOUSE_URL is not set");
+  }
+
+  const url = new URL(env.EVENTS_CLICKHOUSE_URL);
+  url.searchParams.delete("secure");
+
+  return new ClickHouse({
+    url: url.toString(),
+    name: "task-events",
+    keepAlive: {
+      enabled: env.EVENTS_CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
+      idleSocketTtl: env.EVENTS_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
+    },
+    logLevel: env.EVENTS_CLICKHOUSE_LOG_LEVEL,
+    compression: {
+      request: env.EVENTS_CLICKHOUSE_COMPRESSION_REQUEST === "1",
+    },
+    maxOpenConnections: env.EVENTS_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
+  });
+}
+
 // ---------------------------------------------------------------------------
 // Helpers
 // ---------------------------------------------------------------------------
@@ -162,7 +219,14 @@ function hashHostname(url: string): string {
   return createHash("sha256").update(parsed.hostname).digest("hex");
 }
 
-export type ClientType = "standard" | "events" | "replication" | "logs" | "query" | "admin";
+export type ClientType =
+  | "standard"
+  | "events"
+  | "replication"
+  | "sessions_replication"
+  | "logs"
+  | "query"
+  | "admin";
 
 function buildOrgClickhouseClient(url: string, clientType: ClientType): ClickHouse {
   const parsed = new URL(url);
@@ -196,6 +260,18 @@ function buildOrgClickhouseClient(url: string, clientType: ClientType): ClickHou
         compression: { request: true },
         maxOpenConnections: env.RUN_REPLICATION_MAX_OPEN_CONNECTIONS,
       });
+    case "sessions_replication":
+      return new ClickHouse({
+        url: parsed.toString(),
+        name,
+        keepAlive: {
+          enabled: env.SESSION_REPLICATION_KEEP_ALIVE_ENABLED === "1",
+          idleSocketTtl: env.SESSION_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
+        },
+        logLevel: env.SESSION_REPLICATION_CLICKHOUSE_LOG_LEVEL,
+        compression: { request: true },
+        maxOpenConnections: env.SESSION_REPLICATION_MAX_OPEN_CONNECTIONS,
+      });
     case "logs":
       return new ClickHouse({
         url: parsed.toString(),
@@ -266,6 +342,8 @@ export class ClickhouseFactory {
           return defaultClickhouseClient;
         case "replication":
           return defaultRunsReplicationClickhouseClient;
+        case "sessions_replication":
+          return defaultSessionsReplicationClickhouseClient;
         case "logs":
           return defaultLogsClickhouseClient;
         case "query":
@@ -350,26 +428,7 @@ export function getDefaultLogsClickhouseClient(): ClickHouse {
 // ---------------------------------------------------------------------------
 
 function getEventsClickhouseClient(): ClickHouse {
-  if (!env.EVENTS_CLICKHOUSE_URL) {
-    throw new Error("EVENTS_CLICKHOUSE_URL is not set");
-  }
-
-  const url = new URL(env.EVENTS_CLICKHOUSE_URL);
-  url.searchParams.delete("secure");
-
-  return new ClickHouse({
-    url: url.toString(),
-    name: "task-events",
-    keepAlive: {
-      enabled: env.EVENTS_CLICKHOUSE_KEEP_ALIVE_ENABLED === "1",
-      idleSocketTtl: env.EVENTS_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS,
-    },
-    logLevel: env.EVENTS_CLICKHOUSE_LOG_LEVEL,
-    compression: {
-      request: env.EVENTS_CLICKHOUSE_COMPRESSION_REQUEST === "1",
-    },
-    maxOpenConnections: env.EVENTS_CLICKHOUSE_MAX_OPEN_CONNECTIONS,
-  });
+  return defaultEventsClickhouseClient;
 }
 
 function buildEventRepository(store: string, clickhouse: ClickHouse): ClickhouseEventRepository {
@@ -411,6 +470,7 @@ function buildEventRepository(store: string, clickhouse: ClickHouse): Clickhouse
         waitForAsyncInsert: env.EVENTS_CLICKHOUSE_WAIT_FOR_ASYNC_INSERT === "1",
         asyncInsertMaxDataSize: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_MAX_DATA_SIZE,
         asyncInsertBusyTimeoutMs: env.EVENTS_CLICKHOUSE_ASYNC_INSERT_BUSY_TIMEOUT_MS,
+        startTimeMaxAgeMs: env.EVENTS_CLICKHOUSE_START_TIME_MAX_AGE_MS,
         llmMetricsBatchSize: env.LLM_METRICS_BATCH_SIZE,
         llmMetricsFlushInterval: env.LLM_METRICS_FLUSH_INTERVAL_MS,
         llmMetricsMaxBatchSize: env.LLM_METRICS_MAX_BATCH_SIZE,