diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index ae2c43f..5d22b90 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -6,7 +6,7 @@ Contributors are encouraged to communicate with maintainers in issues or other c Source files are located in `src/`. These are built to the `dist/` directory. The published package includes `dist/`, `scripts/`, and the regular npm metadata and documentation files. -The `src/index.ts` is the source for the main export. This is the public re-export of all the various utilities from `src/harperLifecycle.ts`, `targz.ts`, and more. The `src/run.ts` is the source for the `harper-integration-test-run` bin script. The internal `src/harperInstanceRegistry.ts` publishes each running Harper instance to a shared on-disk registry, and `src/harperMonitor.ts` is the singleton monitor process that reads it and reaps instances whose test runner died without cleaning up (see README's *Orphaned Instance Monitor* section). A registry record covers a whole process group, so the monitor — the only half that can see when that group is finished — is also the only writer that removes one; the lifecycle side registers and never deregisters. And the `scripts/setup-loopback.sh` is the source for the `harper-integration-test-setup-loopback` bin script. +The `src/index.ts` is the source for the main export. This is the public re-export of all the various utilities from `src/harperLifecycle.ts`, `targz.ts`, and more. The `src/run.ts` is the source for the `harper-integration-test-run` bin script. The internal `src/harperInstanceRegistry.ts` publishes each running Harper instance to a shared on-disk registry, and `src/harperMonitor.ts` is the singleton monitor process that reads it and reaps instances whose test runner died without cleaning up (see README's *Orphaned Instance Monitor* section). A registry record covers a whole process group, so the monitor — the only half that can see when that group is finished — is also the only writer that removes one; the lifecycle side registers and never deregisters. The pool behind the loopback utilities, `src/loopbackAddressPool.ts`, never writes an unusable pool file back as an empty pool before its quarantine ends (see README's *Loopback Pool Utilities*): every path that rewrites the file returns early when `readPoolFile` reports it quarantined (`null`), so a new write path must do the same. And the `scripts/setup-loopback.sh` is the source for the `harper-integration-test-setup-loopback` bin script. The package is `"type": "module"` — all source files are ESM by default. diff --git a/README.md b/README.md index 04f2307..8ef3398 100644 --- a/README.md +++ b/README.md @@ -292,6 +292,16 @@ These are used internally by `startHarper` and `teardownHarper`, but are exporte - `releaseLoopbackAddress(address: string): Promise` - Returns an address to the pool - `releaseAllLoopbackAddressesForCurrentProcess(): Promise` - Releases all addresses held by this process +The pool is one file shared by every process on the machine: `${TMPDIR}/harper-integration-test-loopback-pool.json`, a JSON array with one slot per address holding the PID of the process that reserved it (or `null`), guarded by the `harper-integration-test-loopback-pool.lock` mutex beside it. Updates are published by renaming a complete `*.pending` file over it, so a process killed mid-update leaves the previous pool intact (and possibly an inert `*.pending` file). Releasing an address clears its slot only if this process holds it. + +A pool file that is unparseable, not an array, or an empty array can still be produced by a writer that doesn't publish by rename (an older version of this package killed mid-write) or by an outside edit. Such a file is either about to be completed or has lost reservations whose holders may not have bound their address yet. Allocation does refuse an address that a Harper node is already listening on, but that check cannot see a reservation nothing is listening on yet. So the file is quarantined rather than reset: + +- `getNextAvailableLoopbackAddress()` waits, and both release functions leave the file untouched, until it has gone 6 minutes without modification. That outlasts `startHarper`'s default startup ceiling (2 minutes, 5 under CI), so a `startHarper` that reserved an address before the file broke has bound it or given up by then. +- After that, the pool is reinitialized with every address free and a warning is logged. From then on, an address still in use is protected only by that listening check. +- Deleting the file skips the wait, but only do that once every integration-test run on the machine — including ones using older versions of this package — has stopped. + +The quarantine bounds, rather than eliminates, the risk of handing out an address twice: a caller that holds an address for longer than that without binding it (for example, with a `startupMaxMs` above 5 minutes) is not covered. + ## Node.js Test Runner > Tests executed by this runner must use the `node:test` API. If you're using a different test framework, use the [lifecycle APIs](#api) directly and refer to the [runner source](./src/run.ts) as a reference implementation. diff --git a/package.json b/package.json index d9aef96..c856a4c 100644 --- a/package.json +++ b/package.json @@ -39,7 +39,7 @@ "scripts": { "check": "tsc", "build": "tsc -p tsconfig.build.json", - "test": "node --test \"test/**/*.test.ts\"", + "test": "node --experimental-test-module-mocks --test \"test/**/*.test.ts\"", "prepack": "npm run build" }, "engines": { diff --git a/src/loopbackAddressPool.ts b/src/loopbackAddressPool.ts index 8a95455..b7c107e 100644 --- a/src/loopbackAddressPool.ts +++ b/src/loopbackAddressPool.ts @@ -1,7 +1,8 @@ import { setTimeout as sleep } from 'node:timers/promises'; +import { randomBytes } from 'node:crypto'; import { join } from 'node:path'; import { tmpdir } from 'node:os'; -import { open, readFile, stat, unlink, writeFile } from 'node:fs/promises'; +import { open, readFile, rename, stat, unlink, writeFile } from 'node:fs/promises'; import { createServer } from 'node:net'; import { findAcceptingPort } from './portUtils.ts'; @@ -29,6 +30,11 @@ const HARPER_LOOPBACK_POOL_LOCK_PATH = join(tmpdir(), 'harper-integration-test-l const LOCK_STALE_TIMEOUT_MS = 10000; const RETRY_DELAY_MS = 1000; +// How long an unusable pool file must go unmodified before it is reinitialized. The reservations it +// lost may not be bound yet, which the conflict canary can't see, so this must outlast startHarper's +// reserve-to-bind time: its startup ceiling (DEFAULT_STARTUP_MAX_MS) plus the setup before it. +const UNUSABLE_POOL_QUARANTINE_MS = 360000; + // Port used as a conflict canary when allocating an address. This MUST be a port that // Harper binds WITHOUT SO_REUSEPORT (the operations API — `OPERATIONS_API_PORT` in // harperLifecycle.ts — does exactly that, main-thread-only). Because that port is @@ -202,36 +208,92 @@ async function withLock(callback: () => Promise): Promise { } } +const quarantinedPoolPaths = new Set(); + +function emptyPool(): LoopbackPool { + return Array(HARPER_LOOPBACK_POOL_COUNT).fill(null); +} + /** * Reads the loopback pool from the pool file. The pool is a JSON array where each * index represents a loopback address (127.0.0.2, 127.0.0.3, etc.) and the value * is either null (available) or a process PID (in use). * - * If the file doesn't exist, creates and returns a new empty pool with all addresses - * marked as available (null). + * An unusable file (unparseable, not an array, or an empty array) may still be mid-write by a + * writer that doesn't publish by rename, or may have lost reservations whose holders haven't bound + * their address yet, so it reads as `null` until it has gone UNUSABLE_POOL_QUARANTINE_MS + * unmodified, and only then is replaced with an empty pool. A non-empty array of another length + * is trusted as-is: two processes configured with different counts would otherwise keep wiping + * each other's reservations. * - * @returns The loopback pool array + * @param poolPath The pool file path (overridable for tests; defaults to the shared pool file) + * @returns The loopback pool array, or `null` while the pool file is quarantined */ -async function readPoolFile(): Promise { +export async function readPoolFile(poolPath: string = HARPER_LOOPBACK_POOL_PATH): Promise { + let problem: string; try { - const content = await readFile(HARPER_LOOPBACK_POOL_PATH, 'utf-8'); - return JSON.parse(content) as LoopbackPool; - } catch (error) { - if ((error as NodeJS.ErrnoException).code === 'ENOENT') { - // If the pool file doesn't exist yet, create it with null entries - return Array(HARPER_LOOPBACK_POOL_COUNT).fill(null); + const parsed: unknown = JSON.parse(await readFile(poolPath, 'utf-8')); + if (Array.isArray(parsed) && parsed.length > 0) { + quarantinedPoolPaths.delete(poolPath); + return parsed as LoopbackPool; } + problem = Array.isArray(parsed) ? 'empty array' : `JSON ${parsed === null ? 'null' : typeof parsed} instead of an array`; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return emptyPool(); + if (!(error instanceof SyntaxError)) throw error; + problem = error.message; + } + + let modifiedMs: number; + try { + modifiedMs = (await stat(poolPath)).mtimeMs; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return emptyPool(); throw error; } + const unmodifiedMs = Date.now() - modifiedMs; + if (unmodifiedMs < UNUSABLE_POOL_QUARANTINE_MS) { + if (!quarantinedPoolPaths.has(poolPath)) { + quarantinedPoolPaths.add(poolPath); + console.warn( + `[loopback-pool] ${poolPath} is unusable (${problem}); allocation waits until it has gone ${UNUSABLE_POOL_QUARANTINE_MS / 1000}s unmodified, then reinitializes it. Deleting it skips the wait, but only once every integration-test run on this machine has stopped.` + ); + } + return null; + } + quarantinedPoolPaths.delete(poolPath); + console.warn(`[loopback-pool] ${poolPath} has been unusable for ${Math.round(unmodifiedMs / 1000)}s (${problem}); reinitializing.`); + const pool = emptyPool(); + await writePoolFile(pool, poolPath); + return pool; } +let pendingWriteCounter = 0; + /** * Writes the loopback pool to the pool file as JSON. * + * Publishes by write-to-temp-then-rename, so readers only ever see the old complete file or + * the new complete file. The pending name must be unpredictable, not just unique: `wx` refuses + * a file already at that name, so a guessable one could be pre-planted to make every write fail. + * * @param pool The loopback pool array to persist + * @param poolPath The pool file path (overridable for tests; defaults to the shared pool file) */ -async function writePoolFile(pool: LoopbackPool): Promise { - await writeFile(HARPER_LOOPBACK_POOL_PATH, JSON.stringify(pool)); +export async function writePoolFile(pool: LoopbackPool, poolPath: string = HARPER_LOOPBACK_POOL_PATH): Promise { + const pendingPath = `${poolPath}.${process.pid}.${++pendingWriteCounter}.${randomBytes(8).toString('hex')}.pending`; + try { + const pendingFileHandle = await open(pendingPath, 'wx'); + try { + await pendingFileHandle.writeFile(JSON.stringify(pool)); + } finally { + await pendingFileHandle.close(); + } + await rename(pendingPath, poolPath); + } catch (error) { + await unlink(pendingPath).catch(() => {}); + throw error; + } } /** @@ -392,7 +454,8 @@ export async function validateLoopbackAddressPool(): Promise<{ * 6. Checks that no other process accepts connections on the address's canary ports * 7. Returns the loopback address (e.g., "127.0.0.2") * - * If no addresses are available, waits and retries until one becomes available. + * If no addresses are available, or the pool file is quarantined (see readPoolFile), waits and + * retries until one becomes available. * * **Pool file location:** `${tmpdir()}/harper-integration-test-loopback-pool.json` * **Lock file location:** `${tmpdir()}/harper-integration-test-loopback-pool.lock` @@ -430,6 +493,7 @@ export async function getNextAvailableLoopbackAddress(): Promise { const assignedIndex = await withLock(async () => { // Read the pool file const loopbackPool = await readPoolFile(); + if (!loopbackPool) return null; // Find the first available index we haven't already found in-use this attempt let index: number | null = null; @@ -548,6 +612,11 @@ function removeDeadProcessesFromPool(loopbackPool: LoopbackPool) { /** * Releases a loopback address back to the pool, making it available for other processes. * + * Only a slot this process holds is cleared: once the pool has been reinitialized, the same slot + * may already be another runner's reservation. Nothing is written while the pool file is + * quarantined (see readPoolFile): the reservation is already gone from it, and rewriting it would + * end the quarantine early. + * * @param address The loopback address to release (e.g., "127.0.0.2") * @throws InvalidLoopbackAddressError if the address format is invalid */ @@ -558,6 +627,7 @@ export async function releaseLoopbackAddress(address: string): Promise { await withLock(async () => { // Read the pool file const loopbackPool = await readPoolFile(); + if (!loopbackPool || loopbackPool[index] !== process.pid) return; // Release the address by setting it to null loopbackPool[index] = null; @@ -575,15 +645,18 @@ export async function releaseAllLoopbackAddressesForCurrentProcess(): Promise { // Read the pool file const loopbackPool = await readPoolFile(); + if (!loopbackPool) return; // Find and release all addresses assigned to this process + let released = false; for (let i = 0; i < loopbackPool.length; i++) { if (loopbackPool[i] === process.pid) { loopbackPool[i] = null; + released = true; } } // Write the updated pool back to the file - await writePoolFile(loopbackPool); + if (released) await writePoolFile(loopbackPool); }); } diff --git a/test/loopbackAddressPool.test.ts b/test/loopbackAddressPool.test.ts index 3ebd39d..0c3ada6 100644 --- a/test/loopbackAddressPool.test.ts +++ b/test/loopbackAddressPool.test.ts @@ -1,7 +1,9 @@ -import { test, after } from 'node:test'; +import { test, after, mock } from 'node:test'; import { deepStrictEqual, match, ok, rejects, strictEqual } from 'node:assert'; import { createServer, type AddressInfo, type Server } from 'node:net'; import { existsSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { mkdtemp, readdir, readFile, rm, utimes, writeFile } from 'node:fs/promises'; +import * as fsPromises from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { setTimeout as sleep } from 'node:timers/promises'; @@ -41,7 +43,7 @@ process.env.HARPER_INTEGRATION_TEST_LOOPBACK_POOL_COUNT = '1'; process.env.HARPER_INTEGRATION_TEST_CONFLICT_PROBE_PORT = String(operationsPort); process.env.HARPER_INTEGRATION_TEST_HTTP_CONFLICT_PROBE_PORT = String(httpPort); delete process.env[ALLOW_ENV]; -const { getNextAvailableLoopbackAddress, releaseLoopbackAddress } = await import('../src/loopbackAddressPool.ts'); +const { getNextAvailableLoopbackAddress, releaseLoopbackAddress, readPoolFile, writePoolFile } = await import('../src/loopbackAddressPool.ts'); const { startHarper, setupHarperWithFixture, publishHarperNode, createHarperContext } = await import( '../src/harperLifecycle.ts' ); @@ -80,6 +82,56 @@ async function refusal(): Promise { throw new Error(`expected getNextAvailableLoopbackAddress to reject, but it handed out ${address}`); } +let freshImportCounter = 0; +/** Monotonic cache-busting suffix — millisecond-resolution timestamps can collide. */ +function freshModuleUrl(): string { + return `../src/loopbackAddressPool.ts?fresh=${++freshImportCounter}`; +} + +async function withTempDir(body: (dir: string) => Promise): Promise { + const dir = await mkdtemp(join(tmpdir(), 'loopback-pool-test-')); + try { + await body(dir); + } finally { + await rm(dir, { recursive: true, force: true }); + } +} + +const POOL_FILE_NAME = 'harper-integration-test-loopback-pool.json'; + +async function ageBeyondQuarantine(path: string): Promise { + const anHourAgo = new Date(Date.now() - 3600000); + await utimes(path, anHourAgo, anHourAgo); +} + +/** + * A fresh copy of the pool module whose pool and lock live in `dir` and which hands out only + * 127.0.0.1. Both canary ports are chosen ephemerally, so a real Harper on this machine can't + * make the result depend on the host. + */ +async function importIsolatedPool(dir: string): Promise { + const [operationsPort, httpPort] = await twoFreePorts(); + const overrides: Record = { + TMPDIR: dir, + TMP: dir, + TEMP: dir, + HARPER_INTEGRATION_TEST_LOOPBACK_POOL_START: '1', + HARPER_INTEGRATION_TEST_LOOPBACK_POOL_COUNT: '1', + HARPER_INTEGRATION_TEST_CONFLICT_PROBE_PORT: String(operationsPort), + HARPER_INTEGRATION_TEST_HTTP_CONFLICT_PROBE_PORT: String(httpPort), + }; + const saved = Object.fromEntries(Object.keys(overrides).map((name) => [name, process.env[name]])); + Object.assign(process.env, overrides); + try { + return await import(freshModuleUrl()); + } finally { + for (const [name, value] of Object.entries(saved)) { + if (value === undefined) delete process.env[name]; + else process.env[name] = value; + } + } +} + test('hands out an address when no other process listens on its ports', async () => { const address = await getNextAvailableLoopbackAddress(); strictEqual(address, HOST); @@ -208,3 +260,147 @@ test('startHarper and setupHarperWithFixture remove the install directory they c await close(listener); } }); + +test('readPoolFile quarantines an unusable pool file, then reinitializes it once it has gone unmodified long enough', async (t) => { + t.mock.method(console, 'warn', () => {}); + await withTempDir(async (dir) => { + const poolPath = join(dir, 'pool.json'); + for (const content of ['[1,2,nul', '', 'null', '{}', '[]']) { + await writeFile(poolPath, content); + strictEqual(await readPoolFile(poolPath), null, `fresh ${JSON.stringify(content)} should be quarantined`); + + await ageBeyondQuarantine(poolPath); + const pool = await readPoolFile(poolPath); + ok(pool && pool.length > 0 && pool.every((slot) => slot === null), `aged ${JSON.stringify(content)} should reinitialize`); + deepStrictEqual(JSON.parse(await readFile(poolPath, 'utf-8')), pool, 'the reinitialized pool should be published'); + } + }); +}); + +test('readPoolFile reads a pool file deleted between its read and its stat as missing', async () => { + const m = mock.module('node:fs/promises', { + namedExports: { + ...fsPromises, + readFile: async () => '[1,2,nul', + stat: async () => { + const error = new Error('simulated ENOENT') as NodeJS.ErrnoException; + error.code = 'ENOENT'; + throw error; + }, + }, + }); + try { + const { readPoolFile: freshReadPoolFile } = await import(freshModuleUrl()); + const pool = await freshReadPoolFile('unused'); + ok(pool.length > 0 && pool.every((slot: number | null) => slot === null)); + } finally { + m.restore(); + } +}); + +test('a reservation lost to an unusable pool file is not reissued until the quarantine ends', async (t) => { + const warn = t.mock.method(console, 'warn', () => {}); + await withTempDir(async (dir) => { + const pool = await importIsolatedPool(dir); + const poolPath = join(dir, POOL_FILE_NAME); + strictEqual(await pool.getNextAvailableLoopbackAddress(), '127.0.0.1'); + + // A writer that doesn't publish by rename dies mid-write, losing that reservation. + await writeFile(poolPath, '['); + const reissued = pool.getNextAvailableLoopbackAddress(); + try { + strictEqual(await Promise.race([reissued.then(() => 'reissued'), sleep(2500).then(() => 'waiting')]), 'waiting'); + + await pool.releaseLoopbackAddress('127.0.0.1'); + await pool.releaseAllLoopbackAddressesForCurrentProcess(); + strictEqual(await readFile(poolPath, 'utf-8'), '[', 'releasing must not end the quarantine'); + + await ageBeyondQuarantine(poolPath); + strictEqual(await Promise.race([reissued, sleep(5000).then(() => 'still waiting')]), '127.0.0.1'); + } finally { + // Unblocks the allocation if an assertion above failed, so the test can't hang. + await rm(poolPath, { force: true }); + await reissued; + } + strictEqual(warn.mock.calls.filter((call) => String(call.arguments[0]).includes('is unusable')).length, 1); + }); +}); + +test('releaseLoopbackAddress clears only a slot this process holds', async () => { + await withTempDir(async (dir) => { + const pool = await importIsolatedPool(dir); + const poolPath = join(dir, POOL_FILE_NAME); + + strictEqual(await pool.getNextAvailableLoopbackAddress(), '127.0.0.1'); + await pool.releaseLoopbackAddress('127.0.0.1'); + deepStrictEqual(JSON.parse(await readFile(poolPath, 'utf-8')), [null]); + + const otherRunnerPid = process.pid + 1; + await writeFile(poolPath, JSON.stringify([otherRunnerPid])); + await pool.releaseLoopbackAddress('127.0.0.1'); + deepStrictEqual(JSON.parse(await readFile(poolPath, 'utf-8')), [otherRunnerPid]); + }); +}); + +test('readPoolFile still rethrows errors unrelated to a missing/corrupt file', async () => { + await withTempDir(async (dir) => { + const poolPath = join(dir, 'pool.json'); + const m = mock.module('node:fs/promises', { + // @types/node 22.0.0 only knows `namedExports`; `exports` isn't in its types yet. + namedExports: { + ...fsPromises, + readFile: async () => { + const error = new Error('simulated EACCES') as NodeJS.ErrnoException; + error.code = 'EACCES'; + throw error; + }, + }, + }); + try { + const { readPoolFile: freshReadPoolFile } = await import(freshModuleUrl()); + await rejects(() => freshReadPoolFile(poolPath), /simulated EACCES/); + } finally { + m.restore(); + } + }); +}); + +test('writePoolFile round-trips through readPoolFile and leaves no leftover pending file', async () => { + await withTempDir(async (dir) => { + const poolPath = join(dir, 'pool.json'); + const pool = [null, 123, null, 456]; + + await writePoolFile(pool, poolPath); + + deepStrictEqual(await readPoolFile(poolPath), pool); + deepStrictEqual(await readdir(dir), ['pool.json']); + }); +}); + +test('writePoolFile never leaves a partially-written file visible at the pool path', async () => { + await withTempDir(async (dir) => { + const poolPath = join(dir, 'pool.json'); + const original = [null, null, null]; + await writeFile(poolPath, JSON.stringify(original)); + + // A hard kill in this same window (vs. this mocked rejection) bypasses catch/finally and + // orphans the pending file — the pool has no reaper for that, unlike the instance registry. + const m = mock.module('node:fs/promises', { + namedExports: { + ...fsPromises, + rename: async () => { + throw new Error('simulated rename failure'); + }, + }, + }); + try { + const { writePoolFile: freshWritePoolFile } = await import(freshModuleUrl()); + await rejects(() => freshWritePoolFile([1, 2, 3], poolPath), /simulated rename failure/); + } finally { + m.restore(); + } + + deepStrictEqual(JSON.parse(await readFile(poolPath, 'utf-8')), original); + deepStrictEqual(await readdir(dir), ['pool.json']); + }); +});