From 0df3070d6df8ecce5f98e85354a2fec92a7fd1d2 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Tue, 11 Aug 2026 20:50:25 -0400 Subject: [PATCH] test(client): accelerate service lifecycle tests (#41879) --- packages/client/src/effect/service.ts | 55 ++++++++------ packages/client/src/promise/service.ts | 52 +++++++------- packages/client/src/service-timing.ts | 33 +++++++++ .../client/test/fixture/service-timing.ts | 26 +++++++ packages/client/test/fixture/service.ts | 5 +- packages/client/test/promise-service.test.ts | 30 +++----- packages/client/test/service.test.ts | 72 +++++++++++-------- 7 files changed, 176 insertions(+), 97 deletions(-) create mode 100644 packages/client/src/service-timing.ts create mode 100644 packages/client/test/fixture/service-timing.ts diff --git a/packages/client/src/effect/service.ts b/packages/client/src/effect/service.ts index 89bcce7418a..90f09f29534 100644 --- a/packages/client/src/effect/service.ts +++ b/packages/client/src/effect/service.ts @@ -4,6 +4,7 @@ import { spawn, type ChildProcess } from "node:child_process" import { homedir } from "node:os" import { join } from "node:path" import type { DiscoverOptions, Endpoint, EnsureOptions, StopOptions } from "../service.js" +import { defaultEnsureTiming, ensureTiming, type EnsureTiming } from "../service-timing.js" export * from "../service.js" /** Contents of the local service registration file. */ @@ -52,11 +53,12 @@ const discoverLocal = Effect.fnUntraced(function* (options: DiscoverOptions) { // becomes discoverable. A contender is never killed merely for slow startup. /** Ensure a healthy, compatible local service is running. */ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOptions = {}) { + const timing = ensureTiming(options) const contenders = new Set() let timeouts: { readonly info: Info; readonly count: number } | undefined let announced = false let lastSpawn = 0 - let spawnDelay = 5_000 + let spawnDelay = timing.spawnDelay const announce = (reason: "missing" | "version-mismatch", previousVersion?: string) => Effect.sync(() => { if (announced) return @@ -80,7 +82,7 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti }) }) const found = yield* Effect.gen(function* () { - const registration = yield* registered(options.file, true) + const registration = yield* registered(options.file, true, timing.requestTimeout) const info = registration.info const service = registration.service if (registration.timedOut && info !== undefined) { @@ -90,20 +92,20 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti } if (timeouts.count >= 3) { yield* announce("missing") - yield* evict(info, options) + yield* evict(info, options, timing) timeouts = undefined lastSpawn = Date.now() - spawnDelay } } else timeouts = undefined if (service !== undefined) { - spawnDelay = 5_000 + spawnDelay = timing.spawnDelay const compatible = !service.legacy && (options.version === undefined || service.version === options.version) if (compatible && service.state === "ready") return Option.some(service) if (compatible && service.state === "failed") return yield* Effect.fail(new Error("Background service failed to start")) if (compatible) return Option.none() yield* announce("version-mismatch", service.version) - yield* kill(service, options).pipe(Effect.ignore) + yield* kill(service, options, timing).pipe(Effect.ignore) lastSpawn = 0 return Option.none() } else if (lastSpawn === 0 && info !== undefined) lastSpawn = Date.now() @@ -111,7 +113,7 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti const finished = [...contenders].filter(contenderFinished) const failure = finished.map(contenderFailure).find((error): error is Error => error !== undefined) if (finished.some((item) => item.child.exitCode === 0)) { - spawnDelay = Math.min(spawnDelay * 2, 30_000) + spawnDelay = Math.min(spawnDelay * 2, timing.maxSpawnDelay) } finished.forEach((item) => contenders.delete(item)) if (failure !== undefined && contenders.size === 0) return yield* Effect.fail(failure) @@ -125,7 +127,7 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti }).pipe( Effect.repeat({ until: Option.isSome, - schedule: Schedule.max([Schedule.spaced("1 second"), Schedule.recurs(120)]), + schedule: Schedule.max([Schedule.spaced(timing.pollInterval), Schedule.recurs(timing.attempts)]), }), ) if (Option.isNone(found)) @@ -150,7 +152,7 @@ function contenderFinished(contender: Contender) { /** Stop the registered local service. */ export const stop = Effect.fn("service.stop")(function* (options: StopOptions = {}) { const existing = yield* find(options) - if (existing !== undefined) yield* kill(existing, options) + if (existing !== undefined) yield* kill(existing, options, defaultEnsureTiming) }) function fallback() { @@ -198,7 +200,11 @@ const probe = Effect.fnUntraced(function* (info: Info, allowLegacy = false) { return (yield* probeResult(info, allowLegacy)).service }) -const probeResult = Effect.fnUntraced(function* (info: Info, allowLegacy = false) { +const probeResult = Effect.fnUntraced(function* ( + info: Info, + allowLegacy = false, + timeout = defaultEnsureTiming.requestTimeout, +) { const endpoint = { url: info.url, auth: @@ -206,7 +212,7 @@ const probeResult = Effect.fnUntraced(function* (info: Info, allowLegacy = false ? undefined : { type: "basic" as const, username: "opencode", password: info.password }, } satisfies Endpoint - const signal = AbortSignal.timeout(2_000) + const signal = AbortSignal.timeout(timeout) const result = yield* Effect.promise(() => fetch(new URL("/api/health", info.url), { headers: headers(endpoint), @@ -249,10 +255,10 @@ const probeResult = Effect.fnUntraced(function* (info: Info, allowLegacy = false } }) -const registered = Effect.fnUntraced(function* (file?: string, allowLegacy = false) { +const registered = Effect.fnUntraced(function* (file?: string, allowLegacy = false, timeout?: number) { const info = yield* read(file) if (info === undefined) return { info: undefined, service: undefined, timedOut: false } - return { info, ...(yield* probeResult(info, allowLegacy)) } + return { info, ...(yield* probeResult(info, allowLegacy, timeout)) } }) // Health-checked lookup without the version gate: lifecycle operations must be @@ -263,7 +269,8 @@ const find = Effect.fnUntraced(function* (options: { readonly file?: string }) { // 50ms cadence bounded at ~5s, shared by stop escalation and each ensure // discovery window. -const poll = Schedule.max([Schedule.spaced("50 millis"), Schedule.recurs(100)]) +const poll = (timing: EnsureTiming) => + Schedule.max([Schedule.spaced(timing.stopPollInterval), Schedule.recurs(timing.stopPollAttempts)]) const signal = (pid: number, name: NodeJS.Signals) => Effect.try({ try: () => process.kill(pid, name), catch: (cause) => cause }).pipe(Effect.ignore) @@ -280,21 +287,25 @@ function same(left: Info, right: Info) { return left.id === right.id && left.version === right.version && left.url === right.url && left.pid === right.pid } -const evict = Effect.fnUntraced(function* (info: Info, options: { readonly file?: string }) { +const evict = Effect.fnUntraced(function* (info: Info, options: { readonly file?: string }, timing: EnsureTiming) { const current = yield* read(options.file) if (current === undefined || !same(current, info)) return yield* signal(info.pid, "SIGTERM") - const done = yield* stopped(info.pid).pipe(Effect.retry(poll), Effect.option) + const done = yield* stopped(info.pid).pipe(Effect.retry(poll(timing)), Effect.option) if (Option.isSome(done)) return const latest = yield* read(options.file) if (latest === undefined || !same(latest, info)) return yield* signal(info.pid, "SIGKILL") - yield* stopped(info.pid).pipe(Effect.retry(poll)) + yield* stopped(info.pid).pipe(Effect.retry(poll(timing))) }) -const kill = Effect.fnUntraced(function* (service: LocalService, options: { readonly file?: string }) { - const requested = yield* requestStop(service) +const kill = Effect.fnUntraced(function* ( + service: LocalService, + options: { readonly file?: string }, + timing: EnsureTiming, +) { + const requested = yield* requestStop(service, timing.requestTimeout) if (requested === "rejected") return if (requested === "unsupported") { // A stale registration may point at a reused PID. Authenticate again @@ -303,25 +314,25 @@ const kill = Effect.fnUntraced(function* (service: LocalService, options: { read if (current === undefined || !same(current.info, service.info)) return yield* signal(service.info.pid, "SIGTERM") } - const done = yield* stopped(service.info.pid).pipe(Effect.retry(poll), Effect.option) + const done = yield* stopped(service.info.pid).pipe(Effect.retry(poll(timing)), Effect.option) if (Option.isSome(done)) return const latest = yield* find(options) if (latest === undefined || !same(latest.info, service.info)) return yield* signal(service.info.pid, "SIGKILL") - yield* stopped(service.info.pid).pipe(Effect.retry(poll)) + yield* stopped(service.info.pid).pipe(Effect.retry(poll(timing))) }) const decodeStopResponse = Schema.decodeUnknownOption(ServiceStatus.StopResponse) -const requestStop = Effect.fnUntraced(function* (service: LocalService) { +const requestStop = Effect.fnUntraced(function* (service: LocalService, timeout = defaultEnsureTiming.requestTimeout) { if (service.info.id === undefined || service.legacy) return "unsupported" as const const response = yield* Effect.tryPromise(() => fetch(new URL("/api/service/stop", service.info.url), { method: "POST", headers: { ...headers(service.endpoint), "content-type": "application/json" }, body: JSON.stringify({ instanceID: service.info.id }), - signal: AbortSignal.timeout(2_000), + signal: AbortSignal.timeout(timeout), }), ).pipe(Effect.option, Effect.map(Option.getOrUndefined)) if (response === undefined || response.status === 404 || response.status === 405) return "unsupported" as const diff --git a/packages/client/src/promise/service.ts b/packages/client/src/promise/service.ts index af4372b8e2b..af81334227d 100644 --- a/packages/client/src/promise/service.ts +++ b/packages/client/src/promise/service.ts @@ -3,6 +3,7 @@ import { spawn, type ChildProcess } from "node:child_process" import { homedir } from "node:os" import { join } from "node:path" import type { DiscoverOptions, Endpoint, Info, EnsureOptions, StopOptions } from "../service.js" +import { defaultEnsureTiming, ensureTiming, type EnsureTiming } from "../service-timing.js" import type { ServiceHealth, ServiceStopResponse } from "./generated/types.js" export * from "../service.js" @@ -32,12 +33,13 @@ async function discoverLocal(options: DiscoverOptions) { /** Ensure a healthy, compatible local service is running. */ export async function ensure(options: EnsureOptions = {}): Promise { - const deadline = Date.now() + 120_000 + const timing = ensureTiming(options) + const deadline = Date.now() + timing.promiseTimeout const contenders = new Set() let timeouts: { readonly info: Info; readonly count: number } | undefined let announced = false let lastSpawn = 0 - let spawnDelay = 5_000 + let spawnDelay = timing.spawnDelay const announce = (reason: "missing" | "version-mismatch", previousVersion?: string) => { if (announced) return @@ -62,7 +64,7 @@ export async function ensure(options: EnsureOptions = {}): Promise { while (true) { if (Date.now() >= deadline) throw new Error("Timed out waiting for the background service to start") - const registration = await registered(options.file, true) + const registration = await registered(options.file, true, timing.requestTimeout) if (registration.timedOut && registration.info !== undefined) { timeouts = { info: registration.info, @@ -70,21 +72,21 @@ export async function ensure(options: EnsureOptions = {}): Promise { } if (timeouts.count >= 3) { announce("missing") - await evict(registration.info, options) + await evict(registration.info, options, timing) timeouts = undefined lastSpawn = Date.now() - spawnDelay } } else timeouts = undefined if (registration.service !== undefined) { - spawnDelay = 5_000 + spawnDelay = timing.spawnDelay const service = registration.service const compatible = !service.legacy && (options.version === undefined || service.version === options.version) if (compatible && service.state === "ready") return service.endpoint if (compatible && service.state === "failed") throw new Error("Background service failed to start") if (!compatible) { announce("version-mismatch", service.version) - await kill(service, options).catch(() => undefined) + await kill(service, options, timing).catch(() => undefined) lastSpawn = 0 } } else { @@ -92,7 +94,7 @@ export async function ensure(options: EnsureOptions = {}): Promise { const finished = [...contenders].filter(contenderFinished) const failure = finished.map(contenderFailure).find((error) => error !== undefined) if (finished.some((item) => item.child.exitCode === 0)) { - spawnDelay = Math.min(spawnDelay * 2, 30_000) + spawnDelay = Math.min(spawnDelay * 2, timing.maxSpawnDelay) } finished.forEach((item) => contenders.delete(item)) if (failure !== undefined && contenders.size === 0) throw failure @@ -103,7 +105,7 @@ export async function ensure(options: EnsureOptions = {}): Promise { lastSpawn = Date.now() } } - await delay(1_000) + await delay(timing.pollInterval) } } @@ -124,7 +126,7 @@ function contenderFinished(contender: Contender) { /** Stop the registered local service. */ export async function stop(options: StopOptions = {}) { const existing = await find(options) - if (existing !== undefined) await kill(existing, options) + if (existing !== undefined) await kill(existing, options, defaultEnsureTiming) } function fallback() { @@ -161,7 +163,7 @@ async function probe(info: Info, allowLegacy = false): Promise undefined) if (response === undefined || response.status === 404 || response.status === 405) return "unsupported" as const const body = (await response.json().catch(() => undefined)) as ServiceStopResponse | undefined diff --git a/packages/client/src/service-timing.ts b/packages/client/src/service-timing.ts new file mode 100644 index 00000000000..8b8a877362b --- /dev/null +++ b/packages/client/src/service-timing.ts @@ -0,0 +1,33 @@ +export type EnsureTiming = { + readonly pollInterval: number + readonly attempts: number + readonly requestTimeout: number + readonly spawnDelay: number + readonly maxSpawnDelay: number + readonly promiseTimeout: number + readonly stopPollInterval: number + readonly stopPollAttempts: number +} + +const timings = new WeakMap() + +export const defaultEnsureTiming: EnsureTiming = { + pollInterval: 1_000, + attempts: 120, + requestTimeout: 2_000, + spawnDelay: 5_000, + maxSpawnDelay: 30_000, + promiseTimeout: 120_000, + stopPollInterval: 50, + stopPollAttempts: 100, +} + +export function ensureTiming(options: object) { + return timings.get(options) ?? defaultEnsureTiming +} + +// Keep test timing out of the public lifecycle option types. +export function withEnsureTiming(options: A, overrides: Partial): A { + timings.set(options, { ...defaultEnsureTiming, ...overrides }) + return options +} diff --git a/packages/client/test/fixture/service-timing.ts b/packages/client/test/fixture/service-timing.ts new file mode 100644 index 00000000000..7a242c14080 --- /dev/null +++ b/packages/client/test/fixture/service-timing.ts @@ -0,0 +1,26 @@ +import { withEnsureTiming } from "../../src/service-timing" + +const timing = { + pollInterval: 20, + requestTimeout: 100, + spawnDelay: 200, + maxSpawnDelay: 1_200, + promiseTimeout: 3_000, + stopPollInterval: 5, +} + +export function accelerate(ensure: (options: A) => B) { + return (options: A) => ensure(withEnsureTiming(options, timing)) +} + +export async function waitForExit(pid: number) { + for (let attempt = 0; attempt < 600; attempt++) { + try { + process.kill(pid, 0) + } catch { + return + } + await Bun.sleep(5) + } + throw new Error(`Timed out waiting for process ${pid}`) +} diff --git a/packages/client/test/fixture/service.ts b/packages/client/test/fixture/service.ts index 508175174ed..d348f2239a8 100644 --- a/packages/client/test/fixture/service.ts +++ b/packages/client/test/fixture/service.ts @@ -17,7 +17,7 @@ if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated" || if (!owner) process.exit(mode === "coordinated-failed-loser" ? 1 : 0) if (mode === "coordinated" || mode === "coordinated-failed-loser") { while ((await Bun.file(registration + ".starts").text()).trim().split("\n").length < 2) await Bun.sleep(10) - if (mode === "coordinated-failed-loser") await Bun.sleep(1_500) + if (mode === "coordinated-failed-loser") await Bun.sleep(Number(delay ?? 1_500)) } else await Bun.sleep(Number(delay)) if (mode === "delayed-failed") process.exit(1) } @@ -30,7 +30,7 @@ const server = Bun.serve({ async fetch(request) { const pathname = new URL(request.url).pathname if (pathname === "/api/service/stop" && mode === "reject-stop") { - await writeFile(registration + ".stop-attempt", "") + await appendFile(registration + ".stop-attempts", process.pid + "\n") return Response.json({ accepted: false }) } if (pathname === "/api/service/stop" && mode === "graceful") { @@ -42,6 +42,7 @@ const server = Bun.serve({ } if (pathname !== "/api/health") return new Response(null, { status: 404 }) requests += 1 + if (mode === "starting") await writeFile(registration + ".health-request", "") if (mode === "hanging") { await appendFile(registration + ".requests", process.pid + "\n") return new Promise(() => {}) diff --git a/packages/client/test/promise-service.test.ts b/packages/client/test/promise-service.test.ts index 7e448fbbfc0..db99adbf148 100644 --- a/packages/client/test/promise-service.test.ts +++ b/packages/client/test/promise-service.test.ts @@ -3,8 +3,10 @@ import { mkdtemp, rm } from "node:fs/promises" import { tmpdir } from "node:os" import { join } from "node:path" import { Service, type EnsureReason } from "../src/promise/service" +import { accelerate, waitForExit } from "./fixture/service-timing" const fixture = join(import.meta.dir, "fixture/service.ts") +const ensure = accelerate(Service.ensure) const processes: Bun.Subprocess[] = [] const directories: string[] = [] @@ -28,7 +30,7 @@ test("ensures a missing service with native promises", async () => { const registration = join(directory, "service.json") const starts: EnsureReason[] = [] - const endpoint = await Service.ensure({ + const endpoint = await ensure({ file: registration, version: "test", command: [process.execPath, fixture, registration, "coordinated"], @@ -42,16 +44,16 @@ test("ensures a missing service with native promises", async () => { process.kill(info.pid, "SIGTERM") await waitForExit(info.pid) } -}, 15_000) +}) test("waits for a live contender when another native contender fails", async () => { const directory = await temp() const registration = join(directory, "service.json") - const endpoint = await Service.ensure({ + const endpoint = await ensure({ file: registration, version: "test", - command: [process.execPath, fixture, registration, "coordinated-failed-loser"], + command: [process.execPath, fixture, registration, "coordinated-failed-loser", "300"], }) const info = await Bun.file(registration).json() try { @@ -60,12 +62,12 @@ test("waits for a live contender when another native contender fails", async () process.kill(info.pid, "SIGTERM") await waitForExit(info.pid) } -}, 15_000) +}) test("reports a failed registered service", async () => { const registration = await setup("failed-owner") - await expect(Service.ensure({ file: registration, version: "test", command: [] })).rejects.toThrow( + await expect(ensure({ file: registration, version: "test", command: [] })).rejects.toThrow( "Background service failed to start", ) }) @@ -81,7 +83,7 @@ test("evicts an unresponsive registered service before starting its replacement" await waitForFile(registration) const original = await Bun.file(registration).json() - const endpoint = await Service.ensure({ + const endpoint = await ensure({ file: registration, version: "test", command: [process.execPath, fixture, registration, "delayed", "10"], @@ -94,7 +96,7 @@ test("evicts an unresponsive registered service before starting its replacement" expect(endpoint.url).toBe(replacement.url) process.kill(replacement.pid, "SIGTERM") await waitForExit(replacement.pid) -}, 20_000) +}) test("requests graceful stop of the exact service instance", async () => { const registration = await setup("graceful") @@ -126,15 +128,3 @@ async function waitForFile(file: string) { } throw new Error(`Timed out waiting for ${file}`) } - -async function waitForExit(pid: number) { - for (let attempt = 0; attempt < 600; attempt++) { - try { - process.kill(pid, 0) - } catch { - return - } - await Bun.sleep(5) - } - throw new Error(`Timed out waiting for process ${pid}`) -} diff --git a/packages/client/test/service.test.ts b/packages/client/test/service.test.ts index 0078e52cc19..6c0dfdecc58 100644 --- a/packages/client/test/service.test.ts +++ b/packages/client/test/service.test.ts @@ -5,8 +5,10 @@ import { mkdtemp, rm, writeFile } from "node:fs/promises" import { tmpdir } from "node:os" import { join } from "node:path" import { Service, type EnsureReason } from "../src/effect/service" +import { accelerate, waitForExit } from "./fixture/service-timing" const fixture = join(import.meta.dir, "fixture/service.ts") +const ensure = accelerate(Service.ensure) const processes: Bun.Subprocess[] = [] const directories: string[] = [] @@ -25,7 +27,7 @@ test("a concurrent same-version start cannot invalidate a resolved endpoint", as const starts: EnsureReason[] = [] const first = run( - Service.ensure({ + ensure({ file: registration, version: "test", command: [], @@ -34,7 +36,7 @@ test("a concurrent same-version start cannot invalidate a resolved endpoint", as ) await waitForFile(registration + ".first-request") - const resolved = await run(Service.ensure({ file: registration, version: "test" })) + const resolved = await run(ensure({ file: registration, version: "test" })) expect(resolved.url).toBe(original.url) await writeFile(registration + ".release", "") @@ -50,9 +52,9 @@ test("waits for a registered service to finish starting", async () => { const registration = join(directory, "service.json") const process = spawn(registration, "starting") await waitForFile(registration) - const result = run(Service.ensure({ file: registration, version: "test", command: [] })) + const result = run(ensure({ file: registration, version: "test", command: [] })) - await Bun.sleep(500) + await waitForFile(registration + ".health-request") expect(process.exitCode).toBe(null) await writeFile(registration + ".release", "") expect((await result).url).toBe((await Bun.file(registration).json()).url) @@ -64,7 +66,7 @@ test("reports a failed registered service without spawning", async () => { const process = spawn(registration, "failed-owner") await waitForFile(registration) - await expect(run(Service.ensure({ file: registration, version: "test", command: [] }))).rejects.toThrow( + await expect(run(ensure({ file: registration, version: "test", command: [] }))).rejects.toThrow( "Background service failed to start", ) expect(process.exitCode).toBe(null) @@ -78,7 +80,7 @@ test("evicts an unresponsive registered service before starting its replacement" const original = await Bun.file(registration).json() const endpoint = await run( - Service.ensure({ + ensure({ file: registration, version: "test", command: [process.execPath, fixture, registration, "delayed", "10"], @@ -92,7 +94,8 @@ test("evicts an unresponsive registered service before starting its replacement" expect(endpoint.url).toBe(replacement.url) expect(await health(endpoint.url)).toEqual({ healthy: true, version: "test", pid: replacement.pid }) process.kill(replacement.pid, "SIGTERM") -}, 20_000) + await waitForExit(replacement.pid) +}) test("requests graceful stop of the exact service instance", async () => { const directory = await temp() @@ -114,7 +117,7 @@ test("does not spawn contenders while an incompatible service rejects replacemen await waitForFile(registration) const controller = new AbortController() const starting = Effect.runPromise( - Service.ensure({ + ensure({ file: registration, version: "test", command: [process.execPath, fixture, contender, "record-start"], @@ -122,8 +125,7 @@ test("does not spawn contenders while an incompatible service rejects replacemen { signal: controller.signal }, ) - await waitForFile(registration + ".stop-attempt") - await Bun.sleep(500) + await waitForLines(registration + ".stop-attempts", 2) controller.abort() await starting.catch(() => undefined) @@ -138,18 +140,18 @@ test("a legacy health response is still replaced", async () => { await waitForFile(registration) const starts: EnsureReason[] = [] - const result = run(Service.ensure({ file: registration, command: [], onStart: (reason) => starts.push(reason) })) + const result = run(ensure({ file: registration, command: [], onStart: (reason) => starts.push(reason) })) await expect(result).rejects.toThrow("Missing service command") expect(starts).toEqual(["version-mismatch"]) await existing.exited -}, 10_000) +}) test("waits for a slow winner while bounding lock probes", async () => { const directory = await temp() const registration = join(directory, "service.json") const endpoint = await run( - Service.ensure({ + ensure({ file: registration, version: "test", command: [process.execPath, fixture, registration, "coordinated"], @@ -162,17 +164,18 @@ test("waits for a slow winner while bounding lock probes", async () => { expect((await Bun.file(registration + ".starts").text()).trim().split("\n")).toHaveLength(2) } finally { process.kill(info.pid, "SIGTERM") + await waitForExit(info.pid) } -}, 15_000) +}) test("waits for a live contender when another contender fails", async () => { const directory = await temp() const registration = join(directory, "service.json") const endpoint = await run( - Service.ensure({ + ensure({ file: registration, version: "test", - command: [process.execPath, fixture, registration, "coordinated-failed-loser"], + command: [process.execPath, fixture, registration, "coordinated-failed-loser", "300"], }), ) const info = await Bun.file(registration).json() @@ -180,62 +183,63 @@ test("waits for a live contender when another contender fails", async () => { expect(endpoint.url).toBe(info.url) } finally { process.kill(info.pid, "SIGTERM") + await waitForExit(info.pid) } -}, 15_000) +}) test("reports a contender that fails to start", async () => { const directory = await temp() const registration = join(directory, "service.json") await expect( run( - Service.ensure({ + ensure({ file: registration, version: "test", command: [process.execPath, fixture, registration, "failed"], }), ), ).rejects.toThrow("Server process exited with code 1") -}, 10_000) +}) test("reports a contender terminated by a signal", async () => { const directory = await temp() const registration = join(directory, "service.json") await expect( run( - Service.ensure({ + ensure({ file: registration, version: "test", command: [process.execPath, fixture, registration, "signal"], }), ), ).rejects.toThrow(/Server process (terminated by|exited with code)/) -}, 10_000) +}) test("reports a slow contender that eventually fails", async () => { const directory = await temp() const registration = join(directory, "service.json") await expect( run( - Service.ensure({ + ensure({ file: registration, version: "test", - command: [process.execPath, fixture, registration, "delayed-failed", "8000"], + command: [process.execPath, fixture, registration, "delayed-failed", "500"], }), ), ).rejects.toThrow("Server process exited with code 1") -}, 15_000) +}) test("replaces an incompatible owner that appears during startup", async () => { const directory = await temp() const registration = join(directory, "service.json") const starting = run( - Service.ensure({ + ensure({ file: registration, version: "test", - command: [process.execPath, fixture, registration, "delayed", "8000"], + command: [process.execPath, fixture, registration, "delayed", "500"], }), ) - await Bun.sleep(1_000) + await waitForFile(registration + ".starts") const old = spawn(registration, "old") await waitForFile(registration) const endpoint = await starting @@ -246,8 +250,9 @@ test("replaces an incompatible owner that appears during startup", async () => { await old.exited } finally { process.kill(info.pid, "SIGTERM") + await waitForExit(info.pid) } -}, 20_000) +}) function run(effect: Effect.Effect) { return Effect.runPromise(effect.pipe(Effect.provide(NodeFileSystem.layer))) @@ -276,6 +281,17 @@ async function waitForFile(file: string) { throw new Error(`Timed out waiting for ${file}`) } +async function waitForLines(file: string, count: number) { + for (let attempt = 0; attempt < 600; attempt++) { + const text = await Bun.file(file) + .text() + .catch(() => "") + if (text.trim().split("\n").length >= count) return + await Bun.sleep(5) + } + throw new Error(`Timed out waiting for ${count} lines in ${file}`) +} + async function health(url: string) { return fetch(new URL("/api/health", url), { signal: AbortSignal.timeout(1_000) }).then((response) => response.json()) }