diff --git a/apps/cli/tests/helpers/cli.ts b/apps/cli/tests/helpers/cli.ts index d4f30c6882..70eab016f2 100644 --- a/apps/cli/tests/helpers/cli.ts +++ b/apps/cli/tests/helpers/cli.ts @@ -16,6 +16,7 @@ import { Path, Predicate, PubSub, + Random, Schema, Scope, Stream, @@ -290,47 +291,61 @@ export const withTempCliProject = ( const portError = (error: Error) => new E2ePortError({ message: error.message, cause: error }); -const pickReleasedPort = Effect.callback((resume) => { - const server = createServer(); - let settled = false; - const settle = (result: Effect.Effect) => { - if (settled) return; - settled = true; - resume(result); - }; - /* Stays attached after cancellation so a late listen error is never an unhandled event. */ - server.on("error", (error) => settle(Effect.fail(portError(error)))); - server.listen(0, "127.0.0.1", () => { - const address = server.address(); - if (address == null || typeof address === "string") { - server.close(() => - settle(Effect.fail(new E2ePortError({ message: "Failed to allocate a free port" }))), - ); - return; - } - server.close((error) => { - settle(error ? Effect.fail(portError(error)) : Effect.succeed(address.port)); +const isPortTaken = (error: NodeJS.ErrnoException) => + error.code === "EADDRINUSE" || error.code === "EACCES"; + +/** Succeeds with whether `port` could be bound on `host`. */ +const probePort = (port: number, host: string) => + Effect.callback((resume) => { + const server = createServer(); + let settled = false; + const settle = (result: Effect.Effect) => { + if (settled) return; + settled = true; + resume(result); + }; + /* Stays attached after cancellation so a late listen error is never an unhandled event. */ + server.on("error", (error: NodeJS.ErrnoException) => + settle(isPortTaken(error) ? Effect.succeed(false) : Effect.fail(portError(error))), + ); + server.listen(port, host, () => { + server.close((error) => { + settle(error ? Effect.fail(portError(error)) : Effect.succeed(true)); + }); + }); + return Effect.sync(() => { + settled = true; + server.close(); }); }); - return Effect.sync(() => { - settled = true; - server.close(); - }); -}); -/** - * Concurrent e2e workers each keep only ports in their own slot, so a port one worker picked and - * released before Docker binds it is never handed to another worker. Slots skip the low bit - * because Linux hands out mostly odd ports for `listen(0)`. +/* + * Ports come from below the stack's own allocation ranges (10000+, see `packages/stack/src/Ports.ts`) + * and the OS ephemeral ranges, so neither a concurrent stack nor an outbound connection can take a + * probed port before Docker binds it. Concurrent e2e workers each keep only ports in their own + * slot, so one worker never hands out a port another worker picked. */ +const FIXED_PORT_RANGE_START = 1_024; +const FIXED_PORT_RANGE_END = 10_000; + const envOrEmpty = (name: string) => Config.String(name).pipe(Config.withDefault("")); +const FIXED_PORT_SPAN = FIXED_PORT_RANGE_END - FIXED_PORT_RANGE_START; +/* Walks the range from a random start, so one worker never hands out the same port twice. */ +let nextPortOffset: number | undefined; + const pickFreePort = Effect.gen(function* () { const e2eWorkers = Number((yield* envOrEmpty("SUPABASE_E2E_WORKERS")) || "1"); const e2eWorkerSlot = Number((yield* envOrEmpty("VITEST_POOL_ID")) || "1") % e2eWorkers; - for (let attempt = 0; attempt < 1000; attempt++) { - const port = yield* pickReleasedPort; - if (Math.floor(port / 2) % e2eWorkers === e2eWorkerSlot) return port; + const start = + nextPortOffset ?? (yield* Random.nextIntBetween(0, FIXED_PORT_SPAN, { halfOpen: true })); + for (let step = 0; step < FIXED_PORT_SPAN; step++) { + const offset = (start + step) % FIXED_PORT_SPAN; + const port = FIXED_PORT_RANGE_START + offset; + if (port % e2eWorkers !== e2eWorkerSlot) continue; + nextPortOffset = offset + 1; + // Docker publishes on every interface; macOS lets that bind succeed over a loopback listener. + if ((yield* probePort(port, "0.0.0.0")) && (yield* probePort(port, "127.0.0.1"))) return port; } return yield* new E2ePortError({ message: `No free port for e2e worker slot ${e2eWorkerSlot} of ${e2eWorkers}`, diff --git a/packages/stack/src/storage/DockerDatabaseStorage.integration.test.ts b/packages/stack/src/storage/DockerDatabaseStorage.integration.test.ts index 259b482ae9..c379cbe225 100644 --- a/packages/stack/src/storage/DockerDatabaseStorage.integration.test.ts +++ b/packages/stack/src/storage/DockerDatabaseStorage.integration.test.ts @@ -6,6 +6,7 @@ import { ConfigProvider, Data, Deferred, + Duration, Effect, Exit, Fiber, @@ -71,6 +72,19 @@ class DockerTestError extends Data.TaggedError("DockerTestError")<{ class PublishBarrierError extends Data.TaggedError("PublishBarrierError")<{ readonly message: string; }> {} +class PhaseStallError extends Data.TaggedError("PhaseStallError")<{ + readonly message: string; +}> {} + +/** Fails with the stalled phase's name before the test's own timeout hides which step hung. */ +const phaseDeadline = + (phase: string, seconds = 5) => + (self: Effect.Effect) => + Effect.timeoutOrElse(self, { + duration: Duration.seconds(seconds), + orElse: () => + Effect.fail(new PhaseStallError({ message: `${phase} did not finish within ${seconds}s` })), + }); const dockerStoragePublishLoopFixture = fileURLToPath( new URL("../../tests/docker-storage-publish-loop-fixture.ts", import.meta.url), @@ -1111,7 +1125,9 @@ describe.runIf(testEngine === "docker")( const cacheRoot = path.join(root, "cache"); yield* fs.makeDirectory(instanceRoot, { recursive: true }); const markerPath = path.join(instanceRoot, ".supabase-database-storage.json"); - const helperImage = yield* preparedPostgresImage(selection.database); + const helperImage = yield* preparedPostgresImage(selection.database).pipe( + phaseDeadline("helper image preparation", 10), + ); const storage = yield* makeDockerDatabaseStorage({ runtime: "docker", target: dockerTarget, @@ -1134,9 +1150,11 @@ describe.runIf(testEngine === "docker")( // the global test-run teardown, since a container that mounted it may not yet be // reaped by the time that teardown runs. yield* Effect.addFinalizer(() => - storage.destroyData(selection.database).pipe(Effect.ignore), + storage + .destroyData(selection.database) + .pipe(Effect.ignore, phaseDeadline("destroyData cleanup"), Effect.orDie), ); - yield* storage.prepare(selection.database); + yield* storage.prepare(selection.database).pipe(phaseDeadline("prepare")); const before = yield* Schema.decodeEffect(Schema.fromJsonString(Marker))( yield* fs.readFileString(markerPath), ); @@ -1160,6 +1178,8 @@ describe.runIf(testEngine === "docker")( : engine(["rm", "--force", ...ids]).pipe(Effect.asVoid); }), Effect.ignore, + phaseDeadline(`helper container cleanup for volume ${volume}`), + Effect.orDie, ), ); // `markInitialized` (below, and inside the killed writer) checks this file through a @@ -1174,7 +1194,7 @@ describe.runIf(testEngine === "docker")( "/bin/sh", "-c", `set -eu; mkdir -p ${quote(`/store/${before.namespace}/data`)}; printf 17 > ${quote(`/store/${before.namespace}/data/PG_VERSION`)}`, - ]); + ]).pipe(phaseDeadline(`PG_VERSION seed in volume ${volume}`)); const writer = yield* spawner.spawn( ChildProcess.make( @@ -1223,9 +1243,9 @@ describe.runIf(testEngine === "docker")( }), ); yield* Effect.sync(() => process.kill(Number(writer.pid), "SIGKILL")); - yield* writer.exitCode.pipe(Effect.ignore); - yield* Fiber.join(output); - yield* Fiber.join(diagnostics); + yield* writer.exitCode.pipe(Effect.ignore, phaseDeadline("killed writer exit")); + yield* Fiber.join(output).pipe(phaseDeadline("killed writer stdout close")); + yield* Fiber.join(diagnostics).pipe(phaseDeadline("killed writer stderr close")); // The killed writer never reached the rename: the prior marker is untouched, and only // its staging file, never a partial target, is left behind. @@ -1237,7 +1257,9 @@ describe.runIf(testEngine === "docker")( (yield* fs.readDirectory(instanceRoot)).filter((entry) => entry.endsWith(".tmp")), ).toHaveLength(1); - yield* storage.markInitialized(selection.database); + yield* storage + .markInitialized(selection.database) + .pipe(phaseDeadline("markInitialized retry")); const afterRetry = yield* Schema.decodeEffect(Schema.fromJsonString(Marker))( yield* fs.readFileString(markerPath), );