From 68c62774ac1bd1971abaa960a0fb7d2ca6f3958d Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Tue, 7 Jul 2026 21:21:28 -0400 Subject: [PATCH] fix(cli): elect one managed daemon --- packages/cli/src/server-process.ts | 28 +++++++-- packages/cli/test/server-process.test.ts | 61 +++++++++++++++++++ packages/core/src/util/effect-flock.ts | 64 ++++++++++++++++---- packages/core/test/util/effect-flock.test.ts | 28 ++++++--- 4 files changed, 155 insertions(+), 26 deletions(-) create mode 100644 packages/cli/test/server-process.test.ts diff --git a/packages/cli/src/server-process.ts b/packages/cli/src/server-process.ts index 2d5c2f8bfa..c0ab18d2d2 100644 --- a/packages/cli/src/server-process.ts +++ b/packages/cli/src/server-process.ts @@ -7,6 +7,7 @@ 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 { EffectFlock } from "@opencode-ai/core/util/effect-flock" import { start } from "@opencode-ai/server/process" import { randomBytes, randomUUID } from "node:crypto" import path from "node:path" @@ -24,10 +25,13 @@ export type Options = { readonly port?: number } +type ManagedServiceOptions = Service.Options & { readonly file: string } + export const run = Effect.fn("cli.server-process.run")((options: Options) => processEffect(options).pipe( + Effect.catchTag("ServiceAlreadyOwned", () => Effect.void), Effect.provide(Updater.layer), - Effect.provide(AppNodeBuilder.build(LayerNode.group([Global.node, AppProcess.node]))), + Effect.provide(AppNodeBuilder.build(LayerNode.group([Global.node, AppProcess.node, EffectFlock.node]))), Effect.provide(NodeServices.layer), ), ) @@ -43,6 +47,17 @@ const processEffect = Effect.fnUntraced(function* (options: Options) { delete process.env.OPENCODE_SERVER_PASSWORD } const config = options.mode === "service" ? yield* ServiceConfig.read() : {} + const serviceOptions = options.mode === "service" ? yield* ServiceConfig.options() : undefined + if (serviceOptions) { + const flock = yield* EffectFlock.Service + yield* flock.tryAcquire(`opencode-service:${serviceOptions.file}`).pipe( + Effect.filterOrFail( + (acquired) => acquired, + () => ({ _tag: "ServiceAlreadyOwned" }) as const, + ), + Effect.retry(Schedule.spaced("100 millis").pipe(Schedule.both(Schedule.recurs(20)))), + ) + } const password = options.mode === "service" ? yield* ServiceConfig.password() @@ -55,7 +70,7 @@ const processEffect = Effect.fnUntraced(function* (options: Options) { port: Option.fromNullishOr(options.port ?? config.port), password, }).pipe(Effect.provide(Logger.layer([], { mergeWithExisting: false }))) - if (options.mode === "service") yield* register(address, password) + if (serviceOptions) yield* register(address, password, serviceOptions) const url = HttpServer.formatAddress(address) console.log(options.mode === "stdio" ? JSON.stringify({ url }) : `server listening on ${url}`) if (options.mode === "default" && !environmentPassword) console.log(`server password ${password}`) @@ -72,9 +87,12 @@ 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) { +const register = Effect.fnUntraced(function* ( + address: HttpServer.Address, + password: string, + options: ManagedServiceOptions, +) { const fs = yield* FileSystem.FileSystem - const options = yield* ServiceConfig.options() const id = randomUUID() const temp = options.file + "." + id + ".tmp" yield* fs.makeDirectory(path.dirname(options.file), { recursive: true }) @@ -98,7 +116,7 @@ const register = Effect.fnUntraced(function* (address: HttpServer.Address, passw ? Effect.void : Effect.try({ try: () => process.kill(process.pid, "SIGTERM"), catch: (cause) => cause }).pipe(Effect.ignore), ), - Effect.repeat(Schedule.spaced("10 seconds")), + Effect.repeat(Schedule.spaced("1 second")), Effect.forkScoped, ) yield* Effect.addFinalizer(() => diff --git a/packages/cli/test/server-process.test.ts b/packages/cli/test/server-process.test.ts new file mode 100644 index 0000000000..88b1913dbe --- /dev/null +++ b/packages/cli/test/server-process.test.ts @@ -0,0 +1,61 @@ +import { expect, test } from "bun:test" +import { spawn } from "node:child_process" +import fs from "node:fs/promises" +import os from "node:os" +import path from "node:path" + +test("concurrent service candidates elect one owner before serving", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-election-")) + const env = { + ...process.env, + 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"), + OPENCODE_DISABLE_AUTOUPDATE: "1", + } + const entry = path.join(import.meta.dir, "../src/index.ts") + const candidates = [ + spawn(process.execPath, [entry, "serve", "--service"], { cwd: path.join(import.meta.dir, ".."), env }), + spawn(process.execPath, [entry, "serve", "--service"], { cwd: path.join(import.meta.dir, ".."), env }), + ] + + try { + const registration = path.join(root, "state", "opencode", "service-local.json") + await waitForFile(registration) + const info = await Bun.file(registration).json() + await waitFor(() => candidates.some((candidate) => candidate.exitCode !== null)) + + expect(candidates.filter((candidate) => candidate.exitCode === null)).toHaveLength(1) + expect(candidates.find((candidate) => candidate.exitCode !== null)?.exitCode).toBe(0) + expect(candidates.find((candidate) => candidate.exitCode === null)?.pid).toBe(info.pid) + expect( + await fetch(new URL("/api/health", info.url), { + headers: { authorization: "Basic " + btoa(`opencode:${info.password}`) }, + }).then((response) => response.ok), + ).toBe(true) + } finally { + await Promise.all(candidates.map(stop)) + await fs.rm(root, { recursive: true, force: true }) + } +}, 20_000) + +async function waitForFile(file: string) { + await waitFor(() => Bun.file(file).exists()) +} + +async function waitFor(check: () => boolean | Promise) { + const timeout = Date.now() + 10_000 + while (Date.now() < timeout) { + if (await check()) return + await Bun.sleep(20) + } + throw new Error("Timed out waiting for service election") +} + +async function stop(process: ReturnType) { + if (process.exitCode !== null || process.signalCode !== null) return + const closed = new Promise((resolve) => process.once("close", () => resolve())) + process.kill("SIGTERM") + await closed +} diff --git a/packages/core/src/util/effect-flock.ts b/packages/core/src/util/effect-flock.ts index b85900118a..a3f70719b9 100644 --- a/packages/core/src/util/effect-flock.ts +++ b/packages/core/src/util/effect-flock.ts @@ -66,6 +66,7 @@ export namespace EffectFlock { ) const decodeMeta = Schema.decodeUnknownSync(LockMetaJson) + const decodeMetaOption = Schema.decodeUnknownOption(LockMetaJson) const encodeMeta = Schema.encodeSync(LockMetaJson) // --------------------------------------------------------------------------- @@ -74,6 +75,7 @@ export namespace EffectFlock { export interface Interface { readonly acquire: (key: string, dir?: string) => Effect.Effect + readonly tryAcquire: (key: string, dir?: string) => Effect.Effect readonly withLock: { (key: string, dir?: string): (body: Effect.Effect) => Effect.Effect (body: Effect.Effect, key: string, dir?: string): Effect.Effect @@ -150,6 +152,26 @@ export namespace EffectFlock { const isStale = Effect.fnUntraced(function* (lockDir: string, heartbeatPath: string, metaPath: string) { const now = wall() + const raw = yield* fs.readFileString(metaPath).pipe( + Effect.map(Option.some), + Effect.catchIf(isPathGone, () => Effect.succeed(Option.none())), + Effect.orDie, + ) + const owner = Option.isSome(raw) ? Option.getOrUndefined(decodeMetaOption(raw.value)) : undefined + if (owner?.hostname === hostname) { + const alive = yield* Effect.try({ + try: () => { + process.kill(owner.pid, 0) + return true + }, + catch: (cause) => { + const code = cause && typeof cause === "object" && "code" in cause ? cause.code : undefined + return code !== "ESRCH" + }, + }).pipe(Effect.orElseSucceed(() => false)) + if (!alive) return true + } + const hb = yield* safeStat(heartbeatPath) if (hb) return now - mtimeMs(hb) > STALE_MS @@ -243,28 +265,46 @@ export namespace EffectFlock { catch: (cause) => new ReleaseError({ detail: "metadata invalid", cause }), }).pipe(Effect.orDie) - if (parsed.token !== handle.token) return yield* Effect.die(new ReleaseError({ detail: "token mismatch" })) + if (parsed.token !== handle.token) yield* Effect.die(new ReleaseError({ detail: "token mismatch" })) yield* forceRemove(handle.lockDir) }) // -- build service -- - const acquire = Effect.fn("EffectFlock.acquire")(function* (key: string, dir?: string) { - const lockDir = dir ?? lockRoot - yield* ensureDir(lockDir) - - const lockfile = path.join(lockDir, Hash.fast(key) + ".lock") - - // acquireRelease: acquire is uninterruptible, release is guaranteed - const handle = yield* Effect.acquireRelease(acquireHandle(lockfile, key), (handle) => release(handle)) - - // Heartbeat fiber — scoped, so it's interrupted before release runs + const heartbeat = Effect.fnUntraced(function* (handle: Handle) { yield* fs .utimes(handle.heartbeatPath, new Date(), new Date()) .pipe(Effect.ignore, Effect.repeat(Schedule.spaced(HEARTBEAT_MS)), Effect.forkScoped) }) + const acquire = Effect.fn("EffectFlock.acquire")(function* (key: string, dir?: string) { + const lockDir = dir ?? lockRoot + yield* ensureDir(lockDir) + const handle = yield* Effect.acquireRelease( + acquireHandle(path.join(lockDir, Hash.fast(key) + ".lock"), key), + (handle) => release(handle), + ) + yield* heartbeat(handle) + }) + + const tryAcquire = Effect.fn("EffectFlock.tryAcquire")(function* (key: string, dir?: string) { + return yield* Effect.uninterruptibleMask(() => + Effect.gen(function* () { + const lockDir = dir ?? lockRoot + yield* ensureDir(lockDir) + const handle = yield* tryAcquireLockDir(path.join(lockDir, Hash.fast(key) + ".lock"), key).pipe( + Effect.map(Option.some), + Effect.catchTag("NotAcquired", () => Effect.succeed(Option.none())), + ) + if (Option.isNone(handle)) return false + yield* Effect.addFinalizer(() => release(handle.value)) + yield* heartbeat(handle.value) + return true + }), + ) + }) + const withLock: Interface["withLock"] = Function.dual( (args) => Effect.isEffect(args[0]), (body: Effect.Effect, key: string, dir?: string): Effect.Effect => @@ -276,7 +316,7 @@ export namespace EffectFlock { ), ) - return Service.of({ acquire, withLock }) + return Service.of({ acquire, tryAcquire, withLock }) }), ) diff --git a/packages/core/test/util/effect-flock.test.ts b/packages/core/test/util/effect-flock.test.ts index a0f737a998..863b607e5d 100644 --- a/packages/core/test/util/effect-flock.test.ts +++ b/packages/core/test/util/effect-flock.test.ts @@ -6,7 +6,6 @@ import os from "os" import { Cause, Effect, Exit } from "effect" import { testEffect } from "../lib/effect" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" -import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { EffectFlock } from "@opencode-ai/core/util/effect-flock" import { Global } from "@opencode-ai/core/global" import { Hash } from "@opencode-ai/core/util/hash" @@ -134,6 +133,24 @@ describe("util.effect-flock", () => { }), ) + it.live( + "tries once without waiting for an active owner", + Effect.gen(function* () { + const flock = yield* EffectFlock.Service + const tmp = yield* Effect.promise(() => fs.mkdtemp(path.join(os.tmpdir(), "eflock-test-"))) + const dir = path.join(tmp, "locks") + + yield* Effect.scoped( + Effect.gen(function* () { + expect(yield* flock.tryAcquire("eflock:try", dir)).toBe(true) + expect(yield* Effect.scoped(flock.tryAcquire("eflock:try", dir))).toBe(false) + }), + ) + expect(yield* Effect.scoped(flock.tryAcquire("eflock:try", dir))).toBe(true) + yield* Effect.promise(() => fs.rm(tmp, { recursive: true, force: true })) + }), + ) + it.live( "withLock data-first", Effect.gen(function* () { @@ -358,7 +375,7 @@ describe("util.effect-flock", () => { ) it.live( - "recovers after a crashed lock owner", + "immediately recovers after a local lock owner crashes", () => Effect.promise(async () => { const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "eflock-crash-")) @@ -371,13 +388,6 @@ describe("util.effect-flock", () => { await waitForFile(ready, 5_000) await stopWorker(proc) - // Backdate lock files so they're past STALE_MS (60s) - const lockDir = lock(dir, "eflock:crash") - const old = new Date(Date.now() - 120_000) - await fs.utimes(lockDir, old, old).catch(() => {}) - await fs.utimes(path.join(lockDir, "heartbeat"), old, old).catch(() => {}) - await fs.utimes(path.join(lockDir, "meta.json"), old, old).catch(() => {}) - const done = path.join(tmp, "done.log") const result = await run({ key: "eflock:crash", dir, done, holdMs: 10 }) expect(result.code).toBe(0)