From 6f920d08bb3b11d06fa75b6d1c9b849e917b6321 Mon Sep 17 00:00:00 2001 From: Dax Raad <826656+thdxr@users.noreply.github.com> Date: Fri, 17 Jul 2026 23:44:37 +0000 Subject: [PATCH] fix(cli): simplify service registration lease --- packages/cli/src/server-process.ts | 58 +++++----- packages/cli/test/service.test.ts | 153 +++++++++++++++++++++----- packages/client/src/effect/service.ts | 1 - packages/client/src/service.ts | 2 - packages/server/src/process.ts | 36 +++--- 5 files changed, 173 insertions(+), 77 deletions(-) diff --git a/packages/cli/src/server-process.ts b/packages/cli/src/server-process.ts index 827979f54dea..13cfaaed0814 100644 --- a/packages/cli/src/server-process.ts +++ b/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, +) { 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) { diff --git a/packages/cli/test/service.test.ts b/packages/cli/test/service.test.ts index b4746e16ca44..28ec5aadbdd8 100644 --- a/packages/cli/test/service.test.ts +++ b/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>) { + 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) +} diff --git a/packages/client/src/effect/service.ts b/packages/client/src/effect/service.ts index 7255bbf74d9f..de452a2843d3 100644 --- a/packages/client/src/effect/service.ts +++ b/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)) diff --git a/packages/client/src/service.ts b/packages/client/src/service.ts index f4df10bf9c3f..96d54871f609 100644 --- a/packages/client/src/service.ts +++ b/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 } diff --git a/packages/server/src/process.ts b/packages/server/src/process.ts index ec375f14cdfe..8fddbe3e2aea 100644 --- a/packages/server/src/process.ts +++ b/packages/server/src/process.ts @@ -21,7 +21,10 @@ export type Options = { readonly password: string readonly instanceID: string readonly service?: { - readonly onListen: (address: HttpServer.Address) => Effect.Effect + readonly onListen: ( + address: HttpServer.Address, + shutdown: Effect.Effect, + ) => Effect.Effect, E, R> } } @@ -44,17 +47,21 @@ export const start = Effect.fn("ServerProcess.start")(function* (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) {