ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(cli): elect managed service by port bind (#37572)

Browse source 
Co-authored-by: Dax Raad <thdxr@users.noreply.github.com>
This commit is contained in:
opencode-agent[bot] 2026-07-17 23:12:36 +00:00 • committed by GitHub
commit 6ea8247e0f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 232 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

78
packages/cli/src/server-process.ts
View file

@ -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 <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<void>((resume) => {
const close = () => resume(Effect.void)

6
packages/cli/src/services/service-config.ts
View file

@ -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,

143
packages/cli/test/service.test.ts
View file

@ -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 <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"),
}
}

25
packages/client/src/effect/service.ts
View file

@ -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<LocalService>()
}).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 }

17
packages/client/src/promise/service.ts
View file

@ -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<Endpoint> {
const deadline = Date.now() + 120_000
const contenders = new Set<Contender>()
let announced = false
let lastSpawn = 0
@ -66,6 +61,7 @@ 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)
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) {

2
packages/client/src/service.ts
View file

@ -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
}