fix(cli): simplify service registration lease (#37576)

Co-authored-by: Dax Raad <826656+thdxr@users.noreply.github.com>

packages/cli/src/server-process.ts

@@ -1,7 +1,7 @@
 export * as ServerProcess from "./server-process"
 
 import { NodeServices } from "@effect/platform-node"
-import { Service, type DiscoverOptions } from "@opencode-ai/client/effect/service"
+import { Service, type DiscoverOptions, type Info } from "@opencode-ai/client/effect/service"
 import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
 import { LayerNode } from "@opencode-ai/core/effect/layer-node"
 import { Global } from "@opencode-ai/core/global"
@@ -69,7 +69,9 @@ const processEffect = Effect.fnUntraced(function* (options: Options) {
         service:
           serviceOptions === undefined
             ? undefined
-            : { onListen: (address) => register(address, password, instanceID, serviceOptions.file) },
+            : {
+                onListen: (address, shutdown) => register(address, password, instanceID, serviceOptions.file, shutdown),
+              },
       }).pipe(
         Effect.provide(Logger.layer([], { mergeWithExisting: false })),
         Effect.catch((error) => {
@@ -108,54 +110,46 @@ const infoJson = Schema.fromJsonString(Service.Info)
 const encodeInfo = Schema.encodeEffect(infoJson)
 const decodeInfo = Schema.decodeUnknownEffect(infoJson)
 
-const register = Effect.fnUntraced(function* (address: HttpServer.Address, password: string, id: string, file: string) {
+const register = Effect.fnUntraced(function* (
+  address: HttpServer.Address,
+  password: string,
+  id: string,
+  file: string,
+  shutdown: Effect.Effect<void>,
+) {
   const fs = yield* FileSystem.FileSystem
   const temp = file + "." + id + ".tmp"
   yield* fs.makeDirectory(path.dirname(file), { recursive: true })
-  const previous = yield* fs.readFileString(file).pipe(
-    Effect.flatMap(decodeInfo),
-    Effect.orElseSucceed(() => undefined),
-  )
   const info = {
     id,
     version: InstallationVersion,
     url: HttpServer.formatAddress(address),
     pid: process.pid,
     password,
-    startedAt: Math.max(Date.now(), (previous?.startedAt ?? 0) + 1),
   }
   const encoded = yield* encodeInfo(info)
-  const publish = fs.writeFileString(temp, encoded, { mode: 0o600 }).pipe(Effect.andThen(fs.rename(temp, file)))
-  yield* publish
   const current = fs.readFileString(file).pipe(
     Effect.flatMap(decodeInfo),
     Effect.orElseSucceed(() => undefined),
   )
-  const assertRegistration = Effect.gen(function* () {
-    const found = yield* current
-    if (
-      found !== undefined &&
-      found.id === info.id &&
-      found.version === info.version &&
-      found.url === info.url &&
-      found.pid === info.pid &&
-      found.password === info.password
-    )
-      return
-    if (found?.startedAt !== undefined && found.startedAt >= info.startedAt) return
-    yield* publish
-  })
-  yield* Effect.addFinalizer(() =>
-    current.pipe(
-      Effect.flatMap((current) => (current?.id === id ? fs.remove(file) : Effect.void)),
-      Effect.ignore,
-    ),
-  )
-  yield* assertRegistration.pipe(
-    Effect.catchCause((cause) => Effect.logWarning("failed to reassert service registration", { cause })),
+  const owns = (found: Info | undefined) =>
+    found?.id === info.id &&
+    found.version === info.version &&
+    found.url === info.url &&
+    found.pid === info.pid &&
+    found.password === info.password
+  yield* fs.writeFileString(temp, encoded, { mode: 0o600 }).pipe(Effect.andThen(fs.rename(temp, file)))
+  yield* current.pipe(
+    Effect.filterOrFail(owns),
     Effect.repeat(Schedule.spaced("5 seconds")),
+    Effect.ignore,
+    Effect.andThen(shutdown),
     Effect.forkScoped,
   )
+  return current.pipe(
+    Effect.flatMap((found) => (owns(found) ? fs.remove(file) : Effect.void)),
+    Effect.ignore,
+  )
 })
 
 const recognizeIncumbent = Effect.fnUntraced(function* (options: DiscoverOptions, hostname: string, port: number) {

packages/cli/test/service.test.ts

@@ -97,6 +97,80 @@ test("preview registration migration never moves stable discovery", async () =>
   }
 })
 
+test("managed service writes its registration once", async () => {
+  const service = await startManagedService("opencode-service-once-")
+  try {
+    const before = await fs.stat(service.registration)
+    await Bun.sleep(6_000)
+    const after = await fs.stat(service.registration)
+    expect(after.ino).toBe(before.ino)
+    expect(after.mtimeMs).toBe(before.mtimeMs)
+    expect(await Bun.file(service.registration).json()).toEqual(service.info)
+  } finally {
+    await stopManagedService(service)
+  }
+}, 30_000)
+
+test("deleting a managed service registration stops its owner", async () => {
+  const service = await startManagedService("opencode-service-delete-")
+  try {
+    await fs.rm(service.registration)
+    expect(await waitForExit(service.owner)).toBe(true)
+    expect(await Bun.file(service.registration).exists()).toBe(false)
+    await expectPortAvailable(service.port)
+  } finally {
+    await stopManagedService(service)
+  }
+}, 30_000)
+
+test("deleting a failed service registration stops its owner", async () => {
+  const service = await startManagedService("opencode-service-failed-delete-", true)
+  try {
+    await waitForFailed(service.info)
+    await fs.rm(service.registration)
+    expect(await waitForExit(service.owner)).toBe(true)
+    await expectPortAvailable(service.port)
+  } finally {
+    await stopManagedService(service)
+  }
+}, 30_000)
+
+test("corrupting a managed service registration stops its owner", async () => {
+  const service = await startManagedService("opencode-service-corrupt-")
+  try {
+    await fs.writeFile(service.registration, "not-json")
+    expect(await waitForExit(service.owner)).toBe(true)
+    expect(await Bun.file(service.registration).text()).toBe("not-json")
+    await expectPortAvailable(service.port)
+  } finally {
+    await stopManagedService(service)
+  }
+}, 30_000)
+
+test("replacing a managed service registration stops its owner and preserves the foreign owner", async () => {
+  const service = await startManagedService("opencode-service-foreign-")
+  const foreign = { ...service.info, id: "foreign-owner", pid: process.pid }
+  try {
+    await fs.writeFile(service.registration, JSON.stringify(foreign))
+    expect(await waitForExit(service.owner)).toBe(true)
+    expect(await Bun.file(service.registration).json()).toEqual(foreign)
+    await expectPortAvailable(service.port)
+  } finally {
+    await stopManagedService(service)
+  }
+}, 30_000)
+
+test("clean managed service shutdown removes its registration", async () => {
+  const service = await startManagedService("opencode-service-clean-")
+  try {
+    await Effect.runPromise(Service.stop({ file: service.registration }).pipe(Effect.provide(NodeFileSystem.layer)))
+    expect(await waitForExit(service.owner)).toBe(true)
+    expect(await Bun.file(service.registration).exists()).toBe(false)
+  } finally {
+    await stopManagedService(service)
+  }
+}, 30_000)
+
 test("concurrent service processes elect one server", async () => {
   const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-election-"))
   const database = path.join(root, "opencode.db")
@@ -163,35 +237,6 @@ test("concurrent service processes elect one server", async () => {
       version: info.version,
       pid: info.pid,
     })
-    const blockedTemp = registration + "." + info.id + ".tmp"
-    await fs.mkdir(blockedTemp)
-    await fs.rm(registration)
-    const repairContender = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" })
-    await Bun.sleep(3_000)
-    expect(await Bun.file(registration).exists()).toBe(false)
-    await fs.rm(blockedTemp, { recursive: true })
-    expect(await Promise.race([repairContender.exited.then(() => true), Bun.sleep(15_000).then(() => false)])).toBe(
-      true,
-    )
-    expect(repairContender.exitCode).toBe(0)
-    const restored = await waitForInfo(registration)
-    expect(restored.id).toBe(info.id)
-    expect(restored.pid).toBe(info.pid)
-    await fs.writeFile(registration, "not-json")
-    const repaired = await waitForInfo(registration)
-    expect(repaired.id).toBe(info.id)
-    expect(repaired.pid).toBe(info.pid)
-    await fs.writeFile(
-      registration,
-      JSON.stringify({ ...info, id: "older-orphan", pid: process.pid, startedAt: info.startedAt! - 1 }),
-    )
-    const reclaimed = await waitForInfo(registration, (value) => value.id === info.id)
-    expect(reclaimed.pid).toBe(info.pid)
-    await fs.writeFile(registration, JSON.stringify({ ...info, id: "newer-owner", startedAt: info.startedAt! + 1 }))
-    await Bun.sleep(6_000)
-    expect((await waitForInfo(registration)).id).toBe("newer-owner")
-    await fs.writeFile(registration, JSON.stringify(info))
-
     const contender = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" })
     try {
       const contenderExited = await Promise.race([
@@ -395,10 +440,24 @@ async function waitForInfo(file: string, accept: (info: Info) => boolean = () =>
   throw new Error("Timed out waiting for service registration")
 }
 
+async function waitForFailed(info: Info) {
+  for (let attempt = 0; attempt < 400; attempt++) {
+    const status = await fetch(new URL("/api/health", info.url), {
+      headers: { authorization: "Basic " + btoa(`opencode:${info.password}`) },
+    })
+      .then((response) => response.status)
+      .catch(() => undefined)
+    if (status === 500) return
+    await Bun.sleep(50)
+  }
+  throw new Error("Timed out waiting for service boot failure")
+}
+
 async function availablePort() {
   const server = Bun.serve({ port: 0, fetch: () => new Response() })
   const port = server.port
   await server.stop(true)
+  if (port === undefined) throw new Error("Server did not bind a port")
   return port
 }
 
@@ -414,3 +473,39 @@ function serviceEnv(root: string) {
     XDG_STATE_HOME: path.join(root, "state"),
   }
 }
+
+async function startManagedService(prefix: string, failBoot = false) {
+  const root = await fs.mkdtemp(path.join(os.tmpdir(), prefix))
+  const port = await availablePort()
+  const registration = path.join(root, "state", "opencode", "service-local.json")
+  await fs.mkdir(path.join(root, "config", "opencode"), { recursive: true })
+  if (failBoot) await fs.mkdir(path.join(root, "database"))
+  await fs.writeFile(path.join(root, "config", "opencode", "service-local.json"), JSON.stringify({ port }))
+  const owner = Bun.spawn([process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"], {
+    env: failBoot ? { ...serviceEnv(root), OPENCODE_DB: path.join(root, "database") } : serviceEnv(root),
+    stderr: "pipe",
+    stdout: "ignore",
+  })
+  const info = await waitForInfo(registration).catch(async (cause) => {
+    owner.kill("SIGTERM")
+    await owner.exited
+    await fs.rm(root, { recursive: true, force: true })
+    throw cause
+  })
+  return { root, port, registration, owner, info }
+}
+
+async function stopManagedService(service: Awaited<ReturnType<typeof startManagedService>>) {
+  service.owner.kill("SIGTERM")
+  await service.owner.exited
+  await fs.rm(service.root, { recursive: true, force: true })
+}
+
+function waitForExit(process: Bun.Subprocess, timeout = 10_000) {
+  return Promise.race([process.exited.then(() => true), Bun.sleep(timeout).then(() => false)])
+}
+
+async function expectPortAvailable(port: number) {
+  const server = Bun.serve({ hostname: "127.0.0.1", port, fetch: () => new Response() })
+  await server.stop(true)
+}

packages/client/src/effect/service.ts

@@ -161,7 +161,6 @@ export const Info = Schema.Struct({
   url: Schema.String,
   pid: Schema.Int.check(Schema.isGreaterThan(0)),
   password: Schema.optional(Schema.String),
-  startedAt: Schema.optional(Schema.Int.check(Schema.isGreaterThan(0))),
 })
 
 const decode = Schema.decodeUnknownEffect(Schema.fromJsonString(Info))

packages/client/src/service.ts

@@ -50,6 +50,4 @@ export type Info = {
   readonly pid: number
   /** Private service password, when authentication is enabled. */
   readonly password?: string
-  /** Registration generation used to resolve owner write races. */
-  readonly startedAt?: number
 }

packages/server/src/process.ts

@@ -21,7 +21,10 @@ export type Options<E = never, R = never> = {
   readonly password: string
   readonly instanceID: string
   readonly service?: {
-    readonly onListen: (address: HttpServer.Address) => Effect.Effect<void, E, R>
+    readonly onListen: (
+      address: HttpServer.Address,
+      shutdown: Effect.Effect<void>,
+    ) => Effect.Effect<Effect.Effect<void>, E, R>
   }
 }
 
@@ -44,17 +47,21 @@ export const start = Effect.fn("ServerProcess.start")(function* <E, R>(options:
   yield* bound.http
     .serve(dispatch(options.password, status, application, shutdown), HttpMiddleware.logger)
     .pipe(withoutParentSpan)
-  if (options.service) yield* options.service.onListen(bound.http.address)
+  if (options.service)
+    yield* options.service.onListen(bound.http.address, Deferred.succeed(shutdown, undefined).pipe(Effect.asVoid)).pipe(
+      Effect.flatMap((cleanup) =>
+        Effect.addFinalizer(() => Scope.close(bound.scope, Exit.void).pipe(Effect.andThen(cleanup))),
+      ),
+      Effect.uninterruptible,
+    )
 
   const parentScope = yield* Scope.Scope
   const applicationScope = yield* Scope.fork(parentScope)
   yield* Effect.addFinalizer(() =>
-    status
-      .beginStopping
-      .pipe(
-        Effect.andThen(Ref.set(application, Option.none())),
-        Effect.andThen(Effect.sync(() => bound.server.closeAllConnections())),
-      ),
+    status.beginStopping.pipe(
+      Effect.andThen(Ref.set(application, Option.none())),
+      Effect.andThen(Effect.sync(() => bound.server.closeAllConnections())),
+    ),
   )
 
   const boot = Effect.gen(function* () {
@@ -112,7 +119,7 @@ function bind(hostname: string, port: number) {
     return yield* Effect.gen(function* () {
       const http = yield* NodeHttpServer.make(() => server, { port, host: hostname })
       yield* Effect.addFinalizer(() => Effect.sync(() => server.closeAllConnections()))
-      return { http, server }
+      return { http, server, scope: serverScope }
     }).pipe(
       Effect.provideService(Scope.Scope, serverScope),
       Effect.onError((cause) => Scope.close(serverScope, Exit.failCause(cause))),
@@ -190,10 +197,13 @@ const control = Effect.fnUntraced(function* (
 
 const healthResponse = Effect.fnUntraced(function* (status: Status.Interface) {
   const state = yield* status.current
-  return HttpServerResponse.jsonUnsafe({ healthy: true, version: InstallationVersion, pid: process.pid }, {
-    status: state.type === "ready" ? 200 : state.type === "failed" ? 500 : 503,
-    headers: state.type === "starting" || state.type === "stopping" ? { "retry-after": "1" } : undefined,
-  })
+  return HttpServerResponse.jsonUnsafe(
+    { healthy: true, version: InstallationVersion, pid: process.pid },
+    {
+      status: state.type === "ready" ? 200 : state.type === "failed" ? 500 : 503,
+      headers: state.type === "starting" || state.type === "stopping" ? { "retry-after": "1" } : undefined,
+    },
+  )
 })
 
 function unavailable(status: Status.State) {