diff --git a/packages/cli/src/server-process.ts b/packages/cli/src/server-process.ts index efd5a38b32..827979f54d 100644 --- a/packages/cli/src/server-process.ts +++ b/packages/cli/src/server-process.ts @@ -1,13 +1,12 @@ export * as ServerProcess from "./server-process" import { NodeServices } from "@effect/platform-node" -import { Service } from "@opencode-ai/client/effect/service" +import { Service, type DiscoverOptions } 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" import { InstallationVersion } from "@opencode-ai/core/installation/version" import { AppProcess } from "@opencode-ai/core/process" -import { ProcessLock } from "@opencode-ai/core/util/process-lock" import { randomBytes, randomUUID } from "node:crypto" import path from "node:path" import { Effect, FileSystem, Logger, Option, Redacted, Schedule, Schema } from "effect" @@ -38,14 +37,15 @@ const processEffect = Effect.fnUntraced(function* (options: Options) { return yield* Effect.scoped( Effect.gen(function* () { const serviceOptions = options.mode === "service" ? yield* ServiceConfig.options() : undefined - if (serviceOptions !== undefined) { - const acquired = yield* ProcessLock.acquire(serviceOptions.file + ".lock").pipe( - Effect.as(true), - Effect.catchTag("ProcessLockHeldError", () => Effect.succeed(false)), - ) - if (!acquired) return yield* Effect.void - if ((yield* Service.discover(serviceOptions)) !== undefined) return yield* Effect.void - } + const config = options.mode === "service" ? yield* ServiceConfig.read() : {} + const hostname = options.hostname ?? config.hostname ?? "127.0.0.1" + const port = options.port ?? config.port ?? (options.mode === "service" ? ServiceConfig.defaultPort() : undefined) + if ( + serviceOptions !== undefined && + port !== undefined && + (yield* Service.incumbent({ ...serviceOptions, url: serviceURL(hostname, port) })) !== undefined + ) + return const { start } = yield* Effect.promise(() => import("@opencode-ai/server/process")) const environmentPassword = yield* Env.password // Keep the lease credential out of the environment inherited by tools. @@ -53,7 +53,6 @@ const processEffect = Effect.fnUntraced(function* (options: Options) { delete process.env.OPENCODE_PASSWORD delete process.env.OPENCODE_SERVER_PASSWORD } - const config = options.mode === "service" ? yield* ServiceConfig.read() : {} const password = options.mode === "service" ? yield* ServiceConfig.password() @@ -63,15 +62,34 @@ const processEffect = Effect.fnUntraced(function* (options: Options) { if (!password) return yield* Effect.fail(new Error("Missing server password")) const instanceID = randomUUID() const server = yield* start({ - hostname: options.hostname ?? config.hostname ?? "127.0.0.1", - port: Option.fromNullishOr(options.port ?? config.port), + hostname, + port: Option.fromNullishOr(port), password, instanceID, service: serviceOptions === undefined ? undefined : { onListen: (address) => register(address, password, instanceID, serviceOptions.file) }, - }).pipe(Effect.provide(Logger.layer([], { mergeWithExisting: false }))) + }).pipe( + Effect.provide(Logger.layer([], { mergeWithExisting: false })), + Effect.catch((error) => { + if (serviceOptions === undefined || port === undefined || !addressInUse(error)) return Effect.fail(error) + return recognizeIncumbent(serviceOptions, hostname, port).pipe( + Effect.flatMap((found) => + found + ? Effect.void + : Effect.fail( + new Error( + `Managed service port ${port} on ${hostname} is already in use by another process. ` + + "Configure another port with `opencode service set port ` and start the service again.", + { cause: error }, + ), + ), + ), + ) + }), + ) + if (server === undefined) return const url = HttpServer.formatAddress(server.address) console.log(options.mode === "stdio" ? JSON.stringify({ url }) : `server listening on ${url}`) if (options.mode === "default" && !environmentPassword) console.log(`server password ${password}`) @@ -90,21 +108,21 @@ 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) { 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))) @@ -124,6 +142,7 @@ const register = Effect.fnUntraced(function* ( found.password === info.password ) return + if (found?.startedAt !== undefined && found.startedAt >= info.startedAt) return yield* publish }) yield* Effect.addFinalizer(() => @@ -139,6 +158,25 @@ const register = Effect.fnUntraced(function* ( ) }) +const recognizeIncumbent = Effect.fnUntraced(function* (options: DiscoverOptions, hostname: string, port: number) { + const found = yield* Service.incumbent({ ...options, url: serviceURL(hostname, port) }).pipe( + Effect.filterOrFail((value) => value !== undefined), + Effect.retry(Schedule.max([Schedule.spaced("100 millis"), Schedule.recurs(60)])), + Effect.option, + ) + return Option.isSome(found) +}) + +function serviceURL(hostname: string, port: number) { + return `http://${hostname.includes(":") ? `[${hostname}]` : hostname}:${port}` +} + +function addressInUse(error: unknown): boolean { + if (typeof error !== "object" || error === null) return false + if ("code" in error && error.code === "EADDRINUSE") return true + return "cause" in error && addressInUse(error.cause) +} + function waitForStdinClose() { return Effect.callback((resume) => { const close = () => resume(Effect.void) diff --git a/packages/cli/src/services/service-config.ts b/packages/cli/src/services/service-config.ts index 52543fcbe4..58c7c76935 100644 --- a/packages/cli/src/services/service-config.ts +++ b/packages/cli/src/services/service-config.ts @@ -30,6 +30,12 @@ export function filename(channel = InstallationChannel) { return `service-${Hash.fast(channel)}.json` } +export function defaultPort(channel = InstallationChannel) { + if (channel === "latest") return 0xc0de + if (channel === "local") return 0xc0df + return 10_000 + (Number.parseInt(Hash.fast(channel).slice(0, 8), 16) % 50_000) +} + export function versionBelongsToChannel( version: string | undefined, channel = InstallationChannel, diff --git a/packages/cli/test/service.test.ts b/packages/cli/test/service.test.ts index 3f6225b383..b4746e16ca 100644 --- a/packages/cli/test/service.test.ts +++ b/packages/cli/test/service.test.ts @@ -1,5 +1,5 @@ import { NodeFileSystem } from "@effect/platform-node" -import { Service } from "@opencode-ai/client/effect/service" +import { Service, type Info } from "@opencode-ai/client/effect/service" import { Database } from "@opencode-ai/core/database/database" import { EventV2 } from "@opencode-ai/core/event" import { EventTable } from "@opencode-ai/core/event/sql" @@ -17,6 +17,13 @@ import os from "node:os" import path from "node:path" import { ServiceConfig } from "../src/services/service-config" +test("managed service ports are stable per installation channel", () => { + expect(ServiceConfig.defaultPort("latest")).toBe(0xc0de) + expect(ServiceConfig.defaultPort("local")).toBe(0xc0df) + expect(ServiceConfig.defaultPort("preview-a")).toBe(ServiceConfig.defaultPort("preview-a")) + expect(ServiceConfig.defaultPort("preview-a")).not.toBe(ServiceConfig.defaultPort("preview-b")) +}) + test("local channel stores service config with the local service filename", async () => { const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-")) try { @@ -130,6 +137,9 @@ test("concurrent service processes elect one server", async () => { ) const command = [process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"] const registration = path.join(root, "state", "opencode", "service-local.json") + const port = await availablePort() + await fs.mkdir(path.join(root, "config", "opencode"), { recursive: true }) + await fs.writeFile(path.join(root, "config", "opencode", "service-local.json"), JSON.stringify({ port })) const processes = Array.from({ length: 10 }, () => Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" })) try { @@ -137,11 +147,13 @@ test("concurrent service processes elect one server", async () => { const winner = processes.find((process) => process.pid === info.pid) const losers = processes.filter((process) => process.pid !== info.pid) const exited = await Promise.all( - losers.map((process) => Promise.race([process.exited.then(() => true), Bun.sleep(10_000).then(() => false)])), + losers.map((process) => Promise.race([process.exited.then(() => true), Bun.sleep(60_000).then(() => false)])), ) expect(exited).toEqual(losers.map(() => true)) expect(winner?.exitCode).toBe(null) + expect(new URL(info.url).port).toBe(String(port)) + expect(await Bun.file(registration + ".lock").exists()).toBe(false) expect( await fetch(new URL("/api/health", info.url), { headers: { authorization: "Basic " + btoa(`opencode:${info.password}`) }, @@ -154,9 +166,14 @@ test("concurrent service processes elect one server", async () => { const blockedTemp = registration + "." + info.id + ".tmp" await fs.mkdir(blockedTemp) await fs.rm(registration) - await Bun.sleep(6_000) + 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) @@ -164,6 +181,16 @@ test("concurrent service processes elect one server", async () => { 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 { @@ -202,9 +229,88 @@ test("concurrent service processes elect one server", async () => { await fs.rm(root, { recursive: true, force: true }) } } -}, 60_000) +}, 120_000) -test("a failed service stays registered and owns the lock until stopped", async () => { +test("configured managed service port overrides the channel default", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-port-")) + const port = await availablePort() + const env = serviceEnv(root) + const registration = path.join(root, "state", "opencode", "service-local.json") + await fs.mkdir(path.join(root, "config", "opencode"), { recursive: true }) + 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, + stderr: "pipe", + stdout: "ignore", + }) + try { + const info = await waitForInfo(registration) + expect(new URL(info.url).port).toBe(String(port)) + await Effect.runPromise(Service.stop({ file: registration }).pipe(Effect.provide(NodeFileSystem.layer))) + await owner.exited + } finally { + owner.kill("SIGTERM") + await owner.exited + await fs.rm(root, { recursive: true, force: true }) + } +}, 30_000) + +test("unrelated managed port occupancy reports an actionable conflict", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-conflict-")) + const listener = Bun.serve({ port: 0, fetch: () => new Response("unrelated") }) + const port = listener.port + const registration = path.join(root, "state", "opencode", "service-local.json") + await fs.mkdir(path.join(root, "config", "opencode"), { recursive: true }) + await fs.writeFile(path.join(root, "config", "opencode", "service-local.json"), JSON.stringify({ port })) + const contender = Bun.spawn([process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"], { + env: serviceEnv(root), + stderr: "pipe", + stdout: "pipe", + }) + try { + expect(await contender.exited).not.toBe(0) + const output = (await new Response(contender.stdout).text()) + (await new Response(contender.stderr).text()) + expect(output).toContain(`Managed service port ${port} on 127.0.0.1 is already in use by another process`) + expect(output).toContain("opencode service set port ") + expect(await Bun.file(registration).exists()).toBe(false) + } finally { + listener.stop(true) + contender.kill("SIGTERM") + await contender.exited + await fs.rm(root, { recursive: true, force: true }) + } +}, 30_000) + +test("stale dead registration is replaced after binding the selected port", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-stale-")) + const port = await availablePort() + const registration = path.join(root, "state", "opencode", "service-local.json") + await fs.mkdir(path.join(root, "config", "opencode"), { recursive: true }) + await fs.mkdir(path.dirname(registration), { recursive: true }) + await fs.writeFile(path.join(root, "config", "opencode", "service-local.json"), JSON.stringify({ port })) + await fs.writeFile( + registration, + JSON.stringify({ id: "dead", version: "dead", url: `http://127.0.0.1:${port}`, pid: 2_147_483_647 }), + ) + const owner = Bun.spawn([process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"], { + env: serviceEnv(root), + stderr: "pipe", + stdout: "ignore", + }) + try { + const info = await waitForInfo(registration, (value) => value.id !== "dead") + expect(new URL(info.url).port).toBe(String(port)) + expect(info.pid).toBe(owner.pid) + await Effect.runPromise(Service.stop({ file: registration }).pipe(Effect.provide(NodeFileSystem.layer))) + await owner.exited + } finally { + owner.kill("SIGTERM") + await owner.exited + await fs.rm(root, { recursive: true, force: true }) + } +}, 30_000) + +test("a failed service stays registered and owns the selected port until stopped", async () => { const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-failed-")) const database = path.join(root, "database") await fs.mkdir(database) @@ -275,13 +381,36 @@ function waitForExecutionStart(file: string, sessionID: SessionV2.ID) { ) } -async function waitForInfo(file: string) { +async function waitForInfo(file: string, accept: (info: Info) => boolean = () => true) { for (let attempt = 0; attempt < 400; attempt++) { const value = await Bun.file(file) .json() .catch(() => undefined) - if (value !== undefined) return Schema.decodeUnknownPromise(Service.Info)(value) + if (value !== undefined) { + const info = await Schema.decodeUnknownPromise(Service.Info)(value) + if (accept(info)) return info + } await Bun.sleep(50) } throw new Error("Timed out waiting for service registration") } + +async function availablePort() { + const server = Bun.serve({ port: 0, fetch: () => new Response() }) + const port = server.port + await server.stop(true) + return port +} + +function serviceEnv(root: string) { + return { + ...process.env, + HOME: root, + OPENCODE_DB: path.join(root, "opencode.db"), + OPENCODE_TEST_HOME: root, + XDG_CACHE_HOME: path.join(root, "cache"), + XDG_CONFIG_HOME: path.join(root, "config"), + XDG_DATA_HOME: path.join(root, "data"), + XDG_STATE_HOME: path.join(root, "state"), + } +} diff --git a/packages/client/src/effect/service.ts b/packages/client/src/effect/service.ts index ffd36bff77..7255bbf74d 100644 --- a/packages/client/src/effect/service.ts +++ b/packages/client/src/effect/service.ts @@ -29,6 +29,17 @@ export const discover = Effect.fn("service.discover")(function* (options: Discov return (yield* discoverLocal(options))?.endpoint }) +/** Recognize an authenticated compatible service bound to an expected URL, including while it starts or fails. */ +export const incumbent = Effect.fn("service.incumbent")(function* ( + options: DiscoverOptions & { readonly url: string }, +) { + const info = yield* read(options.file) + const found = info === undefined ? undefined : yield* probe({ ...info, url: options.url }) + if (found === undefined || found.legacy) return undefined + if (options.version !== undefined && found.version !== options.version) return undefined + return { endpoint: found.endpoint, state: found.state } +}) + const discoverLocal = Effect.fnUntraced(function* (options: DiscoverOptions) { const found = (yield* registered(options.file)).service if (found?.state !== "ready") return undefined @@ -101,8 +112,15 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti lastSpawn = Date.now() } return Option.none() - }).pipe(Effect.repeat({ until: Option.isSome, schedule: Schedule.spaced("1 second") })) - return Option.getOrThrow(found).endpoint + }).pipe( + Effect.repeat({ + until: Option.isSome, + schedule: Schedule.max([Schedule.spaced("1 second"), Schedule.recurs(120)]), + }), + ) + if (Option.isNone(found)) + return yield* Effect.fail(new Error("Timed out waiting for the background service to start")) + return found.value.endpoint }) function contenderFailure(contender: Contender) { @@ -143,6 +161,7 @@ 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)) @@ -273,4 +292,4 @@ const requestStop = Effect.fnUntraced(function* (service: LocalService) { }) /** Effect-based local service lifecycle operations. */ -export const Service = { discover, ensure, stop, headers, Info } +export const Service = { discover, incumbent, ensure, stop, headers, Info } diff --git a/packages/client/src/promise/service.ts b/packages/client/src/promise/service.ts index aecaf95f4f..6d8aa5e327 100644 --- a/packages/client/src/promise/service.ts +++ b/packages/client/src/promise/service.ts @@ -2,13 +2,7 @@ import { readFile } from "node:fs/promises" import { spawn, type ChildProcess } from "node:child_process" import { homedir } from "node:os" import { join } from "node:path" -import type { - DiscoverOptions, - Endpoint, - Info, - EnsureOptions, - StopOptions, -} from "../service.js" +import type { DiscoverOptions, Endpoint, Info, EnsureOptions, StopOptions } from "../service.js" import type { ServiceHealth, ServiceStopResponse } from "./generated/types.js" export * from "../service.js" @@ -38,6 +32,7 @@ async function discoverLocal(options: DiscoverOptions) { /** Ensure a healthy, compatible local service is running. */ export async function ensure(options: EnsureOptions = {}): Promise { + const deadline = Date.now() + 120_000 const contenders = new Set() let announced = false let lastSpawn = 0 @@ -66,6 +61,7 @@ export async function ensure(options: EnsureOptions = {}): Promise { } 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) if (registration.service !== undefined) { @@ -128,7 +124,9 @@ function fallback() { /** Create HTTP authentication headers for a service endpoint. */ export function headers(endpoint: Endpoint) { if (endpoint.auth === undefined) return undefined - return { authorization: "Basic " + Buffer.from(endpoint.auth.username + ":" + endpoint.auth.password).toString("base64") } + return { + authorization: "Basic " + Buffer.from(endpoint.auth.username + ":" + endpoint.auth.password).toString("base64"), + } } async function read(file?: string) { @@ -227,7 +225,8 @@ async function kill(service: LocalService, options: { readonly file?: string }) 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))) throw new Error(`Server process ${service.info.pid} is still running`) + if (!(await waitUntilStopped(service.info.pid))) + throw new Error(`Server process ${service.info.pid} is still running`) } async function requestStop(service: LocalService) { diff --git a/packages/client/src/service.ts b/packages/client/src/service.ts index 96d54871f6..f4df10bf9c 100644 --- a/packages/client/src/service.ts +++ b/packages/client/src/service.ts @@ -50,4 +50,6 @@ 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 }