From 8aaef84de99fe8dbf45ea499fe3080733ad29f58 Mon Sep 17 00:00:00 2001 From: jxom <7336481+jxom@users.noreply.github.com> Date: Sat, 18 Jul 2026 13:19:55 +1000 Subject: [PATCH 1/2] feat: add leased instance pools --- .changeset/tidy-pools-lease.md | 5 + README.md | 55 ++++++- src/Pool.test.ts | 208 ++++++++++++++++++++++-- src/Pool.ts | 245 +++++++++++++++++++++++++---- src/Server.test.ts | 152 ++++++++++++++++-- src/Server.ts | 145 ++++++++++++++--- src/instances/alto.test.ts | 3 +- src/testcontainers/Instance.ts | 15 +- src/testcontainers/compose.test.ts | 23 +++ 9 files changed, 775 insertions(+), 76 deletions(-) create mode 100644 .changeset/tidy-pools-lease.md diff --git a/.changeset/tidy-pools-lease.md b/.changeset/tidy-pools-lease.md new file mode 100644 index 0000000..9700a21 --- /dev/null +++ b/.changeset/tidy-pools-lease.md @@ -0,0 +1,5 @@ +--- +'prool': patch +--- + +Added waitable instance leases and HTTP broker support, fixed pool lifecycle races and dynamic host binding, and preserved zero-second Docker Compose teardown. diff --git a/README.md b/README.md index b018d80..7454896 100644 --- a/README.md +++ b/README.md @@ -52,6 +52,7 @@ You can also create your own custom instances by using the [`Instance.define` fu - [Reference](#reference) - [`Server.create`](#servercreate) - [`Instance.define`](#instancedefine) + - [`Pool.create`](#poolcreate) - [`Pool.define`](#pooldefine) @@ -162,7 +163,7 @@ await bundlerServer.start() ### `Server.create` -Creates a server that manages a pool of instances via a proxy. +Creates a server for a keyed instance proxy or an exclusive lease pool. #### Usage @@ -191,12 +192,29 @@ await executionServer.start() - `/:key/restart`: Restart instance at `key`. - `/healthcheck`: Healthcheck endpoint. +A lease pool exposes `POST /acquire` and `POST /release/:token` instead: + +```ts +import { Instance, Pool, Server } from 'prool' + +const pool = Pool.create({ + instance: Instance.anvil(), + limit: 2, + async reset(instance) { + // Reset application state before reuse. + }, +}) +const server = Server.create({ pool }) +await server.start() +``` + #### API | Name | Description | Type | | ---------- | -------------------------------------------------------- | --------------------------------------- | | `instance` | Instance for the server. | `Instance \| (key: number) => Instance` | | `limit` | Number of instances that can be instantiated in the pool | `number` | +| `pool` | Exclusive lease pool. | `LeasePool` | | `host` | Host for the server. | `string` | | `port` | Port for the server. | `number` | | returns | Server | `Server.Server` | @@ -232,6 +250,39 @@ const foo = Instance.define((parameters: FooParameters) => { | `fn` | Instance definition. | `DefineInstanceFn` | | returns | Instance. | `Instance` | +### `Pool.create` + +Creates a bounded pool of exclusively leased instances. Acquisitions wait in order when every instance is busy. + +If `reset` fails, the instance is destroyed and the release rejects before the slot becomes available again. + +#### Usage + +```ts +import { Instance, Pool } from 'prool' + +const pool = Pool.create({ + instance: Instance.anvil(), + limit: 2, +}) +const lease = await pool.acquire() +try { + await fetch(`http://${lease.instance.host}:${lease.instance.port}`) +} finally { + await lease.release() +} +await pool.close() +``` + +#### API + +| Name | Description | Type | +| ---------- | ---------------------------------------- | ------------ | +| `instance` | Instance to lease. | `Instance` | +| `limit` | Maximum number of concurrent leases. | `number` | +| `reset` | Resets an instance before its next lease. | `(instance: Instance) => Promise \| void` | +| returns | Exclusive lease pool. | `LeasePool` | + ### `Pool.define` Defines a pool of instances. Instances can be started, cached, and stopped against an identifier. @@ -264,4 +315,4 @@ const instance_3 = await pool.start(3) ## License -[MIT](/LICENSE) License \ No newline at end of file +[MIT](/LICENSE) License diff --git a/src/Pool.test.ts b/src/Pool.test.ts index 17f61a9..e00619d 100644 --- a/src/Pool.test.ts +++ b/src/Pool.test.ts @@ -1,28 +1,29 @@ import getPort from 'get-port' import { Instance, Pool, Server } from 'prool' import { + afterAll, afterEach, - beforeAll, describe, expect, expectTypeOf, test, + vi, } from 'vitest' import { altoOptions } from '../test/utils.js' let pool: ReturnType | undefined -const port = await getPort() - -beforeAll(() => - Server.create({ - instance: Instance.anvil({ - chainId: 1, - forkUrl: process.env['VITE_FORK_URL'] ?? 'https://eth.merkle.io', - }), - port, - }).start(), -) +const executionServer = Server.create({ + instance: Instance.anvil({ + chainId: 1, + forkUrl: + process.env['VITE_FORK_URL'] ?? 'https://ethereum-rpc.publicnode.com', + }), +}) +const stopExecutionServer = await executionServer.start() +const port = executionServer.address()!.port + +afterAll(stopExecutionServer) afterEach(async () => { try { @@ -60,6 +61,189 @@ test('preserves named endpoint types', async () => { await namedPool.destroyAll() }) +test('enforces the instance limit across concurrent starts', async () => { + const started = Promise.withResolvers() + const release = Promise.withResolvers() + const instance = Instance.define(() => ({ + host: 'localhost', + name: 'foo', + port: 3000, + async start() { + started.resolve() + await release.promise + }, + async stop() {}, + }))() + const limitedPool = Pool.define({ instance, limit: 1 }) + + const first = limitedPool.start(1) + await started.promise + const limited = expect(limitedPool.start(2)).rejects.toThrowError( + 'Instance limit of 1 reached.', + ) + + release.resolve() + await limited + await first + await limitedPool.destroyAll() +}) + +describe('create', () => { + function instance( + parameters: { + start?: ((id: number) => void) | undefined + stop?: ((id: number) => void) | undefined + } = {}, + ) { + let id = 0 + return Instance.define(() => { + const value = ++id + return { + endpoints: { + metrics: { + host: 'localhost', + port: 9000 + value, + protocol: 'http' as const, + }, + }, + host: 'localhost', + name: 'foo', + port: 3000 + value, + async start() { + parameters.start?.(value) + }, + async stop() { + parameters.stop?.(value) + }, + } + })() + } + + test('leases and reuses instances', async () => { + const starts: number[] = [] + const leasePool = Pool.create({ + instance: instance({ start: (id) => starts.push(id) }), + limit: 1, + }) + + const first = await leasePool.acquire() + const waiting = leasePool.acquire() + let acquired = false + waiting.then(() => { + acquired = true + }) + await Promise.resolve() + + expect(acquired).toBe(false) + expectTypeOf( + first.instance.endpoints.metrics.protocol, + ).toEqualTypeOf<'http'>() + + await first.release() + const second = await waiting + + expect(second.instance).toBe(first.instance) + expect(starts).toHaveLength(1) + + await second.release() + await leasePool.close() + }) + + test('serves waiters in order after resetting', async () => { + const reset = vi.fn(async () => {}) + const leasePool = Pool.create({ instance: instance(), limit: 1, reset }) + const first = await leasePool.acquire() + const order: number[] = [] + const second = leasePool.acquire().then((lease) => { + order.push(2) + return lease + }) + const third = leasePool.acquire().then((lease) => { + order.push(3) + return lease + }) + + await first.release() + const lease_2 = await second + expect(reset).toHaveBeenCalledWith(first.instance) + expect(order).toEqual([2]) + + await lease_2.release() + const lease_3 = await third + expect(order).toEqual([2, 3]) + + await lease_3.release() + await leasePool.close() + }) + + test('replaces an instance when reset fails', async () => { + const stops: number[] = [] + const reset = vi + .fn(async () => {}) + .mockRejectedValueOnce(new Error('reset failed')) + .mockRejectedValueOnce(new Error('reset failed again')) + const leasePool = Pool.create({ + instance: instance({ stop: (id) => stops.push(id) }), + limit: 1, + reset, + }) + const first = await leasePool.acquire() + + await expect(first.release()).rejects.toThrowError('reset failed') + const second = await leasePool.acquire() + + expect(second.instance).not.toBe(first.instance) + expect(stops).toHaveLength(1) + + await expect(second.release()).rejects.toThrowError('reset failed again') + const third = await leasePool.acquire() + expect(third.instance).not.toBe(second.instance) + expect(stops).toHaveLength(2) + + await third.release() + await leasePool.close() + }) + + test('releases a limit reservation after setup fails', async () => { + let creates = 0 + const source = Instance.define(() => { + creates++ + if (creates === 2) throw new Error('create failed') + return { + host: 'localhost', + name: 'foo', + port: 3000, + async start() {}, + async stop() {}, + } + })() + const limitedPool = Pool.define({ instance: source, limit: 1 }) + + await expect(limitedPool.start(1)).rejects.toThrowError('create failed') + await expect(limitedPool.start(2)).resolves.toBeDefined() + + await limitedPool.destroyAll() + }) + + test('closes pending acquisitions', async () => { + const leasePool = Pool.create({ instance: instance(), limit: 1 }) + const lease = await leasePool.acquire() + const waiting = leasePool.acquire() + + await leasePool.close() + + await expect(waiting).rejects.toThrowError('Pool is closed.') + await expect(leasePool.acquire()).rejects.toThrowError('Pool is closed.') + await lease.release() + }) + + test('requires a positive integer limit', () => { + expect(() => Pool.create({ instance: instance(), limit: 0 })).toThrowError( + 'Pool limit must be a positive integer.', + ) + }) +}) + describe.each([ { instance: Instance.anvil({ port: await getPort() }) }, { instance: Instance.tempo({ port: await getPort() }) }, diff --git a/src/Pool.ts b/src/Pool.ts index 8feb1e6..d0efa4f 100644 --- a/src/Pool.ts +++ b/src/Pool.ts @@ -4,6 +4,17 @@ import type { Instance } from './Instance.js' type StartedInstance = ReturnType +export type Lease = { + instance: StartedInstance + release(): Promise +} + +export type LeasePool = { + acquire(): Promise> + close(): Promise + readonly size: number +} + export type Pool< key extends number | string = number | string, instance extends Instance = Instance, @@ -25,6 +36,152 @@ export type Pool< stopAll(): Promise } +/** + * Creates a pool of exclusively leased instances. + * + * @example + * ```ts + * const pool = Pool.create({ instance: anvil(), limit: 2 }) + * const lease = await pool.acquire() + * try { + * // Use lease.instance. + * } finally { + * await lease.release() + * } + * await pool.close() + * ``` + */ +export function create( + parameters: create.Parameters, +): create.ReturnType { + if (!Number.isSafeInteger(parameters.limit) || parameters.limit < 1) + throw new Error('Pool limit must be a positive integer.') + + const available = Array.from( + { length: parameters.limit }, + (_, index) => index + 1, + ) + const leases = new Set() + const operations = new Set>() + const pool = define({ + instance: parameters.instance, + limit: parameters.limit, + }) + const waiters: PromiseWithResolvers>[] = [] + let closePromise: Promise | undefined + let closed = false + + function rejectWaiters(error: Error) { + for (const waiter of waiters) waiter.reject(error) + waiters.length = 0 + } + + function track(promise: Promise): Promise { + operations.add(promise) + promise.then( + () => operations.delete(promise), + () => operations.delete(promise), + ) + return promise + } + + async function grant(slot: number): Promise> { + const instance_ = pool.get(slot) ?? (await pool.start(slot)) + if (closed) throw new Error('Pool is closed.') + leases.add(slot) + let releasePromise: Promise | undefined + return { + instance: instance_, + release() { + releasePromise ??= track(release(slot, instance_)) + return releasePromise + }, + } + } + + function dispatch(slot: number) { + if (closed) return + const waiter = waiters.shift() + if (!waiter) { + available.push(slot) + return + } + track(grant(slot)).then(waiter.resolve, (error) => { + waiter.reject(error) + dispatch(slot) + }) + } + + async function release(slot: number, instance_: StartedInstance) { + if (!leases.delete(slot) || closed) return + try { + await parameters.reset?.(instance_) + } catch (error) { + try { + await pool.destroy(slot) + } catch (destroyError) { + const failure = new AggregateError( + [error, destroyError], + 'Failed to reset or destroy pooled instance.', + ) + closed = true + rejectWaiters(failure) + throw failure + } + dispatch(slot) + throw error + } + dispatch(slot) + } + + return { + async acquire() { + if (closed) throw new Error('Pool is closed.') + const slot = available.shift() + if (slot === undefined) { + const waiter = Promise.withResolvers>() + waiters.push(waiter) + return waiter.promise + } + try { + return await track(grant(slot)) + } catch (error) { + dispatch(slot) + throw error + } + }, + close() { + if (closePromise) return closePromise + closed = true + rejectWaiters(new Error('Pool is closed.')) + closePromise = (async () => { + await Promise.allSettled([...operations]) + await pool.destroyAll() + })() + return closePromise + }, + get size() { + return pool.size + }, + } +} + +export declare namespace create { + export type Parameters = { + /** Instance to lease. */ + instance: instance + /** Maximum number of concurrent leases. */ + limit: number + /** Resets an instance before it is leased again. */ + reset?: + | ((instance: StartedInstance) => Promise | void) + | undefined + } + + export type ReturnType = + LeasePool +} + /** * Defines an instance pool. Instances can be started, cached, and stopped against an identifier. * @@ -48,6 +205,7 @@ export function define< const { limit } = parameters type Instance_ = StartedInstance + const creating = new Set() const instances = new Map() // Define promise instances for mutators to avoid race conditions, and return @@ -78,9 +236,13 @@ export function define< this.stop(key) .then(() => { instances.delete(key) + promises.destroy.delete(key) resolver.resolve() }) - .catch(resolver.reject) + .catch((error) => { + promises.destroy.delete(key) + resolver.reject(error) + }) return resolver.promise }, @@ -96,7 +258,10 @@ export function define< promises.destroyAll = undefined resolver.resolve() }) - .catch(resolver.reject) + .catch((error) => { + promises.destroyAll = undefined + resolver.reject(error) + }) return resolver.promise }, @@ -113,9 +278,14 @@ export function define< instance_ .restart() - .then(resolver.resolve) - .catch(resolver.reject) - .finally(() => promises.restart.delete(key)) + .then(() => { + promises.restart.delete(key) + resolver.resolve() + }) + .catch((error) => { + promises.restart.delete(key) + resolver.reject(error) + }) return resolver.promise }, @@ -125,27 +295,40 @@ export function define< const resolver = Promise.withResolvers() - if (limit && instances.size >= limit) + const isNew = !instances.has(key) + if (isNew && limit && instances.size + creating.size >= limit) throw new Error(`Instance limit of ${limit} reached.`) promises.start.set(key, resolver.promise) - - const instance = - typeof parameters.instance === 'function' - ? parameters.instance(key) - : parameters.instance - const { port = await getPort() } = options - - const instance_ = - instances.get(key) || (instance.create({ port }) as Instance_) - instance_ - .start() - .then(() => { - instances.set(key, instance_) - resolver.resolve(instance_) - }) - .catch(resolver.reject) - .finally(() => promises.start.delete(key)) + if (isNew) creating.add(key) + + try { + const instance = + typeof parameters.instance === 'function' + ? parameters.instance(key) + : parameters.instance + const { port = await getPort() } = options + + const instance_ = + instances.get(key) || (instance.create({ port }) as Instance_) + instance_ + .start() + .then(() => { + instances.set(key, instance_) + creating.delete(key) + promises.start.delete(key) + resolver.resolve(instance_) + }) + .catch((error) => { + creating.delete(key) + promises.start.delete(key) + resolver.reject(error) + }) + } catch (error) { + creating.delete(key) + promises.start.delete(key) + resolver.reject(error) + } return resolver.promise }, @@ -161,9 +344,14 @@ export function define< promises.stop.set(key, resolver.promise) instance_ .stop() - .then(resolver.resolve) - .catch(resolver.reject) - .finally(() => promises.stop.delete(key)) + .then(() => { + promises.stop.delete(key) + resolver.resolve() + }) + .catch((error) => { + promises.stop.delete(key) + resolver.reject(error) + }) return resolver.promise }, @@ -179,7 +367,10 @@ export function define< promises.stopAll = undefined resolver.resolve() }) - .catch(resolver.reject) + .catch((error) => { + promises.stopAll = undefined + resolver.reject(error) + }) return resolver.promise }, diff --git a/src/Server.test.ts b/src/Server.test.ts index 3258730..26c4da7 100644 --- a/src/Server.test.ts +++ b/src/Server.test.ts @@ -1,20 +1,21 @@ +import { request } from 'node:http' import getPort from 'get-port' -import { Instance, Server } from 'prool' -import { beforeAll, describe, expect, test } from 'vitest' +import { Instance, Pool, Server } from 'prool' +import { afterAll, describe, expect, test, vi } from 'vitest' import { type MessageEvent, WebSocket } from 'ws' import { altoOptions } from '../test/utils.js' -const port = await getPort() - -beforeAll(async () => { - await Server.create({ - instance: Instance.anvil({ - chainId: 1, - forkUrl: process.env['VITE_FORK_URL'] ?? 'https://eth.merkle.io', - }), - port, - }).start() +const executionServer = Server.create({ + instance: Instance.anvil({ + chainId: 1, + forkUrl: + process.env['VITE_FORK_URL'] ?? 'https://ethereum-rpc.publicnode.com', + }), }) +const stopExecutionServer = await executionServer.start() +const port = executionServer.address()!.port + +afterAll(stopExecutionServer) test('request: lifecycle endpoint discovery', async () => { const foo = Instance.define(() => { @@ -71,6 +72,133 @@ test('request: lifecycle endpoint discovery', async () => { await stop() }) +test('request: leases pooled instances', async () => { + let instances = 0 + const reset = vi.fn(async () => {}) + const foo = Instance.define(() => { + const id = ++instances + return { + endpoints: { + database: { + host: 'localhost', + port: 5432, + protocol: 'tcp' as const, + }, + }, + host: 'localhost', + name: 'foo', + port: 3000, + async start(_, { setEndpoint }) { + setEndpoint?.({ host: '127.0.0.1', port: 3000 + id }) + }, + async stop() {}, + } + }) + const pool = Pool.create({ instance: foo(), limit: 1, reset }) + const server = Server.create({ host: '127.0.0.1', pool, port: 0 }) + const stop = await server.start() + expect(server.address()!.address).toBe('127.0.0.1') + const url = `http://localhost:${server.address()!.port}` + + const first = await fetch(`${url}/acquire`, { method: 'POST' }).then( + (response) => response.json(), + ) + expect(first).toMatchObject({ + endpoints: { + database: { port: 5432, protocol: 'tcp' }, + default: { host: '127.0.0.1', protocol: 'http' }, + }, + host: '127.0.0.1', + token: expect.any(String), + }) + + const waiting = fetch(`${url}/acquire`, { method: 'POST' }) + expect( + await Promise.race([ + waiting.then(() => true), + new Promise((resolve) => setTimeout(() => resolve(false), 20)), + ]), + ).toBe(false) + + const released = await fetch(`${url}/release/${first.token}`, { + method: 'POST', + }) + expect(released.status).toBe(204) + + const second = await waiting.then((response) => response.json()) + expect(second.port).toBe(first.port) + expect(second.token).not.toBe(first.token) + expect(reset).toHaveBeenCalledOnce() + + expect( + ( + await fetch(`${url}/release/${second.token}`, { + method: 'POST', + }) + ).status, + ).toBe(204) + expect( + ( + await fetch(`${url}/release/${second.token}`, { + method: 'POST', + }) + ).status, + ).toBe(404) + + await stop() +}) + +test('request: releases a lease when acquisition disconnects', async () => { + const reset = vi.fn(async () => {}) + const foo = Instance.define(() => ({ + host: 'localhost', + name: 'foo', + port: 3000, + async start() {}, + async stop() {}, + })) + const pool = Pool.create({ instance: foo(), limit: 1, reset }) + const first = await pool.acquire() + const acquire = vi.spyOn(pool, 'acquire') + const server = Server.create({ pool }) + const stop = await server.start() + const url = `http://localhost:${server.address()!.port}` + const request_ = request(`${url}/acquire`, { method: 'POST' }) + const disconnected = new Promise((resolve) => { + request_.once('error', () => resolve()) + }) + request_.end() + + await vi.waitFor(() => expect(acquire).toHaveBeenCalledOnce()) + request_.destroy(new Error('disconnected')) + await disconnected + await new Promise((resolve) => setTimeout(resolve, 20)) + await first.release() + await vi.waitFor(() => expect(reset).toHaveBeenCalledTimes(2)) + + const next = await pool.acquire() + await next.release() + await stop() +}) + +test('request: reports lease pool teardown failures', async () => { + const foo = Instance.define(() => ({ + host: 'localhost', + name: 'foo', + port: 3000, + async start() {}, + async stop() { + throw new Error('stop failed') + }, + })) + const pool = Pool.create({ instance: foo(), limit: 1 }) + await pool.acquire() + const server = Server.create({ pool }) + await server.start() + + await expect(server.stop()).rejects.toThrowError('stop failed') +}) + test('request: rejects a TCP default proxy', async () => { const database = Instance.define(() => ({ endpoints: { diff --git a/src/Server.ts b/src/Server.ts index a84fc42..a44740a 100644 --- a/src/Server.ts +++ b/src/Server.ts @@ -18,20 +18,21 @@ const websocketProtocols: Partial> = { wss: 'wss', } +type AddressParameters = { + /** Host to run the server on. */ + host?: string | undefined + /** Port to run the server on. */ + port?: number | undefined +} + export type CreateServerParameters = - Pool.define.Parameters & - ( - | { - /** Host to run the server on. */ - host?: string | undefined - /** Port to run the server on. */ - port: number - } - | { - host?: undefined - port?: undefined - } - ) + Pool.define.Parameters & AddressParameters + +export type CreateLeaseServerParameters = + { + /** Exclusively leased instance pool. */ + pool: Pool.LeasePool + } & AddressParameters export type CreateServerReturnType = Omit< Server, @@ -43,11 +44,11 @@ export type CreateServerReturnType = Omit< } /** - * Creates a server that manages a pool of instances via a proxy. + * Creates a server for a keyed instance proxy or an exclusive lease pool. * * @example * ``` - * import { Instance, Server } from 'prool' + * import { Instance, Pool, Server } from 'prool' * * const server = Server.create({ * instance: Instance.anvil(), @@ -63,12 +64,27 @@ export type CreateServerReturnType = Omit< * // "http://localhost:8545/n/stop" * // "http://localhost:8545/n/restart" * // "http://localhost:8545/healthcheck" + * + * const pool = Pool.create({ instance: Instance.anvil(), limit: 2 }) + * const leaseServer = Server.create({ pool }) + * await leaseServer.start() + * // POST /acquire, then POST /release/:token. * ``` */ export function create( parameters: CreateServerParameters, +): CreateServerReturnType +export function create( + parameters: CreateLeaseServerParameters, +): CreateServerReturnType +export function create( + parameters: + | CreateLeaseServerParameters + | CreateServerParameters, ): CreateServerReturnType { - const { host = '::', instance, limit, port } = parameters + if ('pool' in parameters) return createLeaseServer(parameters) + + const { host, instance, limit, port } = parameters const pool = Pool.define({ instance, limit }) const proxy = createProxyServer({ @@ -187,12 +203,13 @@ export function create( return Object.assign(server as any, { start() { return new Promise<() => Promise>((resolve) => { - if (port) server.listen(port, host, () => resolve(this.stop)) + if (host !== undefined || port !== undefined) + server.listen(port ?? 0, host ?? '::', () => resolve(this.stop)) else server.listen(() => resolve(this.stop)) }) }, async stop() { - await Promise.allSettled([ + await settle([ new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())), ), @@ -202,6 +219,90 @@ export function create( }) } +function createLeaseServer( + parameters: CreateLeaseServerParameters, +): CreateServerReturnType { + const { host, pool, port } = parameters + const leases = new Map>() + const server = createServer_(async (request, response) => { + try { + if (request.method === 'OPTIONS') { + response.writeHead(200, { + 'Access-Control-Allow-Origin': '*', + 'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, OPTIONS', + 'Access-Control-Allow-Headers': 'Content-Type, Authorization', + 'Access-Control-Max-Age': '86400', + }) + response.end() + return + } + + const path = new URL(request.url ?? '/', 'http://localhost').pathname + if (request.method === 'POST' && path === '/acquire') { + let disconnected = false + let lease: Pool.Lease | undefined + let token: string | undefined + function release() { + disconnected = true + if (!lease) return + if (token) leases.delete(token) + void lease.release().catch(() => {}) + } + request.once('aborted', release) + response.once('close', () => { + if (!response.writableFinished) release() + }) + lease = await pool.acquire() + if (disconnected || request.destroyed || response.destroyed) { + await lease.release() + return + } + token = crypto.randomUUID() + leases.set(token, lease) + return done(response, 200, { + ...instanceDescriptor(lease.instance), + token, + }) + } + if (request.method === 'POST' && path.startsWith('/release/')) { + const token = path.slice('/release/'.length) + const lease = leases.get(token) + if (!lease) return done(response, 404) + leases.delete(token) + await lease.release() + return done(response, 204) + } + if (path === '/healthcheck') return done(response, 200) + return done(response, 404) + } catch (error) { + if (response.destroyed) return + return done(response, 400, { message: (error as Error).message }) + } + }) + + return Object.assign(server as any, { + start() { + return new Promise<() => Promise>((resolve) => { + if (host !== undefined || port !== undefined) + server.listen(port ?? 0, host ?? '::', () => resolve(this.stop)) + else server.listen(() => resolve(this.stop)) + }) + }, + async stop() { + try { + await settle([ + new Promise((resolve, reject) => + server.close((error) => (error ? reject(error) : resolve())), + ), + pool.close(), + ]) + } finally { + leases.clear() + } + }, + }) +} + function httpProxyTarget(endpoint: Endpoint) { if (endpoint.protocol !== 'http' && endpoint.protocol !== 'https') throw new Error(`Cannot proxy ${endpoint.protocol} endpoint over HTTP.`) @@ -237,3 +338,11 @@ function done(res: ServerResponse, statusCode: number, json?: unknown) { }) .end(json ? JSON.stringify(json) : undefined) } + +async function settle(promises: readonly Promise[]) { + const errors: unknown[] = [] + for (const result of await Promise.allSettled(promises)) + if (result.status === 'rejected') errors.push(result.reason) + if (errors.length === 1) throw errors[0] + if (errors.length > 1) throw new AggregateError(errors, 'Server stop failed.') +} diff --git a/src/instances/alto.test.ts b/src/instances/alto.test.ts index a7966ad..855ec8e 100644 --- a/src/instances/alto.test.ts +++ b/src/instances/alto.test.ts @@ -18,7 +18,8 @@ const defineInstance = (parameters: Partial = {}) => { beforeAll(() => Instance.anvil({ - forkUrl: process.env['VITE_FORK_URL'] ?? 'https://eth.merkle.io', + forkUrl: + process.env['VITE_FORK_URL'] ?? 'https://ethereum-rpc.publicnode.com', port, }).start(), ) diff --git a/src/testcontainers/Instance.ts b/src/testcontainers/Instance.ts index de192b8..fa9b5cf 100644 --- a/src/testcontainers/Instance.ts +++ b/src/testcontainers/Instance.ts @@ -40,6 +40,7 @@ export function compose< parameters: compose.Parameters, options?: Instance.InstanceOptions, ): Instance.Instance> { + const down = composeDownOptions(parameters.down) const initialEndpoints = Object.fromEntries( Object.entries(parameters.endpoints).flatMap(([name, endpoint]) => endpoint @@ -67,7 +68,7 @@ export function compose< async function stopEnvironment() { if (!environment) return const started = environment - await started.down(parameters.down) + await started.down(down) if (environment === started) environment = undefined } @@ -106,9 +107,7 @@ export function compose< for (const [name, endpoint] of endpoints) applyEndpoint?.(name, endpoint) } catch (error) { - const [result] = await Promise.allSettled([ - started.down(parameters.down), - ]) + const [result] = await Promise.allSettled([started.down(down)]) if (result?.status === 'fulfilled') environment = undefined throw error } @@ -148,7 +147,9 @@ export declare namespace compose { : undefined export type DownOptions = { + /** Removes Compose volumes. */ removeVolumes?: boolean | undefined + /** Grace period in milliseconds. Zero kills containers immediately. */ timeout?: number | undefined } @@ -178,6 +179,12 @@ export declare namespace compose { } } +function composeDownOptions(options: compose.DownOptions | undefined) { + if (options?.timeout !== 0) return options + // Testcontainers omits zero; one millisecond becomes Compose's zero-second grace. + return { ...options, timeout: 1 } +} + /** * Defines an instance backed by a Testcontainers container. * diff --git a/src/testcontainers/compose.test.ts b/src/testcontainers/compose.test.ts index caa5857..5bb04a6 100644 --- a/src/testcontainers/compose.test.ts +++ b/src/testcontainers/compose.test.ts @@ -195,3 +195,26 @@ test('retains the environment when stopping fails', async () => { await instance.stop() expect(down).toHaveBeenCalledTimes(2) }) + +test('forwards a zero-second teardown grace period', async () => { + const down = vi.fn(async () => {}) + const instance = Instance.compose({ + down: { timeout: 0 }, + endpoints: { + default: { container: 'api-1', port: 8080, protocol: 'http' }, + }, + environment: () => + environment({ + down, + host: '127.0.0.1', + port: 8080, + services: () => {}, + }), + name: 'services', + }) + + await instance.start() + await instance.stop() + + expect(down).toHaveBeenCalledWith({ timeout: 1 }) +}) From b8b7409b342907a274a4ac6f5eb71961e7eeec7d Mon Sep 17 00:00:00 2001 From: jxom <7336481+jxom@users.noreply.github.com> Date: Sat, 18 Jul 2026 13:25:31 +1000 Subject: [PATCH 2/2] feat: default lease pool concurrency --- .changeset/tidy-pools-lease.md | 2 +- README.md | 3 +-- src/Pool.test.ts | 24 +++++++++++++++--------- src/Pool.ts | 18 +++++++++--------- 4 files changed, 26 insertions(+), 21 deletions(-) diff --git a/.changeset/tidy-pools-lease.md b/.changeset/tidy-pools-lease.md index 9700a21..34d5d17 100644 --- a/.changeset/tidy-pools-lease.md +++ b/.changeset/tidy-pools-lease.md @@ -2,4 +2,4 @@ 'prool': patch --- -Added waitable instance leases and HTTP broker support, fixed pool lifecycle races and dynamic host binding, and preserved zero-second Docker Compose teardown. +Added waitable instance leases with a default of half the available logical CPUs, HTTP broker support, pool race fixes, and zero-second Docker Compose teardown. diff --git a/README.md b/README.md index 7454896..7ea09f9 100644 --- a/README.md +++ b/README.md @@ -263,7 +263,6 @@ import { Instance, Pool } from 'prool' const pool = Pool.create({ instance: Instance.anvil(), - limit: 2, }) const lease = await pool.acquire() try { @@ -279,7 +278,7 @@ await pool.close() | Name | Description | Type | | ---------- | ---------------------------------------- | ------------ | | `instance` | Instance to lease. | `Instance` | -| `limit` | Maximum number of concurrent leases. | `number` | +| `limit` | Maximum concurrent leases. Defaults to half the available logical CPUs. | `number` | | `reset` | Resets an instance before its next lease. | `(instance: Instance) => Promise \| void` | | returns | Exclusive lease pool. | `LeasePool` | diff --git a/src/Pool.test.ts b/src/Pool.test.ts index e00619d..6eb19af 100644 --- a/src/Pool.test.ts +++ b/src/Pool.test.ts @@ -1,3 +1,4 @@ +import * as os from 'node:os' import getPort from 'get-port' import { Instance, Pool, Server } from 'prool' import { @@ -119,14 +120,16 @@ describe('create', () => { })() } - test('leases and reuses instances', async () => { + test('defaults to half the available logical CPUs', async () => { const starts: number[] = [] const leasePool = Pool.create({ instance: instance({ start: (id) => starts.push(id) }), - limit: 1, }) + const limit = Math.max(1, Math.floor(os.availableParallelism() / 2)) - const first = await leasePool.acquire() + const leases = await Promise.all( + Array.from({ length: limit }, () => leasePool.acquire()), + ) const waiting = leasePool.acquire() let acquired = false waiting.then(() => { @@ -136,16 +139,19 @@ describe('create', () => { expect(acquired).toBe(false) expectTypeOf( - first.instance.endpoints.metrics.protocol, + leases[0]!.instance.endpoints.metrics.protocol, ).toEqualTypeOf<'http'>() - await first.release() - const second = await waiting + await leases[0]!.release() + const next = await waiting - expect(second.instance).toBe(first.instance) - expect(starts).toHaveLength(1) + expect(next.instance).toBe(leases[0]!.instance) + expect(starts).toHaveLength(limit) - await second.release() + await Promise.all([ + next.release(), + ...leases.slice(1).map((lease) => lease.release()), + ]) await leasePool.close() }) diff --git a/src/Pool.ts b/src/Pool.ts index d0efa4f..1038670 100644 --- a/src/Pool.ts +++ b/src/Pool.ts @@ -1,3 +1,4 @@ +import * as os from 'node:os' import getPort from 'get-port' import type { Instance } from './Instance.js' @@ -41,7 +42,7 @@ export type Pool< * * @example * ```ts - * const pool = Pool.create({ instance: anvil(), limit: 2 }) + * const pool = Pool.create({ instance: anvil() }) * const lease = await pool.acquire() * try { * // Use lease.instance. @@ -54,18 +55,17 @@ export type Pool< export function create( parameters: create.Parameters, ): create.ReturnType { - if (!Number.isSafeInteger(parameters.limit) || parameters.limit < 1) + const limit = + parameters.limit ?? Math.max(1, Math.floor(os.availableParallelism() / 2)) + if (!Number.isSafeInteger(limit) || limit < 1) throw new Error('Pool limit must be a positive integer.') - const available = Array.from( - { length: parameters.limit }, - (_, index) => index + 1, - ) + const available = Array.from({ length: limit }, (_, index) => index + 1) const leases = new Set() const operations = new Set>() const pool = define({ instance: parameters.instance, - limit: parameters.limit, + limit, }) const waiters: PromiseWithResolvers>[] = [] let closePromise: Promise | undefined @@ -170,8 +170,8 @@ export declare namespace create { export type Parameters = { /** Instance to lease. */ instance: instance - /** Maximum number of concurrent leases. */ - limit: number + /** Maximum concurrent leases. Defaults to half the available logical CPUs. */ + limit?: number | undefined /** Resets an instance before it is leased again. */ reset?: | ((instance: StartedInstance) => Promise | void)