test(client): cover failed service stop escalation

packages/client/src/effect/service.ts

@@ -291,8 +291,8 @@ const terminate = Effect.fnUntraced(function* (info: Info, options: { readonly f
   if (current === undefined || !same(current, info)) return
   yield* signal(info.pid, "SIGTERM")
   const done = yield* stopped(info.pid).pipe(Effect.retry(poll(timing)), Effect.option)
-  // A gracefully stopping server may remove its registration before it releases its port, so only
-  // the exact process we signalled can tell us when it is safe for a restart to continue.
+  // The registration can disappear or change hands before this process exits. Only the PID we
+  // signalled can tell us whether it has stopped, so escalate based on that process.
   if (Option.isNone(done)) {
     yield* signal(info.pid, "SIGKILL")
     yield* stopped(info.pid).pipe(Effect.retry(poll(timing)))

packages/client/src/promise/service.ts

@@ -250,8 +250,8 @@ async function terminate(info: Info, options: { readonly file?: string }, timing
   const current = await read(options.file)
   if (current === undefined || !same(current, info)) return
   signal(info.pid, "SIGTERM")
-  // A gracefully stopping server may remove its registration before it releases its port, so only
-  // the exact process we signalled can tell us when it is safe for a restart to continue.
+  // The registration can disappear or change hands before this process exits. Only the PID we
+  // signalled can tell us whether it has stopped, so escalate based on that process.
   if (!(await waitUntilStopped(info.pid, timing))) {
     signal(info.pid, "SIGKILL")
     if (!(await waitUntilStopped(info.pid, timing))) throw new Error(`Server process ${info.pid} is still running`)

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

@@ -25,6 +25,16 @@ export async function serviceFixture() {
       processes.push(subprocess)
       return subprocess
     },
+    // The service's parent execs `sleep`, which never reaps it, so SIGKILL leaves a zombie that
+    // still answers `kill(pid, 0)`. Terminating the returned parent lets init reap the service.
+    spawnUnreaped(mode: string, ...args: string[]) {
+      const subprocess = Bun.spawn(["sh", "-c", '"$@" & exec sleep 60', "sh", ...command(mode, ...args)], {
+        stdout: "ignore",
+        stderr: "inherit",
+      })
+      processes.push(subprocess)
+      return subprocess
+    },
     // Service.ensure detaches contenders; track the elected process before asserting.
     track(pid: number) {
       pids.add(pid)
@@ -54,3 +64,8 @@ export async function serviceFixture() {
     },
   }
 }
+
+export async function expectPortAvailable(url: string) {
+  const server = Bun.serve({ port: Number(new URL(url).port), fetch: () => new Response() })
+  await server.stop(true)
+}

packages/client/test/fixture/service.ts

@@ -98,7 +98,7 @@ await rename(registration + ".tmp", registration)
 
 async function shutdown(signal?: NodeJS.Signals) {
   if (signal !== undefined) await writeFile(registration + ".signal", signal)
-  // A lingering server removes its registration first and keeps its port while it winds down.
+  // A lingering server unregisters on SIGTERM but keeps running, and holds its port, until killed.
   if (mode === "lingering") {
     await rm(registration, { force: true })
     await writeFile(registration + ".unregistered", "")

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

@@ -1,6 +1,6 @@
 import { expect, test } from "bun:test"
 import { Service, type EnsureReason } from "../src/promise/service"
-import { serviceFixture } from "./fixture/service-fixture"
+import { expectPortAvailable, serviceFixture } from "./fixture/service-fixture"
 import { accelerate } from "./fixture/service-timing"
 
 const ensure = accelerate(Service.ensure)
@@ -160,7 +160,7 @@ test("signals the registered service process", async () => {
   expect(await Bun.file(registration).exists()).toBe(false)
 })
 
-test("stop outlives a server that unregisters before releasing its port", async () => {
+test("stop escalates when the registration disappears before the process exits", async () => {
   await using fixture = await serviceFixture()
   const registration = fixture.registration
   const existing = fixture.spawn("lingering", "5000")
@@ -171,18 +171,33 @@ test("stop outlives a server that unregisters before releasing its port", async
 
   expect(await Bun.file(registration + ".signal").text()).toBe("SIGTERM")
   expect(() => process.kill(original.pid, 0)).toThrow()
-  const listener = Bun.serve({ port: Number(new URL(original.url).port), fetch: () => new Response() })
-  await listener.stop(true)
+  await expectPortAvailable(original.url)
   expect(await Bun.file(registration).exists()).toBe(false)
   await existing.exited
 }, 15_000)
 
+test.skipIf(process.platform === "win32")(
+  "stop fails when the process survives SIGKILL",
+  async () => {
+    await using fixture = await serviceFixture()
+    const registration = fixture.registration
+    fixture.spawnUnreaped("lingering", "60000")
+    await fixture.waitForFile()
+    const original = await Bun.file(registration).json()
+
+    await expect(stop({ file: registration })).rejects.toThrow(`Server process ${original.pid} is still running`)
+    expect(await Bun.file(registration + ".signal").text()).toBe("SIGTERM")
+  },
+  15_000,
+)
+
 test("stop waits for the original process while preserving a newly registered successor", async () => {
   await using fixture = await serviceFixture()
   const registration = fixture.registration
   const existing = fixture.spawn("lingering", "15000")
   await fixture.waitForFile()
   const original = await Bun.file(registration).json()
+  // Default timing keeps the grace period open long enough for the successor to register first.
   const stopping = Service.stop({ file: registration })
 
   try {
@@ -196,8 +211,7 @@ test("stop waits for the original process while preserving a newly registered su
     await stopping
 
     expect(() => process.kill(original.pid, 0)).toThrow()
-    const listener = Bun.serve({ port: Number(new URL(original.url).port), fetch: () => new Response() })
-    await listener.stop(true)
+    await expectPortAvailable(original.url)
     expect(await Bun.file(registration).json()).toEqual(replacement)
     expect(await fetch(new URL("/api/info", replacement.url)).then((response) => response.json())).toMatchObject({
       pid: successor.pid,

packages/client/test/service.test.ts

@@ -3,7 +3,7 @@ import { expect, test } from "bun:test"
 import { Effect, FileSystem } from "effect"
 import { writeFile } from "node:fs/promises"
 import { Service, type EnsureReason } from "../src/effect/service"
-import { serviceFixture } from "./fixture/service-fixture"
+import { expectPortAvailable, serviceFixture } from "./fixture/service-fixture"
 import { accelerate } from "./fixture/service-timing"
 
 const ensure = accelerate(Service.ensure)
@@ -162,7 +162,7 @@ test("signals an unresponsive registered service process", async () => {
   expect(await Bun.file(registration).exists()).toBe(false)
 })
 
-test("stop outlives a server that unregisters before releasing its port", async () => {
+test("stop escalates when the registration disappears before the process exits", async () => {
   await using fixture = await serviceFixture()
   const registration = fixture.registration
   const existing = fixture.spawn("lingering", "5000")
@@ -173,17 +173,33 @@ test("stop outlives a server that unregisters before releasing its port", async
 
   expect(await Bun.file(registration + ".signal").text()).toBe("SIGTERM")
   expect(() => process.kill(original.pid, 0)).toThrow()
-  await expectPortAvailable(Number(new URL(original.url).port))
+  await expectPortAvailable(original.url)
   expect(await Bun.file(registration).exists()).toBe(false)
   await existing.exited
 }, 15_000)
 
+test.skipIf(process.platform === "win32")(
+  "stop fails when the process survives SIGKILL",
+  async () => {
+    await using fixture = await serviceFixture()
+    const registration = fixture.registration
+    fixture.spawnUnreaped("lingering", "60000")
+    await fixture.waitForFile()
+    const original = await Bun.file(registration).json()
+
+    await expect(run(stop({ file: registration }))).rejects.toThrow(`Server process ${original.pid} is still running`)
+    expect(await Bun.file(registration + ".signal").text()).toBe("SIGTERM")
+  },
+  15_000,
+)
+
 test("stop waits for the original process while preserving a newly registered successor", async () => {
   await using fixture = await serviceFixture()
   const registration = fixture.registration
   const existing = fixture.spawn("lingering", "15000")
   await fixture.waitForFile()
   const original = await Bun.file(registration).json()
+  // Default timing keeps the grace period open long enough for the successor to register first.
   const stopping = run(Service.stop({ file: registration }))
 
   try {
@@ -197,7 +213,7 @@ test("stop waits for the original process while preserving a newly registered su
     await stopping
 
     expect(() => process.kill(original.pid, 0)).toThrow()
-    await expectPortAvailable(Number(new URL(original.url).port))
+    await expectPortAvailable(original.url)
     expect(await Bun.file(registration).json()).toEqual(replacement)
     expect(await status(replacement.url)).toMatchObject({ pid: successor.pid })
     expect(successor.exitCode).toBe(null)
@@ -348,8 +364,3 @@ function run<A, E>(effect: Effect.Effect<A, E, FileSystem.FileSystem>) {
 async function status(url: string) {
   return fetch(new URL("/api/info", url), { signal: AbortSignal.timeout(1_000) }).then((response) => response.json())
 }
-
-async function expectPortAvailable(port: number) {
-  const server = Bun.serve({ port, fetch: () => new Response() })
-  await server.stop(true)
-}