diff --git a/packages/client/src/effect/service.ts b/packages/client/src/effect/service.ts index f27b2736209..89bcce7418a 100644 --- a/packages/client/src/effect/service.ts +++ b/packages/client/src/effect/service.ts @@ -53,6 +53,7 @@ const discoverLocal = Effect.fnUntraced(function* (options: DiscoverOptions) { /** Ensure a healthy, compatible local service is running. */ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOptions = {}) { const contenders = new Set() + let timeouts: { readonly info: Info; readonly count: number } | undefined let announced = false let lastSpawn = 0 let spawnDelay = 5_000 @@ -82,6 +83,18 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti const registration = yield* registered(options.file, true) const info = registration.info const service = registration.service + if (registration.timedOut && info !== undefined) { + timeouts = { + info, + count: timeouts !== undefined && same(timeouts.info, info) ? timeouts.count + 1 : 1, + } + if (timeouts.count >= 3) { + yield* announce("missing") + yield* evict(info, options) + timeouts = undefined + lastSpawn = Date.now() - spawnDelay + } + } else timeouts = undefined if (service !== undefined) { spawnDelay = 5_000 const compatible = !service.legacy && (options.version === undefined || service.version === options.version) @@ -182,6 +195,10 @@ type LocalService = { } 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 endpoint = { url: info.url, auth: @@ -189,39 +206,53 @@ const probe = Effect.fnUntraced(function* (info: Info, allowLegacy = false) { ? undefined : { type: "basic" as const, username: "opencode", password: info.password }, } satisfies Endpoint - const response = yield* Effect.tryPromise(() => + const signal = AbortSignal.timeout(2_000) + const result = yield* Effect.promise(() => fetch(new URL("/api/health", info.url), { headers: headers(endpoint), - signal: AbortSignal.timeout(2_000), - }), - ).pipe(Effect.option, Effect.map(Option.getOrUndefined)) - if (response === undefined) return undefined - const body = yield* Effect.tryPromise(() => response.json()).pipe(Effect.option, Effect.map(Option.getOrUndefined)) + signal, + }) + .then(async (response) => ({ response, body: (await response.json()) as unknown })) + .then( + (value) => ({ value }), + (cause: unknown) => ({ cause }), + ), + ) + if ("cause" in result) return { service: undefined, timedOut: signal.aborted } + const response = result.value.response + const body = result.value.body const health = decodeHealth(body) if (Option.isSome(health)) { - if (health.value.pid !== info.pid) return undefined - if (info.version !== undefined && health.value.version !== info.version) return undefined + if (health.value.pid !== info.pid) return { service: undefined, timedOut: false } + if (info.version !== undefined && health.value.version !== info.version) + return { service: undefined, timedOut: false } return { - info, - endpoint, - version: health.value.version, - state: response.ok ? "ready" : response.status === 500 ? "failed" : "waiting", - legacy: false, - } satisfies LocalService + service: { + info, + endpoint, + version: health.value.version, + state: response.ok ? "ready" : response.status === 500 ? "failed" : "waiting", + legacy: false, + } satisfies LocalService, + timedOut: false, + } } if ( !allowLegacy || Option.isNone(decodeLegacyHealth(body)) || (typeof body === "object" && body !== null && ("version" in body || "pid" in body)) ) - return undefined - return { info, endpoint, state: "ready", legacy: true } satisfies LocalService + return { service: undefined, timedOut: false } + return { + service: { info, endpoint, state: "ready", legacy: true } satisfies LocalService, + timedOut: false, + } }) const registered = Effect.fnUntraced(function* (file?: string, allowLegacy = false) { const info = yield* read(file) - if (info === undefined) return { info: undefined, service: undefined } - return { info, service: yield* probe(info, allowLegacy) } + if (info === undefined) return { info: undefined, service: undefined, timedOut: false } + return { info, ...(yield* probeResult(info, allowLegacy)) } }) // Health-checked lookup without the version gate: lifecycle operations must be @@ -249,6 +280,19 @@ 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 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) + 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)) +}) + const kill = Effect.fnUntraced(function* (service: LocalService, options: { readonly file?: string }) { const requested = yield* requestStop(service) if (requested === "rejected") return diff --git a/packages/client/src/promise/service.ts b/packages/client/src/promise/service.ts index c03f4fe5516..87e83b4c282 100644 --- a/packages/client/src/promise/service.ts +++ b/packages/client/src/promise/service.ts @@ -34,6 +34,7 @@ async function discoverLocal(options: DiscoverOptions) { export async function ensure(options: EnsureOptions = {}): Promise { const deadline = Date.now() + 120_000 const contenders = new Set() + let timeouts: { readonly info: Info; readonly count: number } | undefined let announced = false let lastSpawn = 0 let spawnDelay = 5_000 @@ -62,6 +63,19 @@ 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) + if (registration.timedOut && registration.info !== undefined) { + timeouts = { + info: registration.info, + count: + timeouts !== undefined && same(timeouts.info, registration.info) ? timeouts.count + 1 : 1, + } + if (timeouts.count >= 3) { + announce("missing") + await evict(registration.info, options) + timeouts = undefined + lastSpawn = Date.now() - spawnDelay + } + } else timeouts = undefined if (registration.service !== undefined) { spawnDelay = 5_000 @@ -145,6 +159,10 @@ type LocalService = { } async function probe(info: Info, allowLegacy = false): Promise { + return (await probeResult(info, allowLegacy)).service +} + +async function probeResult(info: Info, allowLegacy = false) { const endpoint = { url: info.url, auth: @@ -152,30 +170,48 @@ async function probe(info: Info, allowLegacy = false): Promise undefined) - const body = (await response?.json().catch(() => undefined)) as ServiceHealth | { readonly healthy: true } | undefined + signal, + }) + .then(async (response) => ({ + response, + body: (await response.json()) as ServiceHealth | { readonly healthy: true }, + })) + .then( + (value) => ({ value }), + (cause: unknown) => ({ cause }), + ) + if ("cause" in result) return { service: undefined, timedOut: signal.aborted } + const response = result.value.response + const body = result.value.body if (body !== undefined && "version" in body && "pid" in body) { - if (body.pid !== info.pid) return undefined - if (info.version !== undefined && body.version !== info.version) return undefined + if (body.pid !== info.pid) return { service: undefined, timedOut: false } + if (info.version !== undefined && body.version !== info.version) + return { service: undefined, timedOut: false } return { - info, - endpoint, - version: body.version, - state: response?.ok ? "ready" : response?.status === 500 ? "failed" : "waiting", - legacy: false, + service: { + info, + endpoint, + version: body.version, + state: response.ok ? "ready" : response.status === 500 ? "failed" : "waiting", + legacy: false, + } satisfies LocalService, + timedOut: false, } } - if (!allowLegacy || body?.healthy !== true) return undefined - return { info, endpoint, state: "ready", legacy: true } + if (!allowLegacy || body?.healthy !== true) return { service: undefined, timedOut: false } + return { + service: { info, endpoint, state: "ready", legacy: true } satisfies LocalService, + timedOut: false, + } } async function registered(file?: string, allowLegacy = false) { const info = await read(file) - if (info === undefined) return { info: undefined, service: undefined } - return { info, service: await probe(info, allowLegacy) } + if (info === undefined) return { info: undefined, service: undefined, timedOut: false } + return { info, ...(await probeResult(info, allowLegacy)) } } async function find(options: { readonly file?: string }) { @@ -209,6 +245,18 @@ function same(left: Info, right: Info) { return left.id === right.id && left.version === right.version && left.url === right.url && left.pid === right.pid } +async function evict(info: Info, options: { readonly file?: string }) { + const current = await read(options.file) + if (current === undefined || !same(current, info)) return + signal(info.pid, "SIGTERM") + if (await waitUntilStopped(info.pid)) return + + const latest = await read(options.file) + if (latest === undefined || !same(latest, info)) return + signal(info.pid, "SIGKILL") + if (!(await waitUntilStopped(info.pid))) throw new Error(`Server process ${info.pid} is still running`) +} + async function kill(service: LocalService, options: { readonly file?: string }) { const requested = await requestStop(service) if (requested === "rejected") return diff --git a/packages/client/test/fixture/service.ts b/packages/client/test/fixture/service.ts index 7cbbec20ee8..226f97782bf 100644 --- a/packages/client/test/fixture/service.ts +++ b/packages/client/test/fixture/service.ts @@ -47,6 +47,10 @@ const server = Bun.serve({ } if (pathname !== "/api/health") return new Response(null, { status: 404 }) requests += 1 + if (mode === "hanging") { + await appendFile(registration + ".requests", process.pid + "\n") + return new Promise(() => {}) + } if (mode === "modern" && requests === 1) { await writeFile(registration + ".first-request", "") while (!(await Bun.file(registration + ".release").exists())) await Bun.sleep(5) diff --git a/packages/client/test/promise-service.test.ts b/packages/client/test/promise-service.test.ts index 51744cf50e1..7e448fbbfc0 100644 --- a/packages/client/test/promise-service.test.ts +++ b/packages/client/test/promise-service.test.ts @@ -70,6 +70,32 @@ test("reports a failed registered service", async () => { ) }) +test("evicts an unresponsive registered service before starting its replacement", async () => { + const directory = await temp() + const registration = join(directory, "service.json") + const existing = Bun.spawn([process.execPath, fixture, registration, "hanging"], { + stdout: "ignore", + stderr: "inherit", + }) + processes.push(existing) + await waitForFile(registration) + const original = await Bun.file(registration).json() + + const endpoint = await Service.ensure({ + file: registration, + version: "test", + command: [process.execPath, fixture, registration, "delayed", "10"], + }) + const replacement = await Bun.file(registration).json() + + expect((await Bun.file(registration + ".requests").text()).trim().split("\n")).toHaveLength(3) + 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) + test("requests graceful stop of the exact service instance", async () => { const registration = await setup("graceful") const info = await Bun.file(registration).json() diff --git a/packages/client/test/service.test.ts b/packages/client/test/service.test.ts index 519c2a46df2..0078e52cc19 100644 --- a/packages/client/test/service.test.ts +++ b/packages/client/test/service.test.ts @@ -70,6 +70,30 @@ test("reports a failed registered service without spawning", async () => { expect(process.exitCode).toBe(null) }) +test("evicts an unresponsive registered service before starting its replacement", async () => { + const directory = await temp() + const registration = join(directory, "service.json") + const existing = spawn(registration, "hanging") + await waitForFile(registration) + const original = await Bun.file(registration).json() + + const endpoint = await run( + Service.ensure({ + file: registration, + version: "test", + command: [process.execPath, fixture, registration, "delayed", "10"], + }), + ) + const replacement = await Bun.file(registration).json() + + expect((await Bun.file(registration + ".requests").text()).trim().split("\n")).toHaveLength(3) + 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) + test("requests graceful stop of the exact service instance", async () => { const directory = await temp() const registration = join(directory, "service.json")