diff --git a/.changeset/lazy-servers-reset.md b/.changeset/lazy-servers-reset.md new file mode 100644 index 0000000..bee58f3 --- /dev/null +++ b/.changeset/lazy-servers-reset.md @@ -0,0 +1,5 @@ +--- +'prool': patch +--- + +Added lazy Vitest server setup with worker-scoped controls and fixed lifecycle teardown races. diff --git a/README.md b/README.md index 0597b44..eea6471 100644 --- a/README.md +++ b/README.md @@ -310,7 +310,9 @@ const instance_3 = await pool.start(3) ### Vitest -`Pool.setup` starts one instance per Vitest Node worker and runs `setup` with the instances and Vitest project. `Pool.get` selects the current worker's value from a provided array. Configure the global setup on each project that uses the instances. +`Server.setup` starts one lazy keyed proxy for a Vitest project. Each worker can +use `Server.get` to address, reset, or restart its own instance. `Pool.setup` +eagerly starts one direct instance per worker instead. #### Config @@ -320,11 +322,46 @@ import { defineConfig } from 'vitest/config' export default defineConfig({ test: { globalSetup: './test/setup.global.ts', + setupFiles: './test/setup.ts', }, }) ``` -#### Global setup +#### Lazy Server + +```ts +import { Instance } from 'prool' +import { Server } from 'prool/vitest' +import type { TestProject } from 'vitest/node' + +declare module 'vitest' { + export interface ProvidedContext { + anvil: Server.Context + } +} + +export default Server.setup({ + instance: Instance.anvil(), + setup(server, project: TestProject) { + project.provide('anvil', server) + }, +}) +``` + +```ts +import { Server } from 'prool/vitest' +import { inject } from 'vitest' + +const anvil = Server.get(inject('anvil')) +await anvil.reset({ signal: AbortSignal.timeout(30_000) }) + +export const rpcUrl = anvil.url +``` + +`reset` destroys only that worker's instance, and its next request starts a +fresh one. `restart` retains the pooled instance and endpoint. + +#### Eager Pool ```ts import { Instance } from 'prool' @@ -348,8 +385,6 @@ export default Pool.setup({ }) ``` -#### Worker - ```ts import { Pool } from 'prool/vitest' import { inject } from 'vitest' diff --git a/src/Pool.test.ts b/src/Pool.test.ts index 6eb19af..3d01d63 100644 --- a/src/Pool.test.ts +++ b/src/Pool.test.ts @@ -89,6 +89,178 @@ test('enforces the instance limit across concurrent starts', async () => { await limitedPool.destroyAll() }) +test('destroys an instance while it is starting', async () => { + const starting = Promise.withResolvers() + const release = Promise.withResolvers() + let stops = 0 + const instance = Instance.define(() => ({ + host: 'localhost', + name: 'foo', + port: 3000, + async start() { + starting.resolve() + await release.promise + }, + async stop() { + stops++ + }, + }))() + const pool = Pool.define({ instance }) + + const start = pool.start(1) + await starting.promise + const destroy = pool.destroy(1) + release.resolve() + await Promise.all([start, destroy]) + + expect(stops).toBe(1) + expect(pool.size).toBe(0) +}) + +test('destroys pending starts and rejects new ones in destroyAll', async () => { + const starting = Promise.withResolvers() + const release = Promise.withResolvers() + let stops = 0 + const instance = Instance.define(() => ({ + host: 'localhost', + name: 'foo', + port: 3000, + async start() { + starting.resolve() + await release.promise + }, + async stop() { + stops++ + }, + }))() + const pool = Pool.define({ instance }) + + const start = pool.start(1) + await starting.promise + const destroy = pool.destroyAll() + await expect(pool.start(2)).rejects.toThrowError( + 'Cannot start an instance while destroying the pool.', + ) + release.resolve() + await Promise.all([start, destroy]) + + expect(stops).toBe(1) + expect(pool.size).toBe(0) +}) + +test('starts a fresh instance after an in-progress destroy', async () => { + const stopping = Promise.withResolvers() + const release = Promise.withResolvers() + let starts = 0 + let firstStop = true + const instance = Instance.define(() => ({ + host: 'localhost', + name: 'foo', + port: 3000, + async start() { + starts++ + }, + async stop() { + if (!firstStop) return + firstStop = false + stopping.resolve() + await release.promise + }, + }))() + const pool = Pool.define({ instance }) + + const first = await pool.start(1) + const destroy = pool.destroy(1) + await stopping.promise + const start = pool.start(1) + release.resolve() + await destroy + const second = await start + + expect(second).not.toBe(first) + expect(starts).toBe(2) + expect(pool.size).toBe(1) + await pool.destroyAll() +}) + +test('rejects a pending start when destroyAll joins its destroy', async () => { + const stopping = Promise.withResolvers() + const release = Promise.withResolvers() + let firstStop = true + const instance = Instance.define(() => ({ + host: 'localhost', + name: 'foo', + port: 3000, + async start() {}, + async stop() { + if (!firstStop) return + firstStop = false + stopping.resolve() + await release.promise + }, + }))() + const pool = Pool.define({ instance }) + + await pool.start(1) + const destroy = pool.destroy(1) + await stopping.promise + const start = pool.start(1) + const destroyAll = pool.destroyAll() + release.resolve() + + await expect(start).rejects.toThrowError( + 'Cannot start an instance while destroying the pool.', + ) + await Promise.all([destroy, destroyAll]) + expect(pool.size).toBe(0) +}) + +test('waits for every destroyAll failure before accepting starts', async () => { + const failed = Promise.withResolvers() + const stopping = Promise.withResolvers() + const release = Promise.withResolvers() + const failures = new Set() + const pool = Pool.define({ + instance: (key) => + Instance.define(() => ({ + host: 'localhost', + name: 'foo', + port: 3000, + async start() {}, + async stop() { + if (failures.has(key)) return + failures.add(key) + if (key === 1) { + failed.resolve() + throw new Error('stop 1 failed') + } + stopping.resolve() + await release.promise + throw new Error('stop 2 failed') + }, + }))(), + }) + + await pool.start(1) + await pool.start(2) + const destroy = pool.destroyAll() + await Promise.all([failed.promise, stopping.promise]) + await expect(pool.start(3)).rejects.toThrowError( + 'Cannot start an instance while destroying the pool.', + ) + release.resolve() + + const error = await destroy.catch((error) => error) + expect(error).toBeInstanceOf(AggregateError) + expect(error.errors.map((error: Error) => error.message)).toEqual([ + 'stop 1 failed', + 'stop 2 failed', + ]) + + await pool.destroyAll() + expect(pool.size).toBe(0) +}) + describe('create', () => { function instance( parameters: { diff --git a/src/Pool.ts b/src/Pool.ts index 1038670..fbd5820 100644 --- a/src/Pool.ts +++ b/src/Pool.ts @@ -233,7 +233,14 @@ export function define< promises.destroy.set(key, resolver.promise) - this.stop(key) + const operation = (async () => { + await Promise.allSettled([ + promises.restart.get(key), + promises.start.get(key), + ]) + await this.stop(key) + })() + operation .then(() => { instances.delete(key) promises.destroy.delete(key) @@ -253,19 +260,28 @@ export function define< promises.destroyAll = resolver.promise - Promise.all([...instances.keys()].map((key) => this.destroy(key))) - .then(() => { + const keys = new Set([...instances.keys(), ...creating]) + Promise.allSettled([...keys].map((key) => this.destroy(key))).then( + (results) => { + const errors = results.flatMap((result) => + result.status === 'rejected' ? [result.reason] : [], + ) promises.destroyAll = undefined - resolver.resolve() - }) - .catch((error) => { - promises.destroyAll = undefined - resolver.reject(error) - }) + if (errors.length === 0) resolver.resolve() + else if (errors.length === 1) resolver.reject(errors[0]) + else + resolver.reject( + new AggregateError(errors, 'Failed to destroy pool.'), + ) + }, + ) return resolver.promise }, async restart(key) { + const destroyPromise = promises.destroy.get(key) + if (destroyPromise) await destroyPromise + const restartPromise = promises.restart.get(key) if (restartPromise) return restartPromise @@ -290,6 +306,15 @@ export function define< return resolver.promise }, async start(key, options = {}) { + if (promises.destroyAll) + throw new Error('Cannot start an instance while destroying the pool.') + + const destroyPromise = promises.destroy.get(key) + if (destroyPromise) await destroyPromise + + if (promises.destroyAll) + throw new Error('Cannot start an instance while destroying the pool.') + const startPromise = promises.start.get(key) if (startPromise) return startPromise diff --git a/src/vitest/Server.test.ts b/src/vitest/Server.test.ts new file mode 100644 index 0000000..cc8d5d6 --- /dev/null +++ b/src/vitest/Server.test.ts @@ -0,0 +1,202 @@ +import { Instance } from 'prool' +import { Server } from 'prool/vitest' +import { afterEach, describe, expect, expectTypeOf, test, vi } from 'vitest' + +afterEach(() => { + vi.unstubAllEnvs() +}) + +describe('get', () => { + test('controls the current worker instance', async () => { + vi.stubEnv('VITEST_POOL_ID', '2') + const started: number[] = [] + const stopped: number[] = [] + const { context, project } = testProject(3) + const setup = Server.setup({ + instance: (id) => + Instance.define(() => ({ + host: 'localhost', + name: `worker-${id}`, + port: 3000 + id, + async start() { + started.push(id) + }, + async stop() { + stopped.push(id) + }, + }))(), + setup(server, project) { + expectTypeOf(server).toEqualTypeOf() + project.provide('server', server) + }, + }) + const teardown = await setup(project) + const server = Server.get(context.get('server') as Server.Context) + + expect(server.url).toMatch(/^http:\/\/127\.0\.0\.1:\d+\/2$/) + expect(started).toEqual([]) + + await fetch(`${server.url}/start`, { method: 'POST' }).then((response) => + response.text(), + ) + expect(started).toEqual([2]) + + await server.restart() + expect(started).toEqual([2, 2]) + expect(stopped).toEqual([2]) + + await server.reset({ signal: AbortSignal.timeout(1_000) }) + expect(stopped).toEqual([2, 2]) + + await fetch(`${server.url}/start`, { method: 'POST' }).then((response) => + response.text(), + ) + expect(started).toEqual([2, 2, 2]) + + await teardown() + expect(stopped).toEqual([2, 2, 2]) + }) + + test('reports control failures', async () => { + vi.stubEnv('VITEST_POOL_ID', '1') + let stops = 0 + const { context, project } = testProject(1) + const setup = Server.setup({ + instance: Instance.define(() => ({ + host: 'localhost', + name: 'worker', + port: 3000, + async start() {}, + async stop() { + stops++ + if (stops === 1) throw new Error('stop failed') + }, + }))(), + setup(server, project) { + project.provide('server', server) + }, + }) + const teardown = await setup(project) + const server = Server.get(context.get('server') as Server.Context) + await fetch(`${server.url}/start`, { method: 'POST' }).then((response) => + response.text(), + ) + + await expect(server.restart()).rejects.toThrowError( + 'Failed to restart Vitest server: {"message":"stop failed"}', + ) + + await teardown() + }) +}) + +describe('setup', () => { + test('limits instances to the worker count', async () => { + const { context, project } = testProject(1) + const setup = Server.setup({ + instance: Instance.define(() => ({ + host: 'localhost', + name: 'worker', + port: 3000, + async start() {}, + async stop() {}, + }))(), + setup(server, project) { + project.provide('server', server) + }, + }) + const teardown = await setup(project) + const server = context.get('server') as Server.Context + + const first = await fetch(`${server.url}/1/start`, { method: 'POST' }) + expect(first.status).toBe(200) + await first.text() + + const second = await fetch(`${server.url}/2/start`, { method: 'POST' }) + expect(second.status).toBe(400) + expect(await second.json()).toEqual({ + message: 'Instance limit of 1 reached.', + }) + + await teardown() + }) + + test('stops the server when setup fails', async () => { + const stopped: number[] = [] + const setup = Server.setup({ + instance: (id) => + Instance.define(() => ({ + host: 'localhost', + name: `worker-${id}`, + port: 3000 + id, + async start() {}, + async stop() { + stopped.push(id) + }, + }))(), + async setup(server) { + await fetch(`${server.url}/1/start`, { method: 'POST' }).then( + (response) => response.text(), + ) + throw new Error('setup failed') + }, + }) + + await expect(setup(testProject(2).project)).rejects.toThrowError( + 'setup failed', + ) + expect(stopped).toEqual([1]) + }) + + test('reports setup and teardown failures', async () => { + const setup = Server.setup({ + instance: Instance.define(() => ({ + host: 'localhost', + name: 'worker', + port: 3000, + async start() {}, + async stop() { + throw new Error('stop failed') + }, + }))(), + async setup(server) { + await fetch(`${server.url}/1/start`, { method: 'POST' }).then( + (response) => response.text(), + ) + throw new Error('setup failed') + }, + }) + + const error = await setup(testProject(1).project).catch((error) => error) + + expect(error).toBeInstanceOf(AggregateError) + expect(error.errors.map((error: Error) => error.message)).toEqual([ + 'setup failed', + 'stop failed', + ]) + }) + + test('requires a positive worker count', async () => { + const setup = Server.setup({ + instance: Instance.anvil(), + setup() {}, + }) + + await expect(setup(testProject(0).project)).rejects.toThrowError( + 'Vitest maxWorkers must be a positive integer.', + ) + }) +}) + +function testProject(maxWorkers: number) { + const context = new Map() + return { + context, + project: { + config: { maxWorkers }, + provide(key: string, value: unknown) { + context.set(key, value) + }, + }, + } +} diff --git a/src/vitest/Server.ts b/src/vitest/Server.ts new file mode 100644 index 0000000..5d295a6 --- /dev/null +++ b/src/vitest/Server.ts @@ -0,0 +1,108 @@ +import type { Instance } from '../Instance.js' +import * as ProolServer from '../Server.js' +import * as Pool from './Pool.js' + +/** Serializable server context passed from Vitest global setup. */ +export type Context = { + readonly url: string +} + +/** Options for a worker-scoped server control. */ +export type ControlOptions = { + signal?: AbortSignal | undefined +} + +/** Worker-scoped server URL and lifecycle controls. */ +export type Server = { + readonly url: string + /** Destroys the instance so the next request starts a fresh one. */ + reset(options?: ControlOptions | undefined): Promise + /** Restarts the current instance in place. */ + restart(options?: ControlOptions | undefined): Promise +} + +/** Returns the server URL and controls for the current Vitest pool. */ +export function get(context: Context): Server { + const url = `${context.url.replace(/\/+$/, '')}/${Pool.poolId()}` + return { + url, + reset: (options) => control(url, 'destroy', 'reset', options), + restart: (options) => control(url, 'restart', 'restart', options), + } +} + +/** Creates Vitest global setup with a lazy keyed instance server. */ +export function setup< + instance extends Instance = Instance, + project extends setup.Project = setup.Project, +>(parameters: setup.Parameters): setup.ReturnType { + return async (project) => { + const { maxWorkers } = project.config + if (!Number.isSafeInteger(maxWorkers) || maxWorkers < 1) + throw new Error('Vitest maxWorkers must be a positive integer.') + + const { setup: setup_, ...serverParameters } = parameters + const server = ProolServer.create({ + ...serverParameters, + host: serverParameters.host ?? '127.0.0.1', + limit: maxWorkers, + }) + await server.start() + + const address = server.address()! + const host = address.address.includes(':') + ? `[${address.address}]` + : address.address + const context = { url: `http://${host}:${address.port}` } + + try { + await setup_(context, project) + return () => server.stop() + } catch (error) { + try { + await server.stop() + } catch (stopError) { + throw new AggregateError( + [error, stopError], + 'Failed to set up or stop Vitest server.', + ) + } + throw error + } + } +} + +export declare namespace setup { + /** Options for setting up a lazy Vitest instance server. */ + export type Parameters< + instance extends Instance = Instance, + project extends Project = Project, + > = Omit, 'limit'> & { + /** Configures serializable context provided to every worker. */ + setup(context: Context, project: project): Promise | void + } + + /** Minimal Vitest project interface used by global setup. */ + export type Project = Pool.setup.Project + + /** Vitest global setup function. */ + export type ReturnType = + Pool.setup.ReturnType +} + +async function control( + url: string, + path: 'destroy' | 'restart', + action: 'reset' | 'restart', + options: ControlOptions = {}, +) { + const response = await fetch(`${url}/${path}`, { + method: 'POST', + ...(options.signal ? { signal: options.signal } : {}), + }) + const body = await response.text() + if (!response.ok) + throw new Error( + `Failed to ${action} Vitest server${body ? `: ${body}` : '.'}`, + ) +} diff --git a/src/vitest/index.ts b/src/vitest/index.ts index 17f8eb1..ac2804d 100644 --- a/src/vitest/index.ts +++ b/src/vitest/index.ts @@ -1 +1,2 @@ export * as Pool from './Pool.js' +export * as Server from './Server.js'