diff --git a/docs/features/containers.md b/docs/features/containers.md index 6310b872c..fe5edee88 100644 --- a/docs/features/containers.md +++ b/docs/features/containers.md @@ -716,6 +716,8 @@ The following options can be provided to modify the command execution: 3. **`env`:** A map of environment variables to set inside the container. +4. **`signal`:** An `AbortSignal` to stop waiting for the command. `exec` rejects when the signal aborts. Docker cannot kill a command it has started, so the process keeps running inside the container until it exits or the container stops. + ```js const container = await new GenericContainer("alpine") .withCommand(["sleep", "infinity"]) @@ -727,7 +729,8 @@ const { output, stdout, stderr, exitCode } = await container.exec(["echo", "hell env: { "VAR1": "enabled", "VAR2": "/app/debug.log", - } + }, + signal: AbortSignal.timeout(5_000), }); ``` diff --git a/docs/features/wait-strategies.md b/docs/features/wait-strategies.md index e9c27e04a..1316ba18e 100644 --- a/docs/features/wait-strategies.md +++ b/docs/features/wait-strategies.md @@ -267,3 +267,5 @@ const container = await new GenericContainer("alpine") .withWaitStrategy(new ReadyAfterDelayWaitStrategy()) .start(); ``` + +`checkStartupState` is passed an `AbortSignal` as a third argument. It aborts when the startup timeout is reached, so pass it to Dockerode calls as `abortSignal` to cancel them. diff --git a/packages/modules/couchbase/src/couchbase-container.ts b/packages/modules/couchbase/src/couchbase-container.ts index c7db831c7..5c34c6b22 100644 --- a/packages/modules/couchbase/src/couchbase-container.ts +++ b/packages/modules/couchbase/src/couchbase-container.ts @@ -5,8 +5,8 @@ import { GenericContainer, getContainerRuntimeClient, type InspectResult, - IntervalRetry, log, + poll, type StartedTestContainer, Wait, type WaitStrategy, @@ -402,13 +402,13 @@ export class CouchbaseContainer extends GenericContainer { await this.waitUntilBucketIsHealthy(client, inspectResult, startedTestContainer, bucket, remainingStartupMs()); if (this.enabledServices.has(CouchbaseService.QUERY)) { - await this.waitUntilKeyspaceIsPresent(client, startedTestContainer, bucket, remainingStartupMs()); + await this.waitUntilKeyspaceIsPresent(startedTestContainer, bucket, remainingStartupMs()); } if (bucket.hasPrimaryIndex()) { if (this.enabledServices.has(CouchbaseService.QUERY)) { await this.createPrimaryIndex(startedTestContainer, bucket); - await this.waitUntilPrimaryIndexIsOnline(client, startedTestContainer, bucket, remainingStartupMs()); + await this.waitUntilPrimaryIndexIsOnline(startedTestContainer, bucket, remainingStartupMs()); } else { log.info( `Primary index creation for bucket ${bucket.getName()} ignored, since QUERY service is not present.` @@ -503,12 +503,11 @@ export class CouchbaseContainer extends GenericContainer { } private async waitUntilKeyspaceIsPresent( - client: ContainerRuntimeClient, startedTestContainer: StartedTestContainer, bucket: BucketDefinition, timeoutMs: number ) { - await new IntervalRetry(1000).retryUntil( + await poll( async () => { try { const body = new URLSearchParams(); @@ -533,20 +532,18 @@ export class CouchbaseContainer extends GenericContainer { return undefined; } }, - async (response) => { - if (response === undefined) { - return false; - } - const jsonResponse = (await response.json()) as { results: Array<{ present: boolean }> }; - return jsonResponse.results[0].present; - }, - () => { - const message = `URL /query/service not accessible after ${timeoutMs}ms`; - log.error(message, { containerId: client.container.getById(startedTestContainer.getId()).id }); - - throw new Error(message); - }, - timeoutMs + { + until: async (response) => { + if (response === undefined) { + return false; + } + const jsonResponse = (await response.json()) as { results: Array<{ present: boolean }> }; + return jsonResponse.results[0].present; + }, + intervalMs: 1000, + timeoutMs, + timeoutMessage: `URL /query/service not accessible after ${timeoutMs}ms`, + } ); } @@ -567,12 +564,11 @@ export class CouchbaseContainer extends GenericContainer { } private async waitUntilPrimaryIndexIsOnline( - client: ContainerRuntimeClient, startedTestContainer: StartedTestContainer, bucket: BucketDefinition, timeoutMs: number ) { - await new IntervalRetry(1000).retryUntil( + await poll( async () => { try { const body = new URLSearchParams(); @@ -597,20 +593,18 @@ export class CouchbaseContainer extends GenericContainer { return undefined; } }, - async (response) => { - if (response === undefined) { - return false; - } - const jsonResponse = (await response.json()) as { results: Array<{ online: boolean }> }; - return jsonResponse.results[0].online; - }, - () => { - const message = `URL /query/service not accessible after ${timeoutMs}ms`; - log.error(message, { containerId: client.container.getById(startedTestContainer.getId()).id }); - - throw new Error(message); - }, - timeoutMs + { + until: async (response) => { + if (response === undefined) { + return false; + } + const jsonResponse = (await response.json()) as { results: Array<{ online: boolean }> }; + return jsonResponse.results[0].online; + }, + intervalMs: 1000, + timeoutMs, + timeoutMessage: `URL /query/service not accessible after ${timeoutMs}ms`, + } ); } @@ -658,22 +652,21 @@ export class CouchbaseContainer extends GenericContainer { } private async waitUntilKvAcceptsCredentials(container: StartedTestContainer, timeoutMs: number) { - await new IntervalRetry(100).retryUntil( - async () => { + await poll( + async ({ signal }) => { const { exitCode } = await container.exec( ["/opt/couchbase/bin/mcstat", "--host", `127.0.0.1:${PORTS.KV_PORT}`, "--user", this.username, "connections"], - { env: { CB_PASSWORD: this.password } } + { env: { CB_PASSWORD: this.password }, signal } ); return exitCode; }, - (exitCode) => exitCode === 0, - () => { - const message = `KV service did not accept credentials after ${timeoutMs}ms`; - log.error(message, { containerId: container.getId() }); - - throw new Error(message); - }, - timeoutMs + { + until: (exitCode) => exitCode === 0, + intervalMs: 100, + timeoutMs, + timeoutMessage: `KV service did not accept credentials after ${timeoutMs}ms`, + retryOnError: false, + } ); } diff --git a/packages/modules/mongodb/src/mongodb-atlas-local-container.test.ts b/packages/modules/mongodb/src/mongodb-atlas-local-container.test.ts index b69a73de2..210b134ec 100644 --- a/packages/modules/mongodb/src/mongodb-atlas-local-container.test.ts +++ b/packages/modules/mongodb/src/mongodb-atlas-local-container.test.ts @@ -1,5 +1,5 @@ import mongoose from "mongoose"; -import { IntervalRetry } from "testcontainers"; +import { poll } from "testcontainers"; import { getImage } from "../../../testcontainers/src/utils/test-helper"; import { MongoDBAtlasLocalContainer } from "./mongodb-atlas-local-container"; @@ -114,19 +114,22 @@ describe("MongoDBAtlasLocalContainer", { timeout: 240_000 }, () => { definition: ATLAS_SEARCH_INDEX, }); - await new IntervalRetry(10).retryUntil( + await poll( async () => { const searchIndexes = await collection.listSearchIndexes("AtlasSearchIndex").toArray(); return (searchIndexes[0] as { status?: string } | undefined)?.status; }, - (status) => status?.toUpperCase() === "READY", - () => new Error("Atlas Search index did not become ready in time"), - 5_000 + { + until: (status) => status?.toUpperCase() === "READY", + intervalMs: 10, + timeoutMs: 5_000, + timeoutMessage: "Atlas Search index did not become ready in time", + } ); await collection.insertOne({ test: "tests", test2: 123, test3: true }); - const found = await new IntervalRetry | null, Error>(10).retryUntil( + const found = await poll( async () => collection .aggregate([ @@ -142,15 +145,14 @@ describe("MongoDBAtlasLocalContainer", { timeout: 240_000 }, () => { }, ]) .next(), - (result) => result !== null, - () => new Error("Atlas Search did not return a result in time"), - 5_000 + { + until: (result) => result !== null, + intervalMs: 10, + timeoutMs: 5_000, + timeoutMessage: "Atlas Search did not return a result in time", + } ); - if (found instanceof Error) { - throw found; - } - expect(found).toMatchObject({ test: "tests", test2: 123, test3: true }); } finally { await db.close(); diff --git a/packages/testcontainers/src/common/clock.ts b/packages/testcontainers/src/common/clock.ts deleted file mode 100644 index 0656883f4..000000000 --- a/packages/testcontainers/src/common/clock.ts +++ /dev/null @@ -1,11 +0,0 @@ -export type Time = number; - -export interface Clock { - getTime(): Time; -} - -export class SystemClock implements Clock { - public getTime(): Time { - return Date.now(); - } -} diff --git a/packages/testcontainers/src/common/index.ts b/packages/testcontainers/src/common/index.ts index f9cd5d29f..4557ee3e6 100644 --- a/packages/testcontainers/src/common/index.ts +++ b/packages/testcontainers/src/common/index.ts @@ -1,7 +1,7 @@ export { withFileLock } from "./file-lock"; export { hash } from "./hash"; export { buildLog, composeLog, containerLog, execLog, Logger, log, pullLog } from "./logger"; -export { IntervalRetry, Retry } from "./retry"; +export { IntervalRetry, PollOptions, poll, Retry } from "./retry"; export * from "./time"; export * from "./type-guards"; export { RandomUuid, randomUuid, Uuid } from "./uuid"; diff --git a/packages/testcontainers/src/common/retry.test.ts b/packages/testcontainers/src/common/retry.test.ts index 3755aa974..595269c4d 100644 --- a/packages/testcontainers/src/common/retry.test.ts +++ b/packages/testcontainers/src/common/retry.test.ts @@ -1,6 +1,159 @@ -import { IntervalRetry } from "./retry"; +import { IntervalRetry, poll } from "./retry"; + +describe("Retry", { concurrent: false }, () => { + describe("poll", () => { + it("should not retry when the first attempt resolves", async () => { + const fnMock = vi.fn().mockResolvedValue("result"); + + const result = await poll(() => fnMock(), { intervalMs: 1, timeoutMs: 1000 }); + + expect(result).toEqual("result"); + expect(fnMock).toHaveBeenCalledTimes(1); + }); + + it("should retry until the predicate passes", async () => { + const fnMock = vi.fn().mockResolvedValueOnce(1).mockResolvedValueOnce(2).mockResolvedValueOnce(3); + + const result = await poll(() => fnMock(), { until: (value) => value === 3, intervalMs: 1, timeoutMs: 1000 }); + + expect(result).toEqual(3); + expect(fnMock).toHaveBeenCalledTimes(3); + }); + + it("should pass the attempt number", async () => { + const attempts: number[] = []; + + await poll(({ attempt }) => attempts.push(attempt), { + until: (count) => count === 3, + intervalMs: 1, + timeoutMs: 1000, + }); + + expect(attempts).toEqual([1, 2, 3]); + }); + + it("should retry when an attempt throws", async () => { + const fnMock = vi.fn().mockRejectedValueOnce(new Error("not ready")).mockResolvedValueOnce("result"); + + const result = await poll(() => fnMock(), { intervalMs: 1, timeoutMs: 1000 }); + + expect(result).toEqual("result"); + expect(fnMock).toHaveBeenCalledTimes(2); + }); + + it("should rethrow when an attempt throws and retryOnError is false", async () => { + const fnMock = vi.fn().mockRejectedValue(new Error("not running")); + + await expect(poll(() => fnMock(), { intervalMs: 1, timeoutMs: 1000, retryOnError: false })).rejects.toThrow( + "not running" + ); + expect(fnMock).toHaveBeenCalledTimes(1); + }); + + it("should rethrow when the predicate throws", async () => { + const fnMock = vi.fn().mockResolvedValue("result"); + + await expect( + poll(() => fnMock(), { + until: () => { + throw new Error("invalid predicate"); + }, + intervalMs: 1, + timeoutMs: 1000, + }) + ).rejects.toThrow("invalid predicate"); + expect(fnMock).toHaveBeenCalledTimes(1); + }); + + it("should reject with the timeout message when the predicate never passes", async () => { + const promise = poll(async () => false, { + until: (result) => result, + intervalMs: 1, + timeoutMs: 20, + timeoutMessage: "not ready after 20ms", + }); + + await expect(promise).rejects.toThrow("not ready after 20ms"); + await expect(promise).rejects.not.toHaveProperty("cause"); + }); + + it("should reject with a default timeout message", async () => { + await expect( + poll(async () => false, { until: (result) => result, intervalMs: 1, timeoutMs: 20 }) + ).rejects.toThrow("Timed out after 20ms"); + }); + + it("should reject with the last error as the cause on timeout", async () => { + let attempts = 0; + + const promise = poll( + async () => { + throw new Error(`failure ${++attempts}`); + }, + { intervalMs: 1, timeoutMs: 20 } + ); + + await expect(promise).rejects.toThrow("Timed out after 20ms"); + await expect(promise).rejects.toHaveProperty("cause", new Error(`failure ${attempts}`)); + expect(attempts).toBeGreaterThan(1); + }); + + it("should not time out early when the timeout exceeds the maximum timer delay", async () => { + const fnMock = vi.fn().mockResolvedValueOnce(false).mockResolvedValueOnce(true); + + const result = await poll(() => fnMock(), { until: (value) => value, intervalMs: 20, timeoutMs: Infinity }); + + expect(result).toEqual(true); + }); + + it("should reject at the timeout when an attempt never resolves", { timeout: 1000 }, async () => { + const fnMock = vi.fn(() => new Promise(() => {})); + + await expect( + poll(() => fnMock(), { intervalMs: 1, timeoutMs: 20, timeoutMessage: "not ready after 20ms" }) + ).rejects.toThrow("not ready after 20ms"); + expect(fnMock).toHaveBeenCalledTimes(1); + }); + + it("should abort the signal with the timeout error at the timeout", async () => { + const signals: AbortSignal[] = []; + + const promise = poll( + ({ signal }) => { + signals.push(signal); + return new Promise((_, reject) => signal.addEventListener("abort", () => reject(signal.reason))); + }, + { intervalMs: 1, timeoutMs: 20 } + ); + + await expect(promise).rejects.toThrow("Timed out after 20ms"); + expect(signals).toHaveLength(1); + expect(signals[0].aborted).toBe(true); + await expect(promise).rejects.toBe(signals[0].reason); + }); + + it("should not abort the signal when polling succeeds", async () => { + const signals: AbortSignal[] = []; + + await poll(({ signal }) => signals.push(signal), { intervalMs: 1, timeoutMs: 20 }); + await new Promise((resolve) => setTimeout(resolve, 40)); + + expect(signals[0].aborted).toBe(false); + }); + + it("should not start another attempt after the timeout", async () => { + const fnMock = vi.fn().mockResolvedValue(false); + + await expect( + poll(() => fnMock(), { until: (result) => result, intervalMs: 10, timeoutMs: 25 }) + ).rejects.toThrow(); + const callsAtTimeout = fnMock.mock.calls.length; + await new Promise((resolve) => setTimeout(resolve, 40)); + + expect(fnMock).toHaveBeenCalledTimes(callsAtTimeout); + }); + }); -describe("Retry", () => { describe("IntervalRetry", () => { it("should not retry when predicate succeeds", async () => { const fnMock = vi.fn().mockResolvedValue(true); diff --git a/packages/testcontainers/src/common/retry.ts b/packages/testcontainers/src/common/retry.ts index 3d07821ef..96824c067 100644 --- a/packages/testcontainers/src/common/retry.ts +++ b/packages/testcontainers/src/common/retry.ts @@ -1,37 +1,81 @@ -import { type Clock, SystemClock, type Time } from "./clock"; +import { setTimeout as sleep } from "node:timers/promises"; -export interface Retry { - retryUntil( - fn: () => Promise, - predicate: (result: T) => boolean | Promise, - onTimeout: () => U, - timeoutMs: number - ): Promise; -} +// setTimeout fires after 1ms when the delay exceeds this +const MAX_TIMEOUT_MS = 2 ** 31 - 1; + +export type PollOptions = { + until?: (result: T) => boolean | Promise; + intervalMs: number; + timeoutMs: number; + timeoutMessage?: string; + retryOnError?: boolean; +}; + +export async function poll( + fn: (context: { attempt: number; signal: AbortSignal }) => T | Promise, + options: PollOptions +): Promise { + const { + until, + intervalMs, + timeoutMs, + timeoutMessage = `Timed out after ${timeoutMs}ms`, + retryOnError = true, + } = options; + const controller = new AbortController(); + const { signal } = controller; + let lastError: unknown; -abstract class AbstractRetry implements Retry { - protected constructor(protected readonly clock: Clock = new SystemClock()) {} + const attempts = async (): Promise => { + for (let attempt = 1; ; attempt++) { + let outcome: { result: T } | undefined; + try { + outcome = { result: await fn({ attempt, signal }) }; + lastError = undefined; + } catch (err) { + if (!retryOnError) { + throw err; + } + lastError = err; + } + + if (outcome !== undefined && (until === undefined || (await until(outcome.result)))) { + return outcome.result; + } - public abstract retryUntil( + await sleep(intervalMs, undefined, { signal }); + } + }; + + return new Promise((resolve, reject) => { + const deadline = setTimeout( + () => { + const error = new Error(timeoutMessage, lastError === undefined ? {} : { cause: lastError }); + reject(error); + controller.abort(error); + }, + Math.min(timeoutMs, MAX_TIMEOUT_MS) + ); + + attempts() + .then(resolve, reject) + .finally(() => clearTimeout(deadline)); + }); +} + +/** @deprecated Use {@link poll} instead. */ +export interface Retry { + retryUntil( fn: () => Promise, predicate: (result: T) => boolean | Promise, onTimeout: () => U, timeoutMs: number ): Promise; - - protected hasTimedOut(timeoutMs: number, startTime: Time): boolean { - return this.clock.getTime() - startTime > timeoutMs; - } - - protected wait(duration: number): Promise { - return new Promise((resolve) => setTimeout(resolve, duration)); - } } -export class IntervalRetry extends AbstractRetry { - constructor(private readonly interval: number) { - super(); - } +/** @deprecated Use {@link poll} instead. */ +export class IntervalRetry implements Retry { + constructor(private readonly interval: number) {} public async retryUntil( fn: (attempt: number) => Promise, @@ -39,16 +83,16 @@ export class IntervalRetry extends AbstractRetry { onTimeout: () => U, timeoutMs: number ): Promise { - const startTime = this.clock.getTime(); + const startTime = Date.now(); let attemptNumber = 0; let result = await fn(attemptNumber++); while (!(await predicate(result))) { - if (this.hasTimedOut(timeoutMs, startTime)) { + if (Date.now() - startTime > timeoutMs) { return onTimeout(); } - await this.wait(this.interval); + await sleep(this.interval); result = await fn(attemptNumber++); } diff --git a/packages/testcontainers/src/container-runtime/clients/container/container-client.ts b/packages/testcontainers/src/container-runtime/clients/container/container-client.ts index a6e4020e1..14b086c46 100644 --- a/packages/testcontainers/src/container-runtime/clients/container/container-client.ts +++ b/packages/testcontainers/src/container-runtime/clients/container/container-client.ts @@ -32,7 +32,7 @@ export interface ContainerClient { list(): Promise; create(opts: ContainerCreateOptions): Promise; start(container: Container): Promise; - inspect(container: Container): Promise; + inspect(container: Container, opts?: { signal?: AbortSignal }): Promise; stop(container: Container, opts?: { timeout: number }): Promise; attach(container: Container): Promise; logs(container: Container, opts?: ContainerLogsOptions): Promise; diff --git a/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.test.ts b/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.test.ts index 7d55b4715..b426fad72 100644 --- a/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.test.ts +++ b/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.test.ts @@ -64,5 +64,72 @@ describe("DockerContainerClient", () => { expect(result.stdout).toBe(payload); expect(result.output).toBe(payload); }); + + it("should reject and destroy the Docker stream when the signal aborts", async () => { + const rawStream = new PassThrough(); + let started = false; + const exec = { + start: async () => { + started = true; + return rawStream; + }, + inspect: async () => ({ ExitCode: 0 }), + }; + const container = { id: "container-id", exec: async () => exec }; + const dockerode = { modem: { demuxStream: () => {} } } as unknown as Dockerode; + const client = new DockerContainerClient(dockerode); + const controller = new AbortController(); + + const result = client.exec(container as unknown as Dockerode.Container, ["sleep", "30"], { + signal: controller.signal, + }); + await vi.waitFor(() => expect(started).toBe(true)); + controller.abort(new Error("aborted")); + + await expect(result).rejects.toThrow("aborted"); + expect(rawStream.destroyed).toBe(true); + }); + + it("should not create an exec when the signal has already aborted", async () => { + let created = false; + const container = { + id: "container-id", + exec: async () => { + created = true; + return { + start: async () => new PassThrough().end(), + inspect: async () => ({ ExitCode: 0 }), + }; + }, + }; + const dockerode = { modem: { demuxStream: () => {} } } as unknown as Dockerode; + const client = new DockerContainerClient(dockerode); + + await expect( + client.exec(container as unknown as Dockerode.Container, ["echo", "hi"], { + signal: AbortSignal.abort(new Error("aborted")), + }) + ).rejects.toThrow("aborted"); + expect(created).toBe(false); + }); + }); + + describe("inspect", () => { + it("should reject when the signal aborts", async () => { + const container = { + id: "container-id", + inspect: (opts?: { abortSignal?: AbortSignal }) => + new Promise((_, reject) => { + opts?.abortSignal?.addEventListener("abort", () => reject(opts.abortSignal?.reason)); + }), + }; + const client = new DockerContainerClient({} as Dockerode); + const controller = new AbortController(); + + const result = client.inspect(container as unknown as Dockerode.Container, { signal: controller.signal }); + controller.abort(new Error("aborted")); + + await expect(result).rejects.toThrow("aborted"); + }); }); }); diff --git a/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.ts b/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.ts index bcf21d381..8555f433e 100644 --- a/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.ts +++ b/packages/testcontainers/src/container-runtime/clients/container/docker-container-client.ts @@ -129,11 +129,13 @@ export class DockerContainerClient implements ContainerClient { } } - async inspect(container: Dockerode.Container): Promise { + async inspect(container: Dockerode.Container, opts?: { signal?: AbortSignal }): Promise { try { - return await container.inspect(); + return await container.inspect(opts?.signal && { abortSignal: opts.signal }); } catch (err) { - log.error(`Failed to inspect container: ${err}`, { containerId: container.id }); + if (!opts?.signal?.aborted) { + log.error(`Failed to inspect container: ${err}`, { containerId: container.id }); + } throw err; } } @@ -207,6 +209,7 @@ export class DockerContainerClient implements ContainerClient { } async exec(container: Container, command: string[], opts?: Partial): Promise { + const signal = opts?.signal; const execOptions = this.createExecOptions(command, opts); const outputChunks: string[] = []; @@ -214,12 +217,14 @@ export class DockerContainerClient implements ContainerClient { const stderrChunks: string[] = []; try { + signal?.throwIfAborted(); + if (opts?.log) { log.debug(`Execing container with command "${command.join(" ")}"...`, { containerId: container.id }); } const exec = await container.exec(execOptions); - const stream = await exec.start({ stdin: true, Detach: false, Tty: false }); + const stream = await exec.start({ stdin: true, Detach: false, Tty: false, abortSignal: signal }); const stdoutStream = new PassThrough(); const stderrStream = new PassThrough(); @@ -240,9 +245,9 @@ export class DockerContainerClient implements ContainerClient { processStream(stdoutStream, stdoutChunks); processStream(stderrStream, stderrChunks); - await this.waitForExecStreamToEnd(stream, stdoutStream, stderrStream); + await this.waitForExecStreamToEnd(stream, stdoutStream, stderrStream, signal); - const inspectResult = await exec.inspect(); + const inspectResult = await exec.inspect({ abortSignal: signal }); const exitCode = inspectResult.ExitCode ?? -1; const output = outputChunks.join(""); const stdout = stdoutChunks.join(""); @@ -253,9 +258,11 @@ export class DockerContainerClient implements ContainerClient { } return { output, stdout, stderr, exitCode }; } catch (err) { - log.error(`Failed to exec container with command "${command.join(" ")}": ${err}: ${outputChunks.join("")}`, { - containerId: container.id, - }); + if (!signal?.aborted) { + log.error(`Failed to exec container with command "${command.join(" ")}": ${err}: ${outputChunks.join("")}`, { + containerId: container.id, + }); + } throw err; } } @@ -265,6 +272,7 @@ export class DockerContainerClient implements ContainerClient { Cmd: command, AttachStdout: true, AttachStderr: true, + abortSignal: opts?.signal, }; if (opts?.env !== undefined) { @@ -283,18 +291,26 @@ export class DockerContainerClient implements ContainerClient { private async waitForExecStreamToEnd( stream: Duplex, stdoutStream: PassThrough, - stderrStream: PassThrough + stderrStream: PassThrough, + signal?: AbortSignal ): Promise { + let rejectOnAbort = () => {}; + try { + signal?.throwIfAborted(); + await new Promise((res, rej) => { + rejectOnAbort = () => rej(signal?.reason); stream.on("end", res); stream.on("error", rej); + signal?.addEventListener("abort", rejectOnAbort); }); stdoutStream.end(); stderrStream.end(); await Promise.all([finished(stdoutStream), finished(stderrStream)]); } finally { + signal?.removeEventListener("abort", rejectOnAbort); stream.destroy(); } } diff --git a/packages/testcontainers/src/container-runtime/clients/container/types.ts b/packages/testcontainers/src/container-runtime/clients/container/types.ts index a6fbe40ba..32aa4c6e3 100644 --- a/packages/testcontainers/src/container-runtime/clients/container/types.ts +++ b/packages/testcontainers/src/container-runtime/clients/container/types.ts @@ -1,6 +1,6 @@ import type { Environment } from "../../../types"; -export type ExecOptions = { workingDir: string; user: string; env: Environment; log: boolean }; +export type ExecOptions = { workingDir: string; user: string; env: Environment; log: boolean; signal: AbortSignal }; export type ExecResult = { output: string; stdout: string; stderr: string; exitCode: number }; diff --git a/packages/testcontainers/src/generic-container/generic-container.test.ts b/packages/testcontainers/src/generic-container/generic-container.test.ts index 805de6c97..a7d047c6b 100644 --- a/packages/testcontainers/src/generic-container/generic-container.test.ts +++ b/packages/testcontainers/src/generic-container/generic-container.test.ts @@ -132,6 +132,18 @@ describe("GenericContainer", { timeout: 180_000 }, () => { expect(output).toEqual(stderr); }); + it("should abort a command when the signal aborts", async () => { + await using container = await new GenericContainer("cristianrgreco/testcontainer:1.1.14") + .withExposedPorts(8080) + .start(); + const controller = new AbortController(); + + const result = container.exec(["sleep", "30"], { signal: controller.signal }); + setTimeout(() => controller.abort(new Error("aborted")), 500); + + await expect(result).rejects.toThrow("aborted"); + }); + it("should capture stdout and stderr in the correct order", async () => { await using container = await new GenericContainer("cristianrgreco/testcontainer:1.1.14") .withExposedPorts(8080) diff --git a/packages/testcontainers/src/generic-container/generic-container.ts b/packages/testcontainers/src/generic-container/generic-container.ts index c38bcac44..7bf23fa93 100644 --- a/packages/testcontainers/src/generic-container/generic-container.ts +++ b/packages/testcontainers/src/generic-container/generic-container.ts @@ -171,7 +171,10 @@ export class GenericContainer implements TestContainer { if (!inspectResult.State.Running) { log.debug("Reused container is not running, attempting to start it"); await client.container.start(container); - inspectResult = await inspectContainerUntilPortsExposed(() => client.container.inspect(container), container.id); + inspectResult = await inspectContainerUntilPortsExposed( + (signal) => client.container.inspect(container, { signal }), + container.id + ); } const mappedInspectResult = mapInspectResult(inspectResult); @@ -211,7 +214,7 @@ export class GenericContainer implements TestContainer { log.info(`Started container for image "${this.createOpts.Image}"`, { containerId: container.id }); const inspectResult = await inspectContainerUntilPortsExposed( - () => client.container.inspect(container), + (signal) => client.container.inspect(container, { signal }), container.id ); const mappedInspectResult = mapInspectResult(inspectResult); diff --git a/packages/testcontainers/src/generic-container/inspect-container-util-ports-exposed.test.ts b/packages/testcontainers/src/generic-container/inspect-container-util-ports-exposed.test.ts index dd43ea47e..959973b62 100644 --- a/packages/testcontainers/src/generic-container/inspect-container-util-ports-exposed.test.ts +++ b/packages/testcontainers/src/generic-container/inspect-container-util-ports-exposed.test.ts @@ -66,6 +66,28 @@ describe("inspectContainerUntilPortsExposed", { concurrent: false }, () => { ); }); + it("throws an error at the timeout when the inspect never resolves", { timeout: 1000 }, async () => { + const inspectFn = vi.fn(() => new Promise(() => {})); + + await expect(inspectContainerUntilPortsExposed(inspectFn, "container-id", 10)).rejects.toThrow( + "Timed out after 10ms while waiting for container ports to be bound to the host" + ); + }); + + it("aborts the inspect at the timeout", async () => { + const signals: AbortSignal[] = []; + const inspectFn = (signal: AbortSignal) => + new Promise((_, reject) => { + signals.push(signal); + signal.addEventListener("abort", () => reject(signal.reason)); + }); + + await expect(inspectContainerUntilPortsExposed(inspectFn, "container-id", 10)).rejects.toThrow( + "Timed out after 10ms while waiting for container ports to be bound to the host" + ); + expect(signals.map((signal) => signal.aborted)).toEqual([true]); + }); + it("throws an error when container ports not exposed within timeout", async () => { const data = mockInspectResult({ "8080/tcp": [] }, {}); const inspectFn = vi.fn().mockResolvedValue(data); diff --git a/packages/testcontainers/src/generic-container/inspect-container-util-ports-exposed.ts b/packages/testcontainers/src/generic-container/inspect-container-util-ports-exposed.ts index c7d1ba1db..5291d511b 100644 --- a/packages/testcontainers/src/generic-container/inspect-container-util-ports-exposed.ts +++ b/packages/testcontainers/src/generic-container/inspect-container-util-ports-exposed.ts @@ -1,32 +1,30 @@ import type { ContainerInspectInfo } from "dockerode"; -import { IntervalRetry, log } from "../common"; +import { log, poll } from "../common"; export async function inspectContainerUntilPortsExposed( - inspectFn: () => Promise, + inspectFn: (signal: AbortSignal) => Promise, containerId: string, timeout = 30_000 ): Promise { - const result = await new IntervalRetry(250).retryUntil( - () => inspectFn(), - (inspectResult) => { - const portBindings = inspectResult?.HostConfig?.PortBindings; - if (!portBindings) { - return true; - } - const expectedlyBoundPorts = Object.keys(portBindings); - return expectedlyBoundPorts.every((exposedPort) => inspectResult.NetworkSettings.Ports[exposedPort]?.length > 0); - }, - () => { - const message = `Timed out after ${timeout}ms while waiting for container ports to be bound to the host`; - log.error(message, { containerId }); - return new Error(message); - }, - timeout - ); - - if (result instanceof Error) { - throw result; + try { + return await poll(({ signal }) => inspectFn(signal), { + until: (inspectResult) => { + const portBindings = inspectResult?.HostConfig?.PortBindings; + if (!portBindings) { + return true; + } + const expectedlyBoundPorts = Object.keys(portBindings); + return expectedlyBoundPorts.every( + (exposedPort) => inspectResult.NetworkSettings.Ports[exposedPort]?.length > 0 + ); + }, + intervalMs: 250, + timeoutMs: timeout, + timeoutMessage: `Timed out after ${timeout}ms while waiting for container ports to be bound to the host`, + retryOnError: false, + }); + } catch (err) { + log.error(`${err}`, { containerId }); + throw err; } - - return result; } diff --git a/packages/testcontainers/src/generic-container/started-generic-container.ts b/packages/testcontainers/src/generic-container/started-generic-container.ts index 6539caf40..4b825fbc9 100644 --- a/packages/testcontainers/src/generic-container/started-generic-container.ts +++ b/packages/testcontainers/src/generic-container/started-generic-container.ts @@ -95,7 +95,7 @@ export class StartedGenericContainer implements StartedTestContainer { await client.container.restart(this.container, resolvedOptions); this.inspectResult = await inspectContainerUntilPortsExposed( - () => client.container.inspect(this.container), + (signal) => client.container.inspect(this.container, { signal }), this.container.id ); const mappedInspectResult = mapInspectResult(this.inspectResult); diff --git a/packages/testcontainers/src/index.ts b/packages/testcontainers/src/index.ts index 247498fb6..a96c3f5f7 100644 --- a/packages/testcontainers/src/index.ts +++ b/packages/testcontainers/src/index.ts @@ -1,4 +1,4 @@ -export { IntervalRetry, log, RandomUuid, Retry, randomUuid, Uuid } from "./common"; +export { IntervalRetry, log, PollOptions, poll, RandomUuid, Retry, randomUuid, Uuid } from "./common"; export { ContainerRuntimeClient, getContainerRuntimeClient, ImageName } from "./container-runtime"; export { DockerComposeEnvironment } from "./docker-compose-environment/docker-compose-environment"; export { DownedDockerComposeEnvironment } from "./docker-compose-environment/downed-docker-compose-environment"; diff --git a/packages/testcontainers/src/reaper/reaper.ts b/packages/testcontainers/src/reaper/reaper.ts index 364981bea..abe824182 100644 --- a/packages/testcontainers/src/reaper/reaper.ts +++ b/packages/testcontainers/src/reaper/reaper.ts @@ -1,7 +1,7 @@ import { Socket } from "node:net"; import { userInfo } from "node:os"; import type { ContainerInfo } from "dockerode"; -import { IntervalRetry, log, RandomUuid, withFileLock } from "../common"; +import { log, poll, RandomUuid, withFileLock } from "../common"; import { type ContainerRuntimeClient, ImageName } from "../container-runtime"; // biome-ignore lint/suspicious/noImportCycles: Ryuk and sshd run as GenericContainers; only dereferenced at runtime import { GenericContainer } from "../generic-container/generic-container"; @@ -133,22 +133,25 @@ async function createNewReaper(sessionId: string, remoteSocketPath: string): Pro } async function connectToReaperSocket(host: string, port: number, containerId: string): Promise { - const retryResult = await new IntervalRetry(1000).retryUntil( - (attempt) => { - return new Promise((resolve) => { - log.debug(`Connecting to Reaper (attempt ${attempt + 1}) on "${host}:${port}"...`, { containerId }); - const socket = new Socket(); + return poll( + ({ attempt, signal }) => { + return new Promise((resolve, reject) => { + log.debug(`Connecting to Reaper (attempt ${attempt}) on "${host}:${port}"...`, { containerId }); + const socket = new Socket({ signal }); socket .unref() .on("timeout", () => log.error(`Reaper ${containerId} socket timed out`)) - .on("error", (err) => log.error(`Reaper ${containerId} socket error: ${err}`)) + .on("error", (err) => { + log.error(`Reaper ${containerId} socket error: ${err}`); + reject(err); + }) .on("close", (hadError) => { if (hadError) { log.error(`Connection to Reaper closed with error`, { containerId }); } else { log.warn(`Connection to Reaper closed`, { containerId }); } - resolve(undefined); + reject(new Error(`Connection to Reaper closed`)); }) .connect(port, host, () => { log.debug(`Connected to Reaper`, { containerId }); @@ -156,19 +159,8 @@ async function connectToReaperSocket(host: string, port: number, containerId: st }); }); }, - (result) => result !== undefined, - () => { - const message = `Failed to connect to Reaper`; - log.error(message, { containerId }); - return new Error(message); - }, - 4000 + { intervalMs: 1000, timeoutMs: 4000, timeoutMessage: `Failed to connect to Reaper` } ); - - if (retryResult instanceof Socket) { - return retryResult; - } - throw retryResult; } class RyukReaper implements Reaper { diff --git a/packages/testcontainers/src/types.ts b/packages/testcontainers/src/types.ts index 28242ebf1..8b2ccaf4a 100644 --- a/packages/testcontainers/src/types.ts +++ b/packages/testcontainers/src/types.ts @@ -91,7 +91,7 @@ export type RegistryConfig = { export type BuildArgs = { [key in string]: string }; -export type ExecOptions = { workingDir: string; user: string; env: Environment }; +export type ExecOptions = { workingDir: string; user: string; env: Environment; signal: AbortSignal }; export type ExecResult = { output: string; stdout: string; stderr: string; exitCode: number }; diff --git a/packages/testcontainers/src/utils/test-helper.ts b/packages/testcontainers/src/utils/test-helper.ts index 57911b019..1c891e765 100644 --- a/packages/testcontainers/src/utils/test-helper.ts +++ b/packages/testcontainers/src/utils/test-helper.ts @@ -6,7 +6,7 @@ import path from "node:path"; import type { Readable } from "node:stream"; import type { GetEventsOptions, ImageInspectInfo } from "dockerode"; import { Agent, request } from "undici"; -import { IntervalRetry } from "../common"; +import { poll } from "../common"; import { getContainerRuntimeClient } from "../container-runtime"; import type { StartedDockerComposeEnvironment } from "../docker-compose-environment/started-docker-compose-environment"; import type { GenericContainer } from "../generic-container/generic-container"; @@ -146,21 +146,12 @@ export async function stopStartingContainer(container: GenericContainer, name: s const client = await getContainerRuntimeClient(); const containerStartPromise = container.start(); - const status = await new IntervalRetry(500).retryUntil( - () => - client.container - .getById(name) - .inspect() - .then((i) => i.State.Running) - .catch(() => false), - (running) => running, - () => false, - 10000 - ); - - if (!status) { - throw Error("failed start container"); - } + await poll(({ signal }) => client.container.getById(name).inspect({ abortSignal: signal }), { + until: (inspectResult) => inspectResult.State.Running, + intervalMs: 500, + timeoutMs: 10000, + timeoutMessage: "failed start container", + }); await client.container.getById(name).stop(); await containerStartPromise; diff --git a/packages/testcontainers/src/wait-strategies/health-check-wait-strategy.ts b/packages/testcontainers/src/wait-strategies/health-check-wait-strategy.ts index 6066eb6fd..eb8ce2421 100644 --- a/packages/testcontainers/src/wait-strategies/health-check-wait-strategy.ts +++ b/packages/testcontainers/src/wait-strategies/health-check-wait-strategy.ts @@ -1,5 +1,5 @@ import type Dockerode from "dockerode"; -import { IntervalRetry, log } from "../common"; +import { log, poll } from "../common"; import { getContainerRuntimeClient } from "../container-runtime"; import { getHealthCheckStatusFromInspect } from "./utils/health-check"; import { AbstractWaitStrategy } from "./wait-strategy"; @@ -9,16 +9,15 @@ export class HealthCheckWaitStrategy extends AbstractWaitStrategy { log.debug(`Waiting for health check...`, { containerId: container.id }); const client = await getContainerRuntimeClient(); - const status = await new IntervalRetry(100).retryUntil( - async () => getHealthCheckStatusFromInspect(await client.container.inspect(container)), - (healthCheckStatus) => healthCheckStatus === "healthy" || healthCheckStatus === "unhealthy", - () => { - const timeout = this.startupTimeoutMs; - const message = `Health check not healthy after ${timeout}ms`; - log.error(message, { containerId: container.id }); - throw new Error(message); - }, - this.startupTimeoutMs + const status = await poll( + async ({ signal }) => getHealthCheckStatusFromInspect(await client.container.inspect(container, { signal })), + { + until: (healthCheckStatus) => healthCheckStatus === "healthy" || healthCheckStatus === "unhealthy", + intervalMs: 100, + timeoutMs: this.startupTimeoutMs, + timeoutMessage: `Health check not healthy after ${this.startupTimeoutMs}ms`, + retryOnError: false, + } ); if (status !== "healthy") { diff --git a/packages/testcontainers/src/wait-strategies/host-port-wait-strategy.ts b/packages/testcontainers/src/wait-strategies/host-port-wait-strategy.ts index 1cc5695dc..18eead3d9 100644 --- a/packages/testcontainers/src/wait-strategies/host-port-wait-strategy.ts +++ b/packages/testcontainers/src/wait-strategies/host-port-wait-strategy.ts @@ -1,5 +1,5 @@ import type Dockerode from "dockerode"; -import { IntervalRetry, log } from "../common"; +import { log, poll } from "../common"; import { getContainerRuntimeClient } from "../container-runtime"; import type { BoundPorts } from "../utils/bound-ports"; import { HostPortCheck, InternalPortCheck, type PortCheck } from "./utils/port-check"; @@ -30,7 +30,7 @@ export class HostPortWaitStrategy extends AbstractWaitStrategy { continue; } log.debug(`Waiting for host port ${hostPort}...`, { containerId: container.id }); - await this.waitForPort(container, hostPort, portCheck); + await this.waitForPort(hostPort, portCheck); log.debug(`Host port ${hostPort} ready`, { containerId: container.id }); } log.debug(`Host port wait strategy complete`, { containerId: container.id }); @@ -49,26 +49,19 @@ export class HostPortWaitStrategy extends AbstractWaitStrategy { continue; } log.debug(`Waiting for internal port ${internalPort}...`, { containerId: container.id }); - await this.waitForPort(container, internalPort, portCheck); + await this.waitForPort(internalPort, portCheck); log.debug(`Internal port ${internalPort} ready`, { containerId: container.id }); } log.debug(`Internal port wait strategy complete`, { containerId: container.id }); } - private async waitForPort( - container: Dockerode.Container, - port: number | string, - portCheck: PortCheck - ): Promise { - await new IntervalRetry(100).retryUntil( - () => portCheck.isBound(port), - (isBound) => isBound, - () => { - const message = `Port ${port} not bound after ${this.startupTimeoutMs}ms`; - log.error(message, { containerId: container.id }); - throw new Error(message); - }, - this.startupTimeoutMs - ); + private async waitForPort(port: number | string, portCheck: PortCheck): Promise { + await poll(({ signal }) => portCheck.isBound(port, signal), { + until: (isBound) => isBound, + intervalMs: 100, + timeoutMs: this.startupTimeoutMs, + timeoutMessage: `Port ${port} not bound after ${this.startupTimeoutMs}ms`, + retryOnError: false, + }); } } diff --git a/packages/testcontainers/src/wait-strategies/http-wait-strategy.ts b/packages/testcontainers/src/wait-strategies/http-wait-strategy.ts index d0a8e5332..f1693f6ff 100644 --- a/packages/testcontainers/src/wait-strategies/http-wait-strategy.ts +++ b/packages/testcontainers/src/wait-strategies/http-wait-strategy.ts @@ -1,7 +1,7 @@ import { text } from "node:stream/consumers"; import type Dockerode from "dockerode"; import { Agent, request } from "undici"; -import { IntervalRetry, log } from "../common"; +import { log, poll } from "../common"; import { getContainerRuntimeClient } from "../container-runtime"; import type { BoundPorts } from "../utils/bound-ports"; import { undiciResponseToFetchResponse } from "./utils/undici-response-parser"; @@ -85,40 +85,35 @@ export class HttpWaitStrategy extends AbstractWaitStrategy { const agent = this.createInsecureAgent(); try { - await new IntervalRetry(this.readTimeoutMs).retryUntil( - async () => { - try { - const url = `${this.protocol}://${client.info.containerRuntime.host}:${boundPorts.getBinding(this.port)}${ - this.path - }`; - - if (abortOnContainerExit) { - const containerStatus = (await client.container.inspect(container)).State.Status; - - if (containerStatus === exitStatus) { - containerExited = true; - return; - } + await poll( + async ({ signal }) => { + const url = `${this.protocol}://${client.info.containerRuntime.host}:${boundPorts.getBinding(this.port)}${ + this.path + }`; + + if (abortOnContainerExit) { + const containerStatus = (await client.container.inspect(container, { signal })).State.Status; + + if (containerStatus === exitStatus) { + containerExited = true; + return undefined; } - - return await this.sendRequest(url, agent); - } catch { - return undefined; - } - }, - async (response) => { - if (abortOnContainerExit && containerExited) { - return true; } - return this.matchesPredicates(response); + return this.sendRequest(url, agent, signal); }, - () => { - const message = `URL ${this.path} not accessible after ${this.startupTimeoutMs}ms`; - log.error(message, { containerId: container.id }); - throw new Error(message); - }, - this.startupTimeoutMs + { + until: async (response) => { + if (abortOnContainerExit && containerExited) { + return true; + } + + return this.matchesPredicates(response); + }, + intervalMs: this.readTimeoutMs, + timeoutMs: this.startupTimeoutMs, + timeoutMessage: `URL ${this.path} not accessible after ${this.startupTimeoutMs}ms`, + } ); } finally { // Force-close: status-only predicates never read the body, so a graceful close() could @@ -133,11 +128,11 @@ export class HttpWaitStrategy extends AbstractWaitStrategy { log.debug(`HTTP wait strategy complete`, { containerId: container.id }); } - private async sendRequest(url: string, agent: Agent | undefined): Promise { + private async sendRequest(url: string, agent: Agent | undefined, signal: AbortSignal): Promise { return undiciResponseToFetchResponse( await request(url, { method: this.method, - signal: AbortSignal.timeout(this.readTimeoutMs), + signal: AbortSignal.any([signal, AbortSignal.timeout(this.readTimeoutMs)]), headers: this.headers, dispatcher: agent, }) diff --git a/packages/testcontainers/src/wait-strategies/one-shot-startup-startegy.ts b/packages/testcontainers/src/wait-strategies/one-shot-startup-startegy.ts index 52b3be919..2415ebf1d 100644 --- a/packages/testcontainers/src/wait-strategies/one-shot-startup-startegy.ts +++ b/packages/testcontainers/src/wait-strategies/one-shot-startup-startegy.ts @@ -17,8 +17,12 @@ export class OneShotStartupCheckStrategy extends StartupCheckStrategy { return this.isDockerTimestampNonEmpty(state.StartedAt) && this.isDockerTimestampNonEmpty(state.FinishedAt); } - public async checkStartupState(dockerClient: Dockerode, containerId: string): Promise { - const info = await dockerClient.getContainer(containerId).inspect(); + public async checkStartupState( + dockerClient: Dockerode, + containerId: string, + signal?: AbortSignal + ): Promise { + const info = await dockerClient.getContainer(containerId).inspect(signal && { abortSignal: signal }); if (!this.isContainerStopped(info)) { return "PENDING"; diff --git a/packages/testcontainers/src/wait-strategies/shell-wait-strategy.test.ts b/packages/testcontainers/src/wait-strategies/shell-wait-strategy.test.ts index 026a6eb5e..e58068acb 100644 --- a/packages/testcontainers/src/wait-strategies/shell-wait-strategy.test.ts +++ b/packages/testcontainers/src/wait-strategies/shell-wait-strategy.test.ts @@ -17,4 +17,13 @@ describe("ShellWaitStrategy", { timeout: 180_000 }, () => { .start() ).rejects.toThrowError(`Shell command "stat /tmp/test.lock" not successful`); }); + + it("should timeout when command is still running", async () => { + await expect(() => + new GenericContainer("cristianrgreco/testcontainer:1.1.14") + .withWaitStrategy(Wait.forSuccessfulCommand("sleep 30")) + .withStartupTimeout(1000) + .start() + ).rejects.toThrowError(`Shell command "sleep 30" not successful after 1000ms`); + }); }); diff --git a/packages/testcontainers/src/wait-strategies/shell-wait-strategy.ts b/packages/testcontainers/src/wait-strategies/shell-wait-strategy.ts index 70101e535..ee463abed 100644 --- a/packages/testcontainers/src/wait-strategies/shell-wait-strategy.ts +++ b/packages/testcontainers/src/wait-strategies/shell-wait-strategy.ts @@ -1,5 +1,5 @@ import type Dockerode from "dockerode"; -import { IntervalRetry, log } from "../common"; +import { log, poll } from "../common"; import { getContainerRuntimeClient } from "../container-runtime"; import { AbstractWaitStrategy } from "./wait-strategy"; @@ -12,20 +12,21 @@ export class ShellWaitStrategy extends AbstractWaitStrategy { log.debug(`Waiting for successful shell command "${this.command}"...`, { containerId: container.id }); const client = await getContainerRuntimeClient(); - await new IntervalRetry(100).retryUntil( - async () => { + await poll( + async ({ signal }) => { const { exitCode } = await client.container.exec(container, ["/bin/sh", "-c", this.command], { log: false, + signal, }); return exitCode; }, - (exitCode) => exitCode === 0, - () => { - const message = `Shell command "${this.command}" not successful after ${this.startupTimeoutMs}ms`; - log.error(message, { containerId: container.id }); - throw new Error(message); - }, - this.startupTimeoutMs + { + until: (exitCode) => exitCode === 0, + intervalMs: 100, + timeoutMs: this.startupTimeoutMs, + timeoutMessage: `Shell command "${this.command}" not successful after ${this.startupTimeoutMs}ms`, + retryOnError: false, + } ); log.debug(`Shell wait strategy complete`, { containerId: container.id }); diff --git a/packages/testcontainers/src/wait-strategies/startup-check-strategy.test.ts b/packages/testcontainers/src/wait-strategies/startup-check-strategy.test.ts index 0def1a895..98ab774d6 100644 --- a/packages/testcontainers/src/wait-strategies/startup-check-strategy.test.ts +++ b/packages/testcontainers/src/wait-strategies/startup-check-strategy.test.ts @@ -1,3 +1,4 @@ +import type Dockerode from "dockerode"; import { GenericContainer } from "../generic-container/generic-container"; import { StartupCheckStrategy, type StartupStatus } from "./startup-check-strategy"; @@ -35,6 +36,29 @@ describe("StartupCheckStrategy", { timeout: 180_000 }, () => { ).rejects.toThrowError("Container not accessible after 3000ms"); }); + it("should abort the signal passed to the startup check when the timeout is reached", async () => { + const signals: (AbortSignal | undefined)[] = []; + const waitStrategy = new (class extends StartupCheckStrategy { + public override async checkStartupState( + _dockerClient: Dockerode, + _containerId: string, + signal?: AbortSignal + ): Promise { + signals.push(signal); + return "PENDING"; + } + })(); + + await expect(() => + new GenericContainer("cristianrgreco/testcontainer:1.1.14") + .withWaitStrategy(waitStrategy) + .withStartupTimeout(1000) + .start() + ).rejects.toThrowError("Container not accessible after 1000ms"); + expect(signals.length).toBeGreaterThan(0); + expect(signals.every((signal) => signal?.aborted)).toBe(true); + }); + it("should fail immediately when status FAILED", async () => { const waitStrategy = new (class extends StartupCheckStrategy { public count = 0; diff --git a/packages/testcontainers/src/wait-strategies/startup-check-strategy.ts b/packages/testcontainers/src/wait-strategies/startup-check-strategy.ts index 27341e3b1..f67d54334 100644 --- a/packages/testcontainers/src/wait-strategies/startup-check-strategy.ts +++ b/packages/testcontainers/src/wait-strategies/startup-check-strategy.ts @@ -1,30 +1,31 @@ import type Dockerode from "dockerode"; -import { IntervalRetry, log } from "../common"; +import { poll } from "../common"; import { getContainerRuntimeClient } from "../container-runtime"; import { AbstractWaitStrategy } from "./wait-strategy"; export type StartupStatus = "PENDING" | "SUCCESS" | "FAIL"; export abstract class StartupCheckStrategy extends AbstractWaitStrategy { - public abstract checkStartupState(dockerClient: Dockerode, containerId: string): Promise; + public abstract checkStartupState( + dockerClient: Dockerode, + containerId: string, + signal?: AbortSignal + ): Promise; public override async waitUntilReady(container: Dockerode.Container): Promise { const client = await getContainerRuntimeClient(); - const startupStatus = await new IntervalRetry(1000).retryUntil( - async () => await this.checkStartupState(client.container.dockerode, container.id), - (status) => status === "SUCCESS" || status === "FAIL", - () => { - const message = `Container not accessible after ${this.startupTimeoutMs}ms`; - log.error(message, { containerId: container.id }); - return new Error(message); - }, - this.startupTimeoutMs + const startupStatus = await poll( + ({ signal }) => this.checkStartupState(client.container.dockerode, container.id, signal), + { + until: (status) => status === "SUCCESS" || status === "FAIL", + intervalMs: 1000, + timeoutMs: this.startupTimeoutMs, + timeoutMessage: `Container not accessible after ${this.startupTimeoutMs}ms`, + retryOnError: false, + } ); - if (startupStatus instanceof Error) { - throw startupStatus; - } if (startupStatus === "FAIL") { throw new Error(`Container failed to start for ${container.id}`); } diff --git a/packages/testcontainers/src/wait-strategies/utils/port-check.ts b/packages/testcontainers/src/wait-strategies/utils/port-check.ts index 76d2f31d2..0380cbdaf 100644 --- a/packages/testcontainers/src/wait-strategies/utils/port-check.ts +++ b/packages/testcontainers/src/wait-strategies/utils/port-check.ts @@ -4,20 +4,20 @@ import { log } from "../../common"; import type { ContainerRuntimeClient } from "../../container-runtime"; export interface PortCheck { - isBound(port: number | string): Promise; + isBound(port: number | string, signal?: AbortSignal): Promise; } export class HostPortCheck implements PortCheck { constructor(private readonly client: ContainerRuntimeClient) {} - public isBound(port: number | string): Promise { + public isBound(port: number | string, signal?: AbortSignal): Promise { if (typeof port === "string" && port.toLowerCase().endsWith("/udp")) { log.debug(`Skipping host port check for UDP port ${port} (UDP port checks not supported)`); return Promise.resolve(true); } return new Promise((resolve) => { - const socket = new Socket(); + const socket = new Socket({ signal }); const portNumber = typeof port === "string" ? parseInt(port.split("/")[0], 10) : port; socket .setTimeout(1000) @@ -46,7 +46,7 @@ export class InternalPortCheck implements PortCheck { private readonly container: Dockerode.Container ) {} - public async isBound(port: number | string): Promise { + public async isBound(port: number | string, signal?: AbortSignal): Promise { if (typeof port === "string" && port.toLowerCase().includes("/udp")) { log.debug(`Skipping internal port check for UDP port ${port} (UDP port checks not supported)`, { containerId: this.container.id, @@ -63,7 +63,7 @@ export class InternalPortCheck implements PortCheck { ]; const commandResults = await Promise.all( - commands.map((command) => this.client.container.exec(this.container, command, { log: false })) + commands.map((command) => this.client.container.exec(this.container, command, { log: false, signal })) ); const isBound = commandResults.some((result) => result.exitCode === 0); diff --git a/packages/testcontainers/src/wait-strategies/wait-strategy-abort.test.ts b/packages/testcontainers/src/wait-strategies/wait-strategy-abort.test.ts new file mode 100644 index 000000000..186cfb584 --- /dev/null +++ b/packages/testcontainers/src/wait-strategies/wait-strategy-abort.test.ts @@ -0,0 +1,52 @@ +import type { Container } from "dockerode"; +import type { BoundPorts } from "../utils/bound-ports"; +import { Wait } from "./wait"; + +const { signals } = vi.hoisted(() => ({ signals: [] as AbortSignal[] })); + +vi.mock("../container-runtime", async (importOriginal) => { + const pendingUntilAborted = (opts: { signal: AbortSignal }) => + new Promise((_, reject) => { + signals.push(opts.signal); + opts.signal.addEventListener("abort", () => reject(opts.signal.reason)); + }); + + return { + ...(await importOriginal()), + getContainerRuntimeClient: async () => ({ + info: { containerRuntime: { host: "localhost" } }, + container: { + inspect: (_container: Container, opts: { signal: AbortSignal }) => pendingUntilAborted(opts), + exec: (_container: Container, _command: string[], opts: { signal: AbortSignal }) => pendingUntilAborted(opts), + }, + }), + }; +}); + +const container = { id: "container-id" } as Container; +const boundPorts = { + getBinding: () => 8080, + iterator: () => [["8080/tcp", 8080]], +} as unknown as BoundPorts; + +describe("Aborting Docker calls at the startup timeout", { concurrent: false }, () => { + beforeEach(() => { + signals.length = 0; + }); + + it.for([ + ["health check", Wait.forHealthCheck(), "Health check not healthy after 50ms"], + [ + "shell command", + Wait.forSuccessfulCommand("stat /tmp/app.lock"), + `Shell command "stat /tmp/app.lock" not successful`, + ], + ["listening ports", Wait.forListeningPorts(), /Port 8080(\/tcp)? not bound after 50ms/], + ["HTTP", Wait.forHttp("/", 8080, { abortOnContainerExit: true }), "URL / not accessible after 50ms"], + ] as const)("should abort the Docker call of the %s wait strategy", async ([, waitStrategy, message]) => { + await expect(waitStrategy.withStartupTimeout(50).waitUntilReady(container, boundPorts)).rejects.toThrow(message); + + expect(signals.length).toBeGreaterThan(0); + await vi.waitFor(() => expect(signals.every((signal) => signal.aborted)).toBe(true)); + }); +});