From 1cf74cfb52263d6181626298625b2de99f76d434 Mon Sep 17 00:00:00 2001 From: Dax Raad Date: Tue, 11 Aug 2026 17:12:27 -0400 Subject: [PATCH] fix(client): simplify service startup recovery --- packages/client/src/effect/service.ts | 23 ++++++++-------- packages/client/src/promise/service.ts | 19 +++++++------ packages/client/test/fixture/service.ts | 9 +++---- packages/client/test/promise-service.test.ts | 24 +++-------------- packages/client/test/service.test.ts | 28 ++++---------------- 5 files changed, 31 insertions(+), 72 deletions(-) diff --git a/packages/client/src/effect/service.ts b/packages/client/src/effect/service.ts index 89bcce7418..3b80bdc932 100644 --- a/packages/client/src/effect/service.ts +++ b/packages/client/src/effect/service.ts @@ -48,11 +48,11 @@ const discoverLocal = Effect.fnUntraced(function* (options: DiscoverOptions) { }) // Idempotent ensure-running: reuses a healthy compatible server, replaces a -// version-mismatched one, and otherwise spawns small contenders until a server -// becomes discoverable. A contender is never killed merely for slow startup. +// version-mismatched one, and otherwise spawns one contender until a server +// becomes discoverable. The 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 contenders = new Set() + let contender: Contender | undefined let timeouts: { readonly info: Info; readonly count: number } | undefined let announced = false let lastSpawn = 0 @@ -108,17 +108,16 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti return Option.none() } else if (lastSpawn === 0 && info !== undefined) lastSpawn = Date.now() - 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)) { + const finished = contender !== undefined && contenderFinished(contender) ? contender : undefined + const failure = finished === undefined ? undefined : contenderFailure(finished) + if (finished?.child.exitCode === 0) { spawnDelay = Math.min(spawnDelay * 2, 30_000) } - finished.forEach((item) => contenders.delete(item)) - if (failure !== undefined && contenders.size === 0) return yield* Effect.fail(failure) - // Keep one candidate plus one lock probe so a pre-lock stall cannot block recovery. - if (contenders.size < 2 && Date.now() - lastSpawn >= spawnDelay) { + if (finished !== undefined) contender = undefined + if (failure !== undefined) return yield* Effect.fail(failure) + if (contender === undefined && Date.now() - lastSpawn >= spawnDelay) { yield* announce("missing") - contenders.add(yield* spawnContender) + contender = yield* spawnContender lastSpawn = Date.now() } return Option.none() @@ -206,7 +205,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(10_000) const result = yield* Effect.promise(() => fetch(new URL("/api/health", info.url), { headers: headers(endpoint), diff --git a/packages/client/src/promise/service.ts b/packages/client/src/promise/service.ts index af4372b8e2..d72ba28c4d 100644 --- a/packages/client/src/promise/service.ts +++ b/packages/client/src/promise/service.ts @@ -33,7 +33,7 @@ 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 contenders = new Set() + let contender: Contender | undefined let timeouts: { readonly info: Info; readonly count: number } | undefined let announced = false let lastSpawn = 0 @@ -89,17 +89,16 @@ export async function ensure(options: EnsureOptions = {}): Promise { } } else { if (lastSpawn === 0 && registration.info !== undefined) lastSpawn = Date.now() - const finished = [...contenders].filter(contenderFinished) - const failure = finished.map(contenderFailure).find((error) => error !== undefined) - if (finished.some((item) => item.child.exitCode === 0)) { + const finished = contender !== undefined && contenderFinished(contender) ? contender : undefined + const failure = finished === undefined ? undefined : contenderFailure(finished) + if (finished?.child.exitCode === 0) { spawnDelay = Math.min(spawnDelay * 2, 30_000) } - finished.forEach((item) => contenders.delete(item)) - if (failure !== undefined && contenders.size === 0) throw failure - // Keep one candidate plus one lock probe so a pre-lock stall cannot block recovery. - if (contenders.size < 2 && Date.now() - lastSpawn >= spawnDelay) { + if (finished !== undefined) contender = undefined + if (failure !== undefined) throw failure + if (contender === undefined && Date.now() - lastSpawn >= spawnDelay) { announce("missing") - contenders.add(spawnContender()) + contender = spawnContender() lastSpawn = Date.now() } } @@ -169,7 +168,7 @@ async function probeResult(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(10_000) const result = await fetch(new URL("/api/health", info.url), { headers: headers(endpoint), signal, diff --git a/packages/client/test/fixture/service.ts b/packages/client/test/fixture/service.ts index 508175174e..f8edae6131 100644 --- a/packages/client/test/fixture/service.ts +++ b/packages/client/test/fixture/service.ts @@ -9,16 +9,13 @@ if (mode === "record-start") { } if (mode === "signal") process.kill(process.pid, process.platform === "win32" ? "SIGTERM" : "SIGKILL") -if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated" || mode === "coordinated-failed-loser") { +if (mode === "delayed" || mode === "delayed-failed") { await appendFile(registration + ".starts", process.pid + "\n") const owner = await writeFile(registration + ".owner", String(process.pid), { flag: "wx" }) .then(() => true) .catch(() => false) - 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) - } else await Bun.sleep(Number(delay)) + if (!owner) process.exit(0) + await Bun.sleep(Number(delay)) if (mode === "delayed-failed") process.exit(1) } diff --git a/packages/client/test/promise-service.test.ts b/packages/client/test/promise-service.test.ts index 7e448fbbfc..b5401d8fa7 100644 --- a/packages/client/test/promise-service.test.ts +++ b/packages/client/test/promise-service.test.ts @@ -31,7 +31,7 @@ test("ensures a missing service with native promises", async () => { const endpoint = await Service.ensure({ file: registration, version: "test", - command: [process.execPath, fixture, registration, "coordinated"], + command: [process.execPath, fixture, registration, "delayed", "100"], onStart: (reason) => starts.push(reason), }) const info = await Bun.file(registration).json() @@ -44,24 +44,6 @@ test("ensures a missing service with native promises", async () => { } }, 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({ - file: registration, - version: "test", - command: [process.execPath, fixture, registration, "coordinated-failed-loser"], - }) - const info = await Bun.file(registration).json() - try { - expect(endpoint.url).toBe(info.url) - } finally { - process.kill(info.pid, "SIGTERM") - await waitForExit(info.pid) - } -}, 15_000) - test("reports a failed registered service", async () => { const registration = await setup("failed-owner") @@ -88,13 +70,13 @@ test("evicts an unresponsive registered service before starting its replacement" }) const replacement = await Bun.file(registration).json() - expect((await Bun.file(registration + ".requests").text()).trim().split("\n")).toHaveLength(3) + expect((await Bun.file(registration + ".requests").text()).trim().split("\n").length).toBeGreaterThanOrEqual(2) expect(await existing.exited).toBe(0) expect(replacement.pid).not.toBe(original.pid) expect(endpoint.url).toBe(replacement.url) process.kill(replacement.pid, "SIGTERM") await waitForExit(replacement.pid) -}, 20_000) +}, 45_000) test("requests graceful stop of the exact service instance", async () => { const registration = await setup("graceful") diff --git a/packages/client/test/service.test.ts b/packages/client/test/service.test.ts index 0078e52cc1..e795854692 100644 --- a/packages/client/test/service.test.ts +++ b/packages/client/test/service.test.ts @@ -86,13 +86,13 @@ test("evicts an unresponsive registered service before starting its replacement" ) const replacement = await Bun.file(registration).json() - expect((await Bun.file(registration + ".requests").text()).trim().split("\n")).toHaveLength(3) + expect((await Bun.file(registration + ".requests").text()).trim().split("\n").length).toBeGreaterThanOrEqual(2) expect(await existing.exited).toBe(0) expect(replacement.pid).not.toBe(original.pid) 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) +}, 45_000) test("requests graceful stop of the exact service instance", async () => { const directory = await temp() @@ -145,39 +145,21 @@ test("a legacy health response is still replaced", async () => { await existing.exited }, 10_000) -test("waits for a slow winner while bounding lock probes", async () => { +test("does not spawn another contender while the first is starting", async () => { const directory = await temp() const registration = join(directory, "service.json") const endpoint = await run( Service.ensure({ file: registration, version: "test", - command: [process.execPath, fixture, registration, "coordinated"], + command: [process.execPath, fixture, registration, "delayed", "6000"], }), ) const info = await Bun.file(registration).json() try { expect(endpoint.url).toBe(info.url) expect(await health(endpoint.url)).toEqual({ healthy: true, version: "test", pid: info.pid }) - expect((await Bun.file(registration + ".starts").text()).trim().split("\n")).toHaveLength(2) - } finally { - process.kill(info.pid, "SIGTERM") - } -}, 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({ - file: registration, - version: "test", - command: [process.execPath, fixture, registration, "coordinated-failed-loser"], - }), - ) - const info = await Bun.file(registration).json() - try { - expect(endpoint.url).toBe(info.url) + expect((await Bun.file(registration + ".starts").text()).trim().split("\n")).toHaveLength(1) } finally { process.kill(info.pid, "SIGTERM") }