fix(run-engine,testcontainers): address review - normalize retry bounds, tie-stable test assertions, shared clone/drop helpers

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: d-cs

internal-packages/run-engine/src/engine/index.ts

@@ -288,9 +288,12 @@ export class RunEngine {
       }
     );
 
-    this.snapshotsSinceReplicaRetryDelay = options.readReplicaSnapshotsSinceRetryDelay ?? {
-      minMs: 50,
-      maxMs: 200,
+    // Normalize the bounds, but keep maxMs <= 0 meaning "skip the replica retry".
+    const retryDelay = options.readReplicaSnapshotsSinceRetryDelay ?? { minMs: 50, maxMs: 200 };
+    const retryMinMs = Math.max(0, retryDelay.minMs);
+    this.snapshotsSinceReplicaRetryDelay = {
+      minMs: retryDelay.maxMs > 0 ? Math.min(retryMinMs, retryDelay.maxMs) : retryMinMs,
+      maxMs: retryDelay.maxMs,
     };
 
     const defaultHeartbeatTimeouts: HeartbeatTimeouts = {

internal-packages/run-engine/src/engine/tests/getSnapshotsSince.test.ts

@@ -890,7 +890,11 @@ describe("RunEngine getSnapshotsSince", () => {
         );
         expect(expectedSnapshots.length).toBeGreaterThan(0);
         expect(result!.length).toBe(expectedSnapshots.length);
-        expect(result!.map((s) => s.snapshot.id)).toEqual(expectedSnapshots.map((s) => s.id));
+        // Compare as sorted lists: same-millisecond snapshots have unspecified relative
+        // order in both the engine query and this test's query.
+        expect(result!.map((s) => s.snapshot.id).sort()).toEqual(
+          expectedSnapshots.map((s) => s.id).sort()
+        );
 
         expect(
           await getCounterValue("run_engine.snapshots_since.replica_miss", { outcome: "primary" })
@@ -1010,7 +1014,11 @@ describe("RunEngine getSnapshotsSince", () => {
         );
         expect(expectedSnapshots.length).toBeGreaterThan(0);
         expect(result!.length).toBe(expectedSnapshots.length);
-        expect(result!.map((s) => s.snapshot.id)).toEqual(expectedSnapshots.map((s) => s.id));
+        // Compare as sorted lists: same-millisecond snapshots have unspecified relative
+        // order in both the engine query and this test's query.
+        expect(result!.map((s) => s.snapshot.id).sort()).toEqual(
+          expectedSnapshots.map((s) => s.id).sort()
+        );
 
         // Recovered on the replica retry - the writer was never consulted.
         expect(
@@ -1183,7 +1191,11 @@ describe("RunEngine getSnapshotsSince", () => {
           (s) => s.createdAt.getTime() > since.createdAt.getTime() && s.id !== tail.id
         );
         expect(result!.map((s) => s.snapshot.id)).not.toContain(tail.id);
-        expect(result!.map((s) => s.snapshot.id)).toEqual(expectedSnapshots.map((s) => s.id));
+        // Compare as sorted lists: same-millisecond snapshots have unspecified relative
+        // order in both the engine query and this test's query.
+        expect(result!.map((s) => s.snapshot.id).sort()).toEqual(
+          expectedSnapshots.map((s) => s.id).sort()
+        );
 
         expect(await getCounterValue("run_engine.snapshots_since.replica_miss")).toBe(0);
       } finally {
@@ -1281,7 +1293,11 @@ describe("RunEngine getSnapshotsSince", () => {
           (s) => s.createdAt.getTime() > since.createdAt.getTime()
         );
         expect(expectedSnapshots.length).toBeGreaterThan(0);
-        expect(result!.map((s) => s.snapshot.id)).toEqual(expectedSnapshots.map((s) => s.id));
+        // Compare as sorted lists: same-millisecond snapshots have unspecified relative
+        // order in both the engine query and this test's query.
+        expect(result!.map((s) => s.snapshot.id).sort()).toEqual(
+          expectedSnapshots.map((s) => s.id).sort()
+        );
 
         expect(await getCounterValue("run_engine.snapshots_since.replica_miss")).toBe(0);
       } finally {
@@ -1381,7 +1397,11 @@ describe("RunEngine getSnapshotsSince", () => {
           (s) => s.createdAt.getTime() > since.createdAt.getTime()
         );
         expect(expectedSnapshots.length).toBeGreaterThan(0);
-        expect(result!.map((s) => s.snapshot.id)).toEqual(expectedSnapshots.map((s) => s.id));
+        // Compare as sorted lists: same-millisecond snapshots have unspecified relative
+        // order in both the engine query and this test's query.
+        expect(result!.map((s) => s.snapshot.id).sort()).toEqual(
+          expectedSnapshots.map((s) => s.id).sort()
+        );
 
         expect(await getCounterValue("run_engine.snapshots_since.replica_miss")).toBe(0);
       } finally {

internal-packages/testcontainers/src/index.ts

@@ -181,11 +181,7 @@ const clonedPostgresContainer = async ({}, use: Use<StartedPostgreSqlContainer>)
   const baseUri = container.getConnectionUri();
   const cloneDb = `test_${pgCloneCounter++}`;
 
-  const admin = new PrismaClient({
-    datasources: { db: { url: postgresUriWithDatabase(baseUri, "postgres") } },
-  });
-  await admin.$executeRawUnsafe(`CREATE DATABASE "${cloneDb}" TEMPLATE "${POSTGRES_TEMPLATE_DB}"`);
-  await admin.$disconnect();
+  await createDatabaseFromTemplate(baseUri, cloneDb);
 
   const cloneUri = postgresUriWithDatabase(baseUri, cloneDb);
   const view = new Proxy(container, {
@@ -200,19 +196,36 @@ const clonedPostgresContainer = async ({}, use: Use<StartedPostgreSqlContainer>)
   try {
     await use(view);
   } finally {
-    // Best-effort drop so clones don't pile up in the worker's pg over a long suite. WITH (FORCE)
-    // terminates any lingering backends (pg 13+). A failed drop is harmless - the whole container is
-    // reaped on worker exit - so we never let cleanup fail the test.
-    const cleanup = new PrismaClient({
-      datasources: { db: { url: postgresUriWithDatabase(baseUri, "postgres") } },
-    });
-    try {
-      await cleanup.$executeRawUnsafe(`DROP DATABASE IF EXISTS "${cloneDb}" WITH (FORCE)`);
-    } catch {
-      // ignore - reaped with the container anyway
-    } finally {
-      await cleanup.$disconnect();
-    }
+    await dropCloneDatabase(baseUri, cloneDb);
+  }
+};
+
+const createDatabaseFromTemplate = async (baseUri: string, cloneDb: string) => {
+  const admin = new PrismaClient({
+    datasources: { db: { url: postgresUriWithDatabase(baseUri, "postgres") } },
+  });
+  try {
+    await admin.$executeRawUnsafe(
+      `CREATE DATABASE "${cloneDb}" TEMPLATE "${POSTGRES_TEMPLATE_DB}"`
+    );
+  } finally {
+    await admin.$disconnect();
+  }
+};
+
+// Best-effort drop so clones don't pile up in the worker's pg over a long suite. WITH (FORCE)
+// terminates any lingering backends (pg 13+). A failed drop is harmless - the whole container is
+// reaped on worker exit - so we never let cleanup fail the test.
+const dropCloneDatabase = async (baseUri: string, cloneDb: string) => {
+  const cleanup = new PrismaClient({
+    datasources: { db: { url: postgresUriWithDatabase(baseUri, "postgres") } },
+  });
+  try {
+    await cleanup.$executeRawUnsafe(`DROP DATABASE IF EXISTS "${cloneDb}" WITH (FORCE)`);
+  } catch {
+    // ignore - reaped with the container anyway
+  } finally {
+    await cleanup.$disconnect();
   }
 };
 
@@ -224,11 +237,7 @@ const schemaOnlyPrismaFixture = async ({}: {}, use: Use<PrismaClient>) => {
   const baseUri = container.getConnectionUri();
   const cloneDb = `schema_only_${pgCloneCounter++}`;
 
-  const admin = new PrismaClient({
-    datasources: { db: { url: postgresUriWithDatabase(baseUri, "postgres") } },
-  });
-  await admin.$executeRawUnsafe(`CREATE DATABASE "${cloneDb}" TEMPLATE "${POSTGRES_TEMPLATE_DB}"`);
-  await admin.$disconnect();
+  await createDatabaseFromTemplate(baseUri, cloneDb);
 
   const prisma = new PrismaClient({
     datasources: { db: { url: postgresUriWithDatabase(baseUri, cloneDb) } },
@@ -237,18 +246,7 @@ const schemaOnlyPrismaFixture = async ({}: {}, use: Use<PrismaClient>) => {
     await use(prisma);
   } finally {
     await logCleanup("schemaOnlyPrisma", prisma.$disconnect());
-    // Best-effort drop, mirroring clonedPostgresContainer cleanup - the container is reaped on
-    // worker exit anyway, so never let cleanup fail the test.
-    const cleanup = new PrismaClient({
-      datasources: { db: { url: postgresUriWithDatabase(baseUri, "postgres") } },
-    });
-    try {
-      await cleanup.$executeRawUnsafe(`DROP DATABASE IF EXISTS "${cloneDb}" WITH (FORCE)`);
-    } catch {
-      // ignore - reaped with the container anyway
-    } finally {
-      await cleanup.$disconnect();
-    }
+    await dropCloneDatabase(baseUri, cloneDb);
   }
 };