fix(client): wait for live service contenders (#40515)

packages/client/src/effect/service.ts

@@ -56,7 +56,6 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
   let announced = false
   let lastSpawn = 0
   let spawnDelay = 5_000
-  let ownerHeld = false
   const announce = (reason: "missing" | "version-mismatch", previousVersion?: string) =>
     Effect.sync(() => {
       if (announced) return
@@ -84,7 +83,6 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
     const info = registration.info
     const service = registration.service
     if (service !== undefined) {
-      ownerHeld = false
       spawnDelay = 5_000
       const compatible = !service.legacy && (options.version === undefined || service.version === options.version)
       if (compatible && service.state === "ready") return Option.some(service)
@@ -97,14 +95,13 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
       return Option.none<LocalService>()
     } else if (lastSpawn === 0 && info !== undefined) lastSpawn = Date.now()
 
-    const failure = [...contenders].map(contenderFailure).find((error): error is Error => error !== undefined)
-    if (failure !== undefined) return yield* Effect.fail(failure)
     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)) {
-      ownerHeld = true
       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) {
       yield* announce("missing")

packages/client/src/promise/service.ts

@@ -37,7 +37,6 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
   let announced = false
   let lastSpawn = 0
   let spawnDelay = 5_000
-  let ownerHeld = false
 
   const announce = (reason: "missing" | "version-mismatch", previousVersion?: string) => {
     if (announced) return
@@ -65,7 +64,6 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
     const registration = await registered(options.file, true)
 
     if (registration.service !== undefined) {
-      ownerHeld = false
       spawnDelay = 5_000
       const service = registration.service
       const compatible = !service.legacy && (options.version === undefined || service.version === options.version)
@@ -78,14 +76,13 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
       }
     } else {
       if (lastSpawn === 0 && registration.info !== undefined) lastSpawn = Date.now()
-      const failure = [...contenders].map(contenderFailure).find((error) => error !== undefined)
-      if (failure !== undefined) throw failure
       const finished = [...contenders].filter(contenderFinished)
+      const failure = finished.map(contenderFailure).find((error) => error !== undefined)
       if (finished.some((item) => item.child.exitCode === 0)) {
-        ownerHeld = true
         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) {
         announce("missing")

packages/client/test/fixture/service.ts

@@ -9,14 +9,20 @@ if (mode === "record-start") {
 }
 if (mode === "signal") process.kill(process.pid, process.platform === "win32" ? "SIGTERM" : "SIGKILL")
 
-if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated") {
+if (
+  mode === "delayed" ||
+  mode === "delayed-failed" ||
+  mode === "coordinated" ||
+  mode === "coordinated-failed-loser"
+) {
   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()
-  if (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)
   } else await Bun.sleep(Number(delay))
   if (mode === "delayed-failed") process.exit(1)
 }

packages/client/test/promise-service.test.ts

@@ -44,6 +44,24 @@ 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")
 

packages/client/test/service.test.ts

@@ -141,6 +141,24 @@ test("waits for a slow winner while bounding lock probes", async () => {
   }
 }, 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)
+  } finally {
+    process.kill(info.pid, "SIGTERM")
+  }
+}, 15_000)
+
 test("reports a contender that fails to start", async () => {
   const directory = await temp()
   const registration = join(directory, "service.json")