diff --git a/packages/stack/src/runtime/Container.integration.test.ts b/packages/stack/src/runtime/Container.integration.test.ts index 95777aa01e..8051607c28 100644 --- a/packages/stack/src/runtime/Container.integration.test.ts +++ b/packages/stack/src/runtime/Container.integration.test.ts @@ -821,46 +821,104 @@ describe("container process adapter", { timeout: 120_000 }, () => { ).pipe(Effect.provide(NodeServices.layer)), ); - it.live("classifies a container start the engine could not publish as port allocation", () => - Effect.scoped( - Effect.gen(function* () { - const delegate = yield* ChildProcessSpawner.ChildProcessSpawner; - const runtime = yield* makeContainerRuntime({ - target: containerTarget, - root: ".", - hostGateway: sharedHostGateway, - }); - yield* runtime.prepare(image); - const result = yield* makeContainerRuntime({ - target: containerTarget, - root: ".", - hostGateway: sharedHostGateway, - }).pipe( - Effect.flatMap((rejecting) => - rejecting.launch({ - image, - stackId: "c".repeat(64), - instanceId: "port-allocated", - env: {}, - args: ["-e", "process.exit(0)"], - ports: [8080], - }), - ), - Effect.provideService( - ChildProcessSpawner.ChildProcessSpawner, - makePortAllocatedStartSpawner(delegate), - ), - Effect.exit, - ); - const failure = Exit.isFailure(result) - ? Option.getOrUndefined(Cause.findErrorOption(result.cause)) - : undefined; - if (!(failure instanceof ContainerLaunchError)) - return yield* Effect.die("publish failure did not retain cleanup authority"); - yield* failure.process.remove; - expect(failureKind(failure)).toBe("port-allocation"); - }), - ).pipe(Effect.provide(NodeServices.layer)), + it.live( + "gives up after bounded attempts when a published host port keeps colliding, leaving only the last container", + () => + Effect.scoped( + Effect.gen(function* () { + const delegate = yield* ChildProcessSpawner.ChildProcessSpawner; + const runtime = yield* makeContainerRuntime({ + target: containerTarget, + root: ".", + hostGateway: sharedHostGateway, + }); + yield* runtime.prepare(image); + const { spawner, started } = makeFailingStartSpawner( + delegate, + dockerPortAllocated, + Number.POSITIVE_INFINITY, + ); + const result = yield* makeContainerRuntime({ + target: containerTarget, + root: ".", + hostGateway: sharedHostGateway, + }).pipe( + Effect.flatMap((rejecting) => + rejecting.launch({ + image, + stackId: "c".repeat(64), + instanceId: "port-allocated", + env: {}, + args: ["-e", "process.exit(0)"], + ports: [8080], + }), + ), + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + Effect.exit, + ); + const failure = Exit.isFailure(result) + ? Option.getOrUndefined(Cause.findErrorOption(result.cause)) + : undefined; + if (!(failure instanceof ContainerLaunchError)) + return yield* Effect.die("publish failure did not retain cleanup authority"); + expect(failureKind(failure)).toBe("port-allocation"); + expect(failure.failure.message).toContain("port is already allocated"); + expect(started).toHaveLength(3); + expect(new Set(started).size).toBe(3); + expect(yield* Effect.forEach(started, exists)).toEqual([false, false, true]); + expect(failure.process.id).toBe(started[2]); + yield* failure.process.remove; + expect(yield* exists(started[2]!)).toBe(false); + }), + ).pipe(Effect.provide(NodeServices.layer)), + ); + + it.live( + "recreates a container whose published host port was taken before the engine bound it", + () => + Effect.scoped( + Effect.gen(function* () { + const delegate = yield* ChildProcessSpawner.ChildProcessSpawner; + const fs = yield* FileSystem.FileSystem; + const { spawner, started, envFiles } = makeFailingStartSpawner( + delegate, + pastaAddressInUse, + 1, + ); + const process = yield* makeContainerRuntime({ + target: containerTarget, + root: ".", + hostGateway: sharedHostGateway, + }).pipe( + Effect.flatMap((flaky) => + flaky.prepare(image).pipe( + Effect.andThen( + flaky.launch({ + image, + stackId: "c".repeat(64), + instanceId: "port-recreated", + env: {}, + ports: [8080], + args: [ + "-e", + "Bun.serve({ hostname: '0.0.0.0', port: 8080, fetch() { return new Response('ok') } }); console.log('ready')", + ], + }), + ), + ), + ), + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + ); + expect(started).toHaveLength(2); + expect(process.id).toBe(started[1]); + expect(yield* exists(started[0]!)).toBe(false); + expect(yield* exists(started[1]!)).toBe(true); + expect(process.ports[8080]).toBeDefined(); + expect(envFiles).toHaveLength(2); + expect(yield* fs.exists(envFiles[0]!)).toBe(false); + expect(yield* fs.exists(envFiles[1]!)).toBe(true); + }), + ).pipe(Effect.provide(NodeServices.layer)), ); it.live("keeps exit observation shared after a caller cancels its wait", () => @@ -1757,26 +1815,43 @@ const makePullFailureSpawner = ( return delegate.spawn(command); }); -/** Fails the engine's `start` the way Docker reports a published host port another listener holds. */ -const makePortAllocatedStartSpawner = (delegate: ChildProcessSpawnerService["Service"]) => - ChildProcessSpawner.make((command) => - ChildProcess.isStandardCommand(command) && - command.command === testEngine && - command.args[0] === "start" - ? delegate.spawn( +const dockerPortAllocated = + "Error response from daemon: driver failed programming external connectivity on endpoint port-allocated: Bind for 127.0.0.1:54321 failed: port is already allocated"; +const pastaAddressInUse = + 'Error: unable to start container "abc": pasta failed with exit code 1:\nListen failed for HOST TCP port 127.0.0.1/42651: Address in use'; + +/** Fails the first `failures` engine `start` calls with `message`, recording every started container name. */ +const makeFailingStartSpawner = ( + delegate: ChildProcessSpawnerService["Service"], + message: string, + failures: number, +) => { + const started: Array = []; + const envFiles: Array = []; + const spawner = ChildProcessSpawner.make((command) => { + if (ChildProcess.isStandardCommand(command) && command.command === testEngine) { + const envFile = command.args[command.args.indexOf("--env-file") + 1]; + if (command.args[0] === "create" && envFile !== undefined) envFiles.push(envFile); + } + if ( + !ChildProcess.isStandardCommand(command) || + command.command !== testEngine || + command.args[0] !== "start" + ) + return delegate.spawn(command); + started.push(String(command.args.at(-1))); + return started.length > failures + ? delegate.spawn(command) + : delegate.spawn( ChildProcess.make( process.execPath, - [ - "-e", - `console.error(${JSON.stringify( - "Error response from daemon: driver failed programming external connectivity on endpoint port-allocated: Bind for 127.0.0.1:54321 failed: port is already allocated", - )}); process.exit(1)`, - ], + ["-e", `console.error(${JSON.stringify(message)}); process.exit(1)`], { stdin: "ignore" }, ), - ) - : delegate.spawn(command), - ); + ); + }); + return { spawner, started, envFiles }; +}; const hostGatewayRejectionScript = `console.error(${JSON.stringify( 'Error response from daemon: invalid IP address in add-host: "host-gateway"', diff --git a/packages/stack/src/runtime/Container.ts b/packages/stack/src/runtime/Container.ts index c8361fb761..cbabf4bbf7 100644 --- a/packages/stack/src/runtime/Container.ts +++ b/packages/stack/src/runtime/Container.ts @@ -95,7 +95,12 @@ export interface ContainerRuntime { /** The engine picks published host ports (`127.0.0.1::port`), so a refusal is its own allocation. */ const portAllocated = (message: string) => - /port is already allocated|address already in use/iu.test(message); + /port is already allocated|address already in use|listen failed for host tcp port[^\n]*address in use/iu.test( + message, + ); + +/** Podman releases the host port it picks at create before pasta binds it at start, so another listener can take it. */ +const PORT_COLLISION_ATTEMPTS = 3; /** A refused registry connection also reads "connection refused", so a pull needs the daemon named. */ const pullEngineUnreachable = (cause: unknown, message: string) => @@ -716,10 +721,10 @@ export const makeContainerRuntime = (options: { ? hostGateway.prefetch(hostGatewayProbe(image, spec)).pipe(Effect.as("host-gateway")) : hostGateway.resolve(hostGatewayProbe(image, spec)); - const launch = Effect.fn("Container.launch")(function* ( + const launchOnce = Effect.fnUntraced(function* ( spec: ContainerSpec, - interactive = false, - oneOff = false, + interactive: boolean, + oneOff: boolean, ) { const owner = yield* Scope.Scope; const image = (yield* Ref.get(mirrored)).get(spec.image) ?? spec.image; @@ -1115,6 +1120,45 @@ export const makeContainerRuntime = (options: { }), ); }); + const launchWithRetry = ( + spec: ContainerSpec, + interactive: boolean, + oneOff: boolean, + owner: Scope.Scope, + attempt: number, + ): ReturnType => + Effect.gen(function* () { + yield* Effect.annotateCurrentSpan({ "retry.attempt_count": attempt }); + const attemptScope = yield* Scope.fork(owner); + return yield* launchOnce(spec, interactive, oneOff).pipe( + Effect.provideService(Scope.Scope, attemptScope), + Effect.catchTag("ContainerLaunchError", (error) => + attempt >= PORT_COLLISION_ATTEMPTS || error.failure.kind !== "port-allocation" + ? Effect.fail(error) + : Effect.logDebug(`Retrying container launch: ${error.failure.message}`).pipe( + Effect.andThen( + error.process.remove.pipe( + Effect.tapError((cause) => + Effect.logWarning( + `Failed to remove collided container ${error.process.id}: ${cause.message}`, + ), + ), + Effect.mapError(() => error), + ), + ), + Effect.andThen(Scope.close(attemptScope, Exit.void)), + Effect.andThen(launchWithRetry(spec, interactive, oneOff, owner, attempt + 1)), + ), + ), + ); + }); + const launch = Effect.fn("Container.launch")(function* ( + spec: ContainerSpec, + interactive = false, + oneOff = false, + ) { + return yield* launchWithRetry(spec, interactive, oneOff, yield* Scope.Scope, 1); + }); return { prepare, prepareImage,