fix(client): bound unresponsive service recovery

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<Contender>()
+  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<LocalService>()
     } 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<LocalService>()
@@ -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(3_000)
   const result = yield* Effect.promise(() =>
     fetch(new URL("/api/health", info.url), {
       headers: headers(endpoint),

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<Endpoint> {
   const deadline = Date.now() + 120_000
-  const contenders = new Set<Contender>()
+  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<Endpoint> {
       }
     } 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(3_000)
   const result = await fetch(new URL("/api/health", info.url), {
     headers: headers(endpoint),
     signal,

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)
 }
 

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)
+}, 30_000)
 
 test("requests graceful stop of the exact service instance", async () => {
   const registration = await setup("graceful")

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)
+}, 30_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")
   }