fix(client): require authenticated service stop

packages/client/src/effect/service.ts

@@ -56,7 +56,6 @@ const discoverLocal = Effect.fnUntraced(function* (options: DiscoverOptions) {
 export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOptions = {}) {
   const timing = ensureTiming(options)
   const contenders = new Set<ServiceContender>()
-  let timeouts: { readonly info: Info; readonly count: number } | undefined
   let announced = false
   let lastSpawn = 0
   let spawnDelay = timing.spawnDelay
@@ -80,18 +79,6 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
     const registration = yield* registered(options.file, true, timing.requestTimeout)
     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, timing)
-        timeouts = undefined
-        lastSpawn = Date.now() - spawnDelay
-      }
-    } else timeouts = undefined
     if (service !== undefined) {
       spawnDelay = timing.spawnDelay
       const compatible = !service.legacy && matchesVersion(service.version, options)
@@ -99,8 +86,8 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
       if (compatible && service.state === "failed")
         return yield* Effect.fail(new Error("Background service failed to start"))
       if (compatible) return Option.none<LocalService>()
+      yield* kill(service, timing)
       yield* announce("version-mismatch", service.version)
-      yield* kill(service, options, timing).pipe(Effect.ignore)
       lastSpawn = 0
       return Option.none<LocalService>()
     } else if (lastSpawn === 0 && info !== undefined) lastSpawn = Date.now()
@@ -133,8 +120,12 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
 
 /** Stop the registered local service. */
 export const stop = Effect.fn("service.stop")(function* (options: StopOptions = {}) {
-  const existing = yield* find(options)
-  if (existing !== undefined) yield* kill(existing, options, defaultEnsureTiming)
+  const registration = yield* registered(options.file, true)
+  if (registration.service !== undefined) yield* kill(registration.service, defaultEnsureTiming)
+  if (registration.service === undefined && registration.info !== undefined)
+    return yield* Effect.fail(
+      new Error("Background service is not responding; stop its process manually and try again"),
+    )
 })
 
 function fallback() {
@@ -243,20 +234,10 @@ const registered = Effect.fnUntraced(function* (file?: string, allowLegacy = fal
   return { info, ...(yield* probeResult(info, allowLegacy, timeout)) }
 })
 
-// Health-checked lookup without the version gate: lifecycle operations must be
-// able to see (and replace or stop) a server from a different version.
-const find = Effect.fnUntraced(function* (options: { readonly file?: string }) {
-  return (yield* registered(options.file, true)).service
-})
-
-// 50ms cadence bounded at ~5s, shared by stop escalation and each ensure
-// discovery window.
+// Poll until an authenticated stop exits, bounded by the configured stop window.
 const poll = (timing: EnsureTiming) =>
   Schedule.max([Schedule.spaced(timing.stopPollInterval), Schedule.recurs(timing.stopPollAttempts)])
 
-const signal = (pid: number, name: NodeJS.Signals) =>
-  Effect.try({ try: () => process.kill(pid, name), catch: (cause) => cause }).pipe(Effect.ignore)
-
 const stopped = Effect.fnUntraced(function* (pid: number) {
   const running = yield* Effect.try({ try: () => process.kill(pid, 0), catch: () => false }).pipe(
     Effect.orElseSucceed(() => false),
@@ -265,44 +246,14 @@ const stopped = Effect.fnUntraced(function* (pid: number) {
   return yield* Effect.fail(new Error(`Server process ${pid} is still running`))
 })
 
-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 }, timing: EnsureTiming) {
-  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(timing)), 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(timing)))
-})
-
-const kill = Effect.fnUntraced(function* (
-  service: LocalService,
-  options: { readonly file?: string },
-  timing: EnsureTiming,
-) {
+const kill = Effect.fnUntraced(function* (service: LocalService, timing: EnsureTiming) {
   const requested = yield* requestStop(service, timing.requestTimeout)
-  if (requested === "rejected") return
-  if (requested === "unsupported") {
-    // A stale registration may point at a reused PID. Authenticate again
-    // immediately before the legacy signal fallback.
-    const current = yield* find(options)
-    if (current === undefined || !same(current.info, service.info)) return
-    yield* signal(service.info.pid, "SIGTERM")
-  }
+  if (requested === "rejected") return yield* Effect.fail(new Error("Background service rejected the stop request"))
+  if (requested === "unsupported")
+    return yield* Effect.fail(new Error("Background service does not support authenticated stop requests"))
   const done = yield* stopped(service.info.pid).pipe(Effect.retry(poll(timing)), Effect.option)
   if (Option.isSome(done)) return
-
-  const latest = yield* find(options)
-  if (latest === undefined || !same(latest.info, service.info)) return
-  yield* signal(service.info.pid, "SIGKILL")
-  yield* stopped(service.info.pid).pipe(Effect.retry(poll(timing)))
+  return yield* Effect.fail(new Error("Background service accepted the stop request but did not exit"))
 })
 
 const decodeStopResponse = Schema.decodeUnknownOption(ServiceStatus.StopResponse)
@@ -317,7 +268,8 @@ const requestStop = Effect.fnUntraced(function* (service: LocalService, timeout
       signal: AbortSignal.timeout(timeout),
     }),
   ).pipe(Effect.option, Effect.map(Option.getOrUndefined))
-  if (response === undefined || response.status === 404 || response.status === 405) return "unsupported" as const
+  if (response === undefined) return "rejected" as const
+  if (response.status === 404 || response.status === 405) return "unsupported" as const
   const body = yield* Effect.tryPromise(() => response.json()).pipe(Effect.option, Effect.map(Option.getOrUndefined))
   const decoded = decodeStopResponse(body)
   if (!response.ok || Option.isNone(decoded) || !decoded.value.accepted) return "rejected" as const

packages/client/src/promise/service.ts

@@ -37,7 +37,6 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
   const timing = ensureTiming(options)
   const deadline = Date.now() + timing.promiseTimeout
   const contenders = new Set<ServiceContender>()
-  let timeouts: { readonly info: Info; readonly count: number } | undefined
   let announced = false
   let lastSpawn = 0
   let spawnDelay = timing.spawnDelay
@@ -61,19 +60,6 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
     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, timing.requestTimeout)
-      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, timing)
-          timeouts = undefined
-          lastSpawn = Date.now() - spawnDelay
-        }
-      } else timeouts = undefined
-
       if (registration.service !== undefined) {
         spawnDelay = timing.spawnDelay
         const service = registration.service
@@ -81,8 +67,8 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
         if (compatible && service.state === "ready") return service.endpoint
         if (compatible && service.state === "failed") throw new Error("Background service failed to start")
         if (!compatible) {
+          await kill(service, timing)
           announce("version-mismatch", service.version)
-          await kill(service, options, timing).catch(() => undefined)
           lastSpawn = 0
         }
       } else {
@@ -110,8 +96,10 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
 
 /** Stop the registered local service. */
 export async function stop(options: StopOptions = {}) {
-  const existing = await find(options)
-  if (existing !== undefined) await kill(existing, options, defaultEnsureTiming)
+  const registration = await registered(options.file, true)
+  if (registration.service !== undefined) await kill(registration.service, defaultEnsureTiming)
+  if (registration.service === undefined && registration.info !== undefined)
+    throw new Error("Background service is not responding; stop its process manually and try again")
 }
 
 function fallback() {
@@ -199,16 +187,6 @@ async function registered(file?: string, allowLegacy = false, timeout?: number)
   return { info, ...(await probeResult(info, allowLegacy, timeout)) }
 }
 
-async function find(options: { readonly file?: string }) {
-  return (await registered(options.file, true)).service
-}
-
-function signal(pid: number, name: NodeJS.Signals) {
-  try {
-    process.kill(pid, name)
-  } catch {}
-}
-
 function stopped(pid: number) {
   try {
     process.kill(pid, 0)
@@ -226,37 +204,12 @@ async function waitUntilStopped(pid: number, timing: EnsureTiming) {
   return false
 }
 
-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 }, timing: EnsureTiming) {
-  const current = await read(options.file)
-  if (current === undefined || !same(current, info)) return
-  signal(info.pid, "SIGTERM")
-  if (await waitUntilStopped(info.pid, timing)) return
-
-  const latest = await read(options.file)
-  if (latest === undefined || !same(latest, info)) return
-  signal(info.pid, "SIGKILL")
-  if (!(await waitUntilStopped(info.pid, timing))) throw new Error(`Server process ${info.pid} is still running`)
-}
-
-async function kill(service: LocalService, options: { readonly file?: string }, timing: EnsureTiming) {
+async function kill(service: LocalService, timing: EnsureTiming) {
   const requested = await requestStop(service, timing.requestTimeout)
-  if (requested === "rejected") return
-  if (requested === "unsupported") {
-    const current = await find(options)
-    if (current === undefined || !same(current.info, service.info)) return
-    signal(service.info.pid, "SIGTERM")
-  }
+  if (requested === "rejected") throw new Error("Background service rejected the stop request")
+  if (requested === "unsupported") throw new Error("Background service does not support authenticated stop requests")
   if (await waitUntilStopped(service.info.pid, timing)) return
-
-  const latest = await find(options)
-  if (latest === undefined || !same(latest.info, service.info)) return
-  signal(service.info.pid, "SIGKILL")
-  if (!(await waitUntilStopped(service.info.pid, timing)))
-    throw new Error(`Server process ${service.info.pid} is still running`)
+  throw new Error("Background service accepted the stop request but did not exit")
 }
 
 async function requestStop(service: LocalService, timeout = defaultEnsureTiming.requestTimeout) {
@@ -267,7 +220,8 @@ async function requestStop(service: LocalService, timeout = defaultEnsureTiming.
     body: JSON.stringify({ instanceID: service.info.id }),
     signal: AbortSignal.timeout(timeout),
   }).catch(() => undefined)
-  if (response === undefined || response.status === 404 || response.status === 405) return "unsupported" as const
+  if (response === undefined) return "rejected" as const
+  if (response.status === 404 || response.status === 405) return "unsupported" as const
   const body = (await response.json().catch(() => undefined)) as ServiceStopResponse | undefined
   if (!response.ok || body?.accepted !== true) return "rejected" as const
   return "accepted" as const

packages/client/test/fixture/service.ts

@@ -28,7 +28,7 @@ if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated" ||
 
 let requests = 0
 let version = "test"
-if (mode === "old" || mode === "reject-stop") version = "old"
+if (mode === "old" || mode === "reject-stop" || mode === "stop-hanging" || mode === "stop-accepted-hanging") version = "old"
 if (mode === "incompatible") version = "1.9.0"
 if (mode === "compatible" || mode === "delayed-compatible") version = "2.1.0-next.1"
 const id = crypto.randomUUID()
@@ -40,7 +40,15 @@ const server = Bun.serve({
       await appendFile(registration + ".stop-attempts", process.pid + "\n")
       return Response.json({ accepted: false })
     }
-    if (pathname === "/api/service/stop" && mode === "graceful") {
+    if (pathname === "/api/service/stop" && mode === "stop-hanging") {
+      await appendFile(registration + ".stop-attempts", process.pid + "\n")
+      return new Promise<Response>(() => {})
+    }
+    if (pathname === "/api/service/stop" && mode === "stop-accepted-hanging") {
+      await appendFile(registration + ".stop-attempts", process.pid + "\n")
+      return Response.json({ accepted: true })
+    }
+    if (pathname === "/api/service/stop" && (mode === "graceful" || mode === "old" || mode === "incompatible")) {
       const body = await request.json()
       if (typeof body !== "object" || body === null || body.instanceID !== id) return Response.json({ accepted: false })
       await writeFile(registration + ".stop", JSON.stringify(body))
@@ -63,7 +71,13 @@ const server = Bun.serve({
     if (mode === "starting" && !(await Bun.file(registration + ".release").exists()))
       return Response.json({ healthy: true, version, pid: process.pid }, { status: 503 })
     if (mode === "failed-owner") return Response.json({ healthy: true, version, pid: process.pid }, { status: 500 })
-    if (mode === "starting" || mode === "graceful" || mode === "reject-stop")
+    if (
+      mode === "starting" ||
+      mode === "graceful" ||
+      mode === "reject-stop" ||
+      mode === "stop-hanging" ||
+      mode === "stop-accepted-hanging"
+    )
       return Response.json({ healthy: true, version, pid: process.pid })
     return Response.json({ healthy: true, version, pid: process.pid })
   },

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

@@ -100,7 +100,7 @@ test("reports a bounded contender stderr tail with native promises", async () =>
   expect(error.message.length).toBeLessThan(9_000)
 }, 10_000)
 
-test("evicts an unresponsive registered service before starting its replacement", async () => {
+test("never evicts an unresponsive registered service automatically", async () => {
   const directory = await temp()
   const registration = join(directory, "service.json")
   const existing = Bun.spawn([process.execPath, fixture, registration, "hanging"], {
@@ -111,19 +111,25 @@ test("evicts an unresponsive registered service before starting its replacement"
   await waitForFile(registration)
   const original = await Bun.file(registration).json()
 
-  const endpoint = await ensure({
+  const result = ensure({
     file: registration,
     version: "test",
-    command: [process.execPath, fixture, registration, "delayed", "10"],
+    command: [process.execPath, fixture, registration, "record-start"],
   })
-  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)
+  await waitForLines(registration + ".requests", 3)
+
+  expect(existing.exitCode).toBe(null)
+  expect(await Bun.file(registration).json()).toEqual(original)
+  await expect(result).rejects.toThrow()
+})
+
+test("explicit native stop refuses to signal an unidentified unresponsive PID", async () => {
+  const registration = await setup("hanging")
+  const info = await Bun.file(registration).json()
+
+  await expect(Service.stop({ file: registration })).rejects.toThrow("stop its process manually")
+
+  expect(process.kill(info.pid, 0)).toBe(true)
 })
 
 test("requests graceful stop of the exact service instance", async () => {
@@ -135,6 +141,29 @@ test("requests graceful stop of the exact service instance", async () => {
   expect(await Bun.file(registration + ".stop").json()).toEqual({ instanceID: info.id })
 })
 
+test.each([
+  ["reject-stop", "rejected the stop request"],
+  ["stop-hanging", "rejected the stop request"],
+  ["stop-accepted-hanging", "accepted the stop request but did not exit"],
+  ["legacy", "does not support authenticated stop requests"],
+])("native replacement fails closed for %s service stop", async (mode, message) => {
+  const registration = await setup(mode)
+  const directory = await temp()
+  const contender = join(directory, "contender.json")
+  const info = await Bun.file(registration).json()
+
+  await expect(
+    ensure({
+      file: registration,
+      version: "test",
+      command: [process.execPath, fixture, contender, "record-start"],
+    }),
+  ).rejects.toThrow(message)
+
+  expect(await Bun.file(contender + ".started").exists()).toBe(false)
+  expect(process.kill(info.pid, 0)).toBe(true)
+})
+
 async function setup(mode: string) {
   const directory = await temp()
   const registration = join(directory, "service.json")
@@ -156,3 +185,14 @@ async function waitForFile(file: string) {
   }
   throw new Error(`Timed out waiting for ${file}`)
 }
+
+async function waitForLines(file: string, count: number) {
+  for (let attempt = 0; attempt < 600; attempt++) {
+    const text = await Bun.file(file)
+      .text()
+      .catch(() => "")
+    if (text.trim().split("\n").length >= count) return
+    await Bun.sleep(5)
+  }
+  throw new Error(`Timed out waiting for ${count} lines in ${file}`)
+}

packages/client/test/service.test.ts

@@ -118,29 +118,39 @@ 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 () => {
+test("never evicts an unresponsive registered service automatically", 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(
+  const controller = new AbortController()
+  const result = Effect.runPromise(
     ensure({
       file: registration,
       version: "test",
-      command: [process.execPath, fixture, registration, "delayed", "10"],
-    }),
+      command: [process.execPath, fixture, registration, "record-start"],
+    }).pipe(Effect.provide(NodeFileSystem.layer)),
+    { signal: controller.signal },
   )
-  const replacement = await Bun.file(registration).json()
+  await waitForLines(registration + ".requests", 3)
+  controller.abort()
+  await result.catch(() => undefined)
 
-  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")
-  await waitForExit(replacement.pid)
+  expect(existing.exitCode).toBe(null)
+  expect(await Bun.file(registration).json()).toEqual(original)
+})
+
+test("explicit stop refuses to signal an unidentified unresponsive PID", async () => {
+  const directory = await temp()
+  const registration = join(directory, "service.json")
+  const existing = spawn(registration, "hanging")
+  await waitForFile(registration)
+
+  await expect(run(Service.stop({ file: registration }))).rejects.toThrow("stop its process manually")
+
+  expect(existing.exitCode).toBe(null)
 })
 
 test("requests graceful stop of the exact service instance", async () => {
@@ -161,36 +171,39 @@ test("does not spawn contenders while an incompatible service rejects replacemen
   const contender = join(directory, "contender.json")
   const existing = spawn(registration, "reject-stop")
   await waitForFile(registration)
-  const controller = new AbortController()
-  const starting = Effect.runPromise(
+  const starting = run(
     ensure({
       file: registration,
       version: "test",
       command: [process.execPath, fixture, contender, "record-start"],
-    }).pipe(Effect.provide(NodeFileSystem.layer)),
-    { signal: controller.signal },
+    }),
   )
 
-  await waitForLines(registration + ".stop-attempts", 2)
-  controller.abort()
-  await starting.catch(() => undefined)
+  await expect(starting).rejects.toThrow("Background service rejected the stop request")
 
   expect(await Bun.file(contender + ".started").exists()).toBe(false)
+  expect((await Bun.file(registration + ".stop-attempts").text()).trim().split("\n")).toHaveLength(1)
   expect(existing.exitCode).toBe(null)
 })
 
-test("a legacy health response is still replaced", async () => {
+test.each([
+  ["stop-hanging", "rejected the stop request"],
+  ["stop-accepted-hanging", "accepted the stop request but did not exit"],
+  ["legacy", "does not support authenticated stop requests"],
+])("replacement fails closed for %s service stop", async (mode, message) => {
   const directory = await temp()
   const registration = join(directory, "service.json")
-  const existing = spawn(registration, "legacy")
+  const contender = join(directory, "contender.json")
+  const existing = spawn(registration, mode)
   await waitForFile(registration)
 
-  const starts: EnsureReason[] = []
-  const result = run(ensure({ file: registration, command: [], onStart: (reason) => starts.push(reason) }))
+  const result = run(
+    ensure({ file: registration, version: "test", command: [process.execPath, fixture, contender, "record-start"] }),
+  )
 
-  await expect(result).rejects.toThrow("Missing service command")
-  expect(starts).toEqual(["version-mismatch"])
-  await existing.exited
+  await expect(result).rejects.toThrow(message)
+  expect(await Bun.file(contender + ".started").exists()).toBe(false)
+  expect(existing.exitCode).toBe(null)
 })
 
 test("waits for a slow winner while bounding lock probes", async () => {