perf(testcontainers): speed up Postgres-backed test fixtures

Speeds up the Postgres-backed test fixtures.

- One Postgres container per image is now shared by every test file and
test process (testcontainers reuse). The schema lives in a template
database named by its hash and built once under an advisory lock; each
test still gets its own cloned database. Set
`TESTCONTAINERS_REUSE_ENABLE=false` to get a private container.
- Test Postgres runs with `fsync` and `full_page_writes` off on tmpfs.
On PG14 every `CREATE DATABASE ... TEMPLATE` forces a checkpoint, so
each per-test clone drops from seconds to about 100ms.
- The schema is applied as one SQL script from `prisma migrate diff
--from-empty`, cached on disk by schema and Prisma version, in place of
`prisma db push`.
- `postgresAndRedisTest` and `replicationContainerTest` clone a database
per test from a Postgres private to the test file (logical replication
slots are cluster-wide and tests name them), and drop the clone's slots
before dropping it. The e2e servers (`startTestServer`,
`startSessionStreamTestServer`) clone theirs from the shared Postgres.
- `containerTest` and `containerTestWithIsolatedRedis` no longer boot
ClickHouse. Tests that read it use the new
`containerTestWithClickhouse`.
`containerTestWithIsolatedRedisNoClickhouse` is folded into
`containerTestWithIsolatedRedis`.

Mono-RevId: d3e80a3518642770f0edfb15726b08a36e55686e

Author: ericallam

apps/webapp/test/apiRunListPresenter.test.ts

@@ -46,7 +46,7 @@ vi.mock("~/db.server", async () => {
 });
 
 import { ClickHouse } from "@internal/clickhouse";
-import { containerTest } from "@internal/testcontainers";
+import { containerTestWithClickhouse } from "@internal/testcontainers";
 import { CURRENT_API_VERSION } from "~/api/versions";
 import { ApiRunListPresenter } from "~/presenters/v3/ApiRunListPresenter.server";
 import {
@@ -71,7 +71,7 @@ function setupClients(prisma: unknown, clickhouseUrl: string): ClickHouse {
 }
 
 describe("ApiRunListPresenter public /runs list", () => {
-  containerTest(
+  containerTestWithClickhouse(
     "genuinely-empty env returns { data: [], pagination } without error",
     async ({ clickhouseContainer, prisma }) => {
       setupClients(prisma, clickhouseContainer.getConnectionUrl());
@@ -97,7 +97,7 @@ describe("ApiRunListPresenter public /runs list", () => {
     }
   );
 
-  containerTest(
+  containerTestWithClickhouse(
     "env scoping resolves via the control-plane _replica handle (filter[env], 4th arg omitted)",
     async ({ clickhouseContainer, prisma }) => {
       const clickhouse = setupClients(prisma, clickhouseContainer.getConnectionUrl());
@@ -128,7 +128,7 @@ describe("ApiRunListPresenter public /runs list", () => {
     }
   );
 
-  containerTest(
+  containerTestWithClickhouse(
     "single-DB passthrough: no readThroughDeps lists the seeded runs unchanged",
     async ({ clickhouseContainer, prisma }) => {
       const clickhouse = setupClients(prisma, clickhouseContainer.getConnectionUrl());

apps/webapp/test/clickhouseQueryMetrics.test.ts

@@ -1,5 +1,5 @@
 import { ClickHouse } from "@internal/clickhouse";
-import { containerTest } from "@internal/testcontainers";
+import { containerTestWithClickhouse } from "@internal/testcontainers";
 import { describe, expect, vi } from "vitest";
 import { RunsRepository } from "~/services/runsRepository/runsRepository.server";
 import {
@@ -15,7 +15,7 @@ vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} }));
 vi.setConfig({ testTimeout: 90_000 });
 
 describe("clickhouse query metrics", () => {
-  containerTest(
+  containerTestWithClickhouse(
     "records duration + read_rows on success and an error metric with the ClickHouse error type",
     async ({ clickhouseContainer, prisma }) => {
       const ctx = await seedParents(prisma, "chm");

apps/webapp/test/dashboardAgentRunTrace.test.ts

@@ -4,7 +4,7 @@
 // real Postgres and ClickHouse behind them.
 import { ClickHouse, type TaskEventV2Input } from "@internal/clickhouse";
 import { setupAuthenticatedEnvironment } from "@internal/run-engine/tests";
-import { containerTest } from "@internal/testcontainers";
+import { containerTestWithClickhouse } from "@internal/testcontainers";
 import type { PrismaClient } from "@trigger.dev/database";
 import { describe, expect, vi } from "vitest";
 import {
@@ -413,7 +413,7 @@ function depthOf(rootSpan: DashboardAgentTraceSpan | undefined): number {
 }
 
 describe("the dashboard agent run trace", () => {
-  containerTest(
+  containerTestWithClickhouse(
     "reports the same millisecond durations, task slugs and customer-visible events on both event stores",
     async ({ prisma, clickhouseContainer }) => {
       const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
@@ -506,7 +506,7 @@ describe("the dashboard agent run trace", () => {
     TIMEOUT_MS
   );
 
-  containerTest(
+  containerTestWithClickhouse(
     "keeps a chain well under the depth guard fully intact and JSON-serializable",
     async ({ prisma }) => {
       const { repository, environmentId, organizationId, startedAt } =
@@ -529,7 +529,7 @@ describe("the dashboard agent run trace", () => {
     TIMEOUT_MS
   );
 
-  containerTest(
+  containerTestWithClickhouse(
     "does not truncate a chain of exactly the depth guard",
     async ({ prisma }) => {
       const { repository, environmentId, organizationId, startedAt } =
@@ -557,7 +557,7 @@ describe("the dashboard agent run trace", () => {
     TIMEOUT_MS
   );
 
-  containerTest(
+  containerTestWithClickhouse(
     "truncates a chain deeper than the depth guard instead of throwing",
     async ({ prisma }) => {
       const { repository, environmentId, organizationId, startedAt } =
@@ -586,7 +586,7 @@ describe("the dashboard agent run trace", () => {
     },
     TIMEOUT_MS
   );
-  containerTest(
+  containerTestWithClickhouse(
     "builds a child run's trace when the trace root is outside the child's window",
     async ({ prisma, clickhouseContainer }) => {
       const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");

apps/webapp/test/engine/streamBatchItems.test.ts

@@ -19,7 +19,7 @@ import { PostgresRunStore } from "@internal/run-store";
 import { setupAuthenticatedEnvironment } from "@internal/run-engine/tests";
 // Per-test redis isolation: each test runs its own RunEngine whose background work outlives the test
 // body. NoClickhouse because this suite never touches ClickHouse - skips the worker-scoped boot+migrate.
-import { containerTestWithIsolatedRedisNoClickhouse as containerTest } from "@internal/testcontainers";
+import { containerTestWithIsolatedRedis as containerTest } from "@internal/testcontainers";
 import { trace } from "@opentelemetry/api";
 import type { PrismaClient } from "@trigger.dev/database";
 import { BatchId } from "@trigger.dev/core/v3/isomorphic";

apps/webapp/test/locateAgentObject.test.ts

@@ -8,7 +8,7 @@ import type { LocateResult } from "@internal/dashboard-agent-contracts";
 import type { PrismaClient } from "@trigger.dev/database";
 import { ClickHouse } from "@internal/clickhouse";
 import { RunId } from "@trigger.dev/core/v3/isomorphic";
-import { containerTest } from "@internal/testcontainers";
+import { containerTestWithClickhouse } from "@internal/testcontainers";
 import { setTimeout as sleep } from "node:timers/promises";
 import { describe, expect, vi } from "vitest";
 import {
@@ -202,7 +202,7 @@ function locateTest(
   name: string,
   fn: (ctx: { prisma: PrismaClient; w: AgentWorld }) => Promise<void>
 ) {
-  containerTest(name, async ({ prisma }) => {
+  containerTestWithClickhouse(name, async ({ prisma }) => {
     db.client = prisma;
     await fn({ prisma, w: await seedAgentWorld(prisma) });
   });
@@ -217,7 +217,7 @@ type ErrorCtx = {
 
 /** Same, plus a real `ClickHouse` client wired into the mocked factory for the error kind. */
 function errorLocateTest(name: string, fn: (ctx: ErrorCtx) => Promise<void>) {
-  containerTest(name, async ({ prisma, clickhouseContainer, clickhouseClient }) => {
+  containerTestWithClickhouse(name, async ({ prisma, clickhouseContainer, clickhouseClient }) => {
     db.client = prisma;
     const clickhouse = new ClickHouse({ url: clickhouseContainer.getConnectionUrl() });
     chFactory.getClickhouseForOrganization.mockResolvedValue(clickhouse);

apps/webapp/test/queueArchive.test.ts

@@ -23,7 +23,7 @@ vi.mock("~/services/clickhouse/clickhouseFactoryInstance.server", () => ({
 import { ClickHouse } from "@internal/clickhouse";
 import { RunEngine } from "@internal/run-engine";
 import { setupAuthenticatedEnvironment, setupBackgroundWorker } from "@internal/run-engine/tests";
-import { containerTest } from "@internal/testcontainers";
+import { containerTestWithClickhouse } from "@internal/testcontainers";
 import { trace } from "@opentelemetry/api";
 import { generateFriendlyId } from "@trigger.dev/core/v3/isomorphic";
 import type { PrismaClient } from "@trigger.dev/database";
@@ -112,7 +112,7 @@ async function triggerOnQueue(
 }
 
 describe("QueueListPresenter archived filter", () => {
-  containerTest(
+  containerTestWithClickhouse(
     "exclude hides archived queues across list, search and counts; include keeps them",
     async ({ prisma, redisOptions }) => {
       const engine = buildEngine(prisma, redisOptions);
@@ -159,7 +159,7 @@ describe("QueueListPresenter archived filter", () => {
     }
   );
 
-  containerTest(
+  containerTestWithClickhouse(
     "returns archived queues with queued runs as active, and skips idle archived queues",
     async ({ prisma, redisOptions }) => {
       const engine = buildEngine(prisma, redisOptions);
@@ -209,7 +209,7 @@ describe("QueueListPresenter archived filter", () => {
 });
 
 describe("public queue shape", () => {
-  containerTest(
+  containerTestWithClickhouse(
     "the public API item never exposes archive state",
     async ({ prisma, redisOptions }) => {
       const engine = buildEngine(prisma, redisOptions);
@@ -229,32 +229,35 @@ describe("public queue shape", () => {
 });
 
 describe("QueueAllocationPresenter", () => {
-  containerTest("excludes archived queues from the allocated sum", async ({ prisma }) => {
-    const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
-    await createQueue(prisma, environment, "emails", { concurrencyLimit: 20 });
-    await createQueue(prisma, environment, "images", { concurrencyLimit: 30 });
-    await createQueue(prisma, environment, "old", {
-      concurrencyLimit: 25,
-      archivedAt: new Date(),
-    });
-
-    const allocation = await new QueueAllocationPresenter(prisma, prisma).call({
-      environment: { ...environment, maximumConcurrencyLimit: 100 },
-    });
-
-    expect(allocation).toEqual({ totalQueues: 2, allocated: 50, unlimitedCount: 0 });
-
-    // With archiving disabled for the org, the tile counts every queue, like the list.
-    const unfiltered = await new QueueAllocationPresenter(prisma, prisma).call({
-      environment: { ...environment, maximumConcurrencyLimit: 100 },
-      excludeArchived: false,
-    });
-    expect(unfiltered).toEqual({ totalQueues: 3, allocated: 75, unlimitedCount: 0 });
-  });
+  containerTestWithClickhouse(
+    "excludes archived queues from the allocated sum",
+    async ({ prisma }) => {
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      await createQueue(prisma, environment, "emails", { concurrencyLimit: 20 });
+      await createQueue(prisma, environment, "images", { concurrencyLimit: 30 });
+      await createQueue(prisma, environment, "old", {
+        concurrencyLimit: 25,
+        archivedAt: new Date(),
+      });
+
+      const allocation = await new QueueAllocationPresenter(prisma, prisma).call({
+        environment: { ...environment, maximumConcurrencyLimit: 100 },
+      });
+
+      expect(allocation).toEqual({ totalQueues: 2, allocated: 50, unlimitedCount: 0 });
+
+      // With archiving disabled for the org, the tile counts every queue, like the list.
+      const unfiltered = await new QueueAllocationPresenter(prisma, prisma).call({
+        environment: { ...environment, maximumConcurrencyLimit: 100 },
+        excludeArchived: false,
+      });
+      expect(unfiltered).toEqual({ totalQueues: 3, allocated: 75, unlimitedCount: 0 });
+    }
+  );
 });
 
 describe("ArchiveQueueService", () => {
-  containerTest(
+  containerTestWithClickhouse(
     "archives an idle queue without touching limits, overrides, paused state or worker links",
     async ({ prisma, redisOptions }) => {
       const engine = buildEngine(prisma, redisOptions);
@@ -296,28 +299,31 @@ describe("ArchiveQueueService", () => {
     }
   );
 
-  containerTest("refuses to archive a queue with queued runs", async ({ prisma, redisOptions }) => {
-    const engine = buildEngine(prisma, redisOptions);
-    onTestFinished(() => engine.quit());
-    const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
-    await setupBackgroundWorker(engine, environment, "busy-task");
-    await deployWithout(engine, environment);
-    const queue = await prisma.taskQueue.findFirstOrThrow({
-      where: { runtimeEnvironmentId: environment.id, name: "task/busy-task" },
-    });
-    await triggerOnQueue(engine, prisma, environment, "busy-task", queue.name);
-
-    const result = await new ArchiveQueueService(prisma, engine).archive(
-      environment,
-      queue.friendlyId
-    );
-    expect(result.isErr() && result.error.type).toBe("queue_has_active_runs");
+  containerTestWithClickhouse(
+    "refuses to archive a queue with queued runs",
+    async ({ prisma, redisOptions }) => {
+      const engine = buildEngine(prisma, redisOptions);
+      onTestFinished(() => engine.quit());
+      const environment = await setupAuthenticatedEnvironment(prisma, "PRODUCTION");
+      await setupBackgroundWorker(engine, environment, "busy-task");
+      await deployWithout(engine, environment);
+      const queue = await prisma.taskQueue.findFirstOrThrow({
+        where: { runtimeEnvironmentId: environment.id, name: "task/busy-task" },
+      });
+      await triggerOnQueue(engine, prisma, environment, "busy-task", queue.name);
 
-    const after = await prisma.taskQueue.findFirstOrThrow({ where: { id: queue.id } });
-    expect(after.archivedAt).toBeNull();
-  });
+      const result = await new ArchiveQueueService(prisma, engine).archive(
+        environment,
+        queue.friendlyId
+      );
+      expect(result.isErr() && result.error.type).toBe("queue_has_active_runs");
+
+      const after = await prisma.taskQueue.findFirstOrThrow({ where: { id: queue.id } });
+      expect(after.archivedAt).toBeNull();
+    }
+  );
 
-  containerTest(
+  containerTestWithClickhouse(
     "counts runs handed to a worker queue but not yet picked up as active",
     async ({ prisma, redisOptions }) => {
       const engine = buildEngine(prisma, redisOptions);
@@ -364,7 +370,7 @@ describe("ArchiveQueueService", () => {
     }
   );
 
-  containerTest(
+  containerTestWithClickhouse(
     "refuses to archive paused queues, limit-0 or combined-limit-0 queues and named concurrency limits",
     async ({ prisma, redisOptions }) => {
       const engine = buildEngine(prisma, redisOptions);
@@ -404,7 +410,7 @@ describe("ArchiveQueueService", () => {
     }
   );
 
-  containerTest(
+  containerTestWithClickhouse(
     "check reports the reason and run count without archiving",
     async ({ prisma, redisOptions }) => {
       const engine = buildEngine(prisma, redisOptions);
@@ -430,7 +436,7 @@ describe("ArchiveQueueService", () => {
     }
   );
 
-  containerTest(
+  containerTestWithClickhouse(
     "refuses to archive a queue the current or an unpromoted deploy declares",
     async ({ prisma, redisOptions }) => {
       const engine = buildEngine(prisma, redisOptions);
@@ -470,7 +476,7 @@ describe("ArchiveQueueService", () => {
     }
   );
 
-  containerTest(
+  containerTestWithClickhouse(
     "a deploy created in the same millisecond as the current one still blocks archiving",
     async ({ prisma, redisOptions }) => {
       const engine = buildEngine(prisma, redisOptions);
@@ -506,7 +512,7 @@ describe("ArchiveQueueService", () => {
     }
   );
 
-  containerTest(
+  containerTestWithClickhouse(
     "a queue update landing between the check and the write aborts the archive",
     async ({ prisma, redisOptions }) => {
       const engine = buildEngine(prisma, redisOptions);
@@ -538,7 +544,7 @@ describe("ArchiveQueueService", () => {
     }
   );
 
-  containerTest(
+  containerTestWithClickhouse(
     "unarchive brings the queue back to the default list",
     async ({ prisma, redisOptions }) => {
       const engine = buildEngine(prisma, redisOptions);
@@ -565,7 +571,7 @@ describe("ArchiveQueueService", () => {
 });
 
 describe("QueueListPresenter ranked sort", () => {
-  containerTest(
+  containerTestWithClickhouse(
     "a busy archived queue doesn't take a slot or push queues off the last page",
     async ({ prisma, redisOptions, clickhouseContainer }) => {
       const engine = buildEngine(prisma, redisOptions);

apps/webapp/test/runEngineBatchTriggerStoreRouting.test.ts

@@ -17,7 +17,7 @@ vi.mock("~/db.server", () => ({
 import { RunEngine } from "@internal/run-engine";
 import { setupAuthenticatedEnvironment } from "@internal/run-engine/tests";
 import { PostgresRunStore, RoutingRunStore } from "@internal/run-store";
-import { containerTestWithIsolatedRedisNoClickhouse as containerTest } from "@internal/testcontainers";
+import { containerTestWithIsolatedRedis as containerTest } from "@internal/testcontainers";
 import { trace } from "@opentelemetry/api";
 import { BatchId } from "@trigger.dev/core/v3/isomorphic";
 import type { PrismaClient } from "@trigger.dev/database";

apps/webapp/test/runsListQueryError.test.ts

@@ -1,5 +1,5 @@
 import { ClickHouse } from "@internal/clickhouse";
-import { containerTest } from "@internal/testcontainers";
+import { containerTestWithClickhouse } from "@internal/testcontainers";
 import { describe, expect, vi } from "vitest";
 import {
   RunsListQueryError,
@@ -16,7 +16,7 @@ vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} }));
 vi.setConfig({ testTimeout: 90_000 });
 
 describe("runs list query error handling", () => {
-  containerTest(
+  containerTestWithClickhouse(
     "a ClickHouse resource-limit error surfaces as RunsListQueryError",
     async ({ clickhouseContainer, prisma }) => {
       const ctx = await seedParents(prisma, "qerr");