feat: add leased instance pools

.changeset/tidy-pools-lease.md

@@ -0,0 +1,5 @@
+---
+'prool': patch
+---
+
+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.

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,38 @@ 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(),
+})
+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 concurrent leases. Defaults to half the available logical CPUs. | `number`     |
+| `reset`    | Resets an instance before its next lease. | `(instance: Instance) => Promise<void> \| 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 +314,4 @@ const instance_3 = await pool.start(3)
 
 ## License
 
-[MIT](/LICENSE) License
+[MIT](/LICENSE) License

src/Pool.test.ts

@@ -1,28 +1,30 @@
+import * as os from 'node:os'
 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<typeof Pool.define> | 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 +62,194 @@ test('preserves named endpoint types', async () => {
   await namedPool.destroyAll()
 })
 
+test('enforces the instance limit across concurrent starts', async () => {
+  const started = Promise.withResolvers<void>()
+  const release = Promise.withResolvers<void>()
+  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('defaults to half the available logical CPUs', async () => {
+    const starts: number[] = []
+    const leasePool = Pool.create({
+      instance: instance({ start: (id) => starts.push(id) }),
+    })
+    const limit = Math.max(1, Math.floor(os.availableParallelism() / 2))
+
+    const leases = await Promise.all(
+      Array.from({ length: limit }, () => leasePool.acquire()),
+    )
+    const waiting = leasePool.acquire()
+    let acquired = false
+    waiting.then(() => {
+      acquired = true
+    })
+    await Promise.resolve()
+
+    expect(acquired).toBe(false)
+    expectTypeOf(
+      leases[0]!.instance.endpoints.metrics.protocol,
+    ).toEqualTypeOf<'http'>()
+
+    await leases[0]!.release()
+    const next = await waiting
+
+    expect(next.instance).toBe(leases[0]!.instance)
+    expect(starts).toHaveLength(limit)
+
+    await Promise.all([
+      next.release(),
+      ...leases.slice(1).map((lease) => lease.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() }) },