From dd842f9862682ed21f3c0c3f380f46ef8120d481 Mon Sep 17 00:00:00 2001 From: Dax Raad Date: Wed, 8 Jul 2026 11:45:38 -0400 Subject: [PATCH] fix(cli): elect one service process --- packages/cli/src/server-process.ts | 30 +++++++- packages/cli/test/service.test.ts | 47 ++++++++++++- packages/core/src/util/effect-flock.ts | 73 +++++++++++++------- packages/core/test/util/effect-flock.test.ts | 23 ++++++ 4 files changed, 145 insertions(+), 28 deletions(-) diff --git a/packages/cli/src/server-process.ts b/packages/cli/src/server-process.ts index 2d5c2f8bfa..39c1e256d3 100644 --- a/packages/cli/src/server-process.ts +++ b/packages/cli/src/server-process.ts @@ -7,10 +7,11 @@ 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" -import { Effect, FileSystem, Logger, Option, Redacted, Schedule, Schema } from "effect" +import { Effect, Exit, FileSystem, Logger, Option, Redacted, Schedule, Schema, Scope } from "effect" import { HttpServer } from "effect/unstable/http" import { Env } from "./env" import { ServiceConfig } from "./services/service-config" @@ -27,7 +28,7 @@ export type Options = { export const run = Effect.fn("cli.server-process.run")((options: Options) => processEffect(options).pipe( 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), ), ) @@ -36,6 +37,16 @@ const processEffect = Effect.fnUntraced(function* (options: Options) { if (options.mode === "service") yield* Effect.sync(() => process.chdir(Global.Path.home)) return yield* Effect.scoped( Effect.gen(function* () { + const serviceOptions = options.mode === "service" ? yield* ServiceConfig.options() : undefined + const lockScope = serviceOptions === undefined ? undefined : yield* acquireServiceLock(serviceOptions.file) + if ( + serviceOptions !== undefined && + lockScope !== undefined && + (yield* Service.discover(serviceOptions)) !== undefined + ) { + yield* Scope.close(lockScope, Exit.void) + return + } const environmentPassword = yield* Env.password // Keep the lease credential out of the environment inherited by tools. if (options.mode === "stdio") { @@ -55,7 +66,10 @@ 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 (lockScope !== undefined) { + yield* register(address, password) + yield* Scope.close(lockScope, Exit.void) + } 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}`) @@ -66,6 +80,16 @@ const processEffect = Effect.fnUntraced(function* (options: Options) { ) }) +const acquireServiceLock = Effect.fnUntraced(function* (file: string) { + const flock = yield* EffectFlock.Service + const scope = yield* Scope.make() + yield* Effect.addFinalizer((exit) => Scope.close(scope, exit)) + yield* flock + .acquire(`service:${file}`, undefined, { staleMs: 3_000, timeoutMs: 3_000 }) + .pipe(Effect.provideService(Scope.Scope, scope)) + return scope +}) + // The latest atomic registration wins. A displaced process notices the new id, // exits, and cannot remove its successor's registration from its finalizer. const infoJson = Schema.fromJsonString(Service.Info) diff --git a/packages/cli/test/service.test.ts b/packages/cli/test/service.test.ts index 68073851c1..ac1f9ac1cf 100644 --- a/packages/cli/test/service.test.ts +++ b/packages/cli/test/service.test.ts @@ -1,7 +1,8 @@ import { NodeFileSystem } from "@effect/platform-node" +import { Service } from "@opencode-ai/client/effect" import { Global } from "@opencode-ai/core/global" import { expect, test } from "bun:test" -import { Effect } from "effect" +import { Effect, Schema } from "effect" import fs from "node:fs/promises" import os from "node:os" import path from "node:path" @@ -24,3 +25,47 @@ test("local channel stores service config with the local service filename", asyn await fs.rm(root, { recursive: true, force: true }) } }) + +test("concurrent service processes elect one server", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-election-")) + const env = { + ...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"), + } + const command = [process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"] + const first = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" }) + const second = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" }) + + try { + const registration = path.join(root, "state", "opencode", "service-local.json") + const info = await waitForInfo(registration) + const winner = info.pid === first.pid ? first : second + const loser = info.pid === first.pid ? second : first + const exited = await Promise.race([loser.exited.then(() => true), Bun.sleep(10_000).then(() => false)]) + + expect(exited).toBe(true) + expect(winner.exitCode).toBe(null) + } finally { + first.kill("SIGTERM") + second.kill("SIGTERM") + await Promise.all([first.exited, second.exited]) + await fs.rm(root, { recursive: true, force: true }) + } +}) + +async function waitForInfo(file: string) { + for (let attempt = 0; attempt < 200; attempt++) { + const value = await Bun.file(file) + .json() + .catch(() => undefined) + if (value !== undefined) return Schema.decodeUnknownPromise(Service.Info)(value) + await Bun.sleep(50) + } + throw new Error("Timed out waiting for service registration") +} diff --git a/packages/core/src/util/effect-flock.ts b/packages/core/src/util/effect-flock.ts index b85900118a..4b528d2602 100644 --- a/packages/core/src/util/effect-flock.ts +++ b/packages/core/src/util/effect-flock.ts @@ -36,20 +36,24 @@ export namespace EffectFlock { export type LockError = LockTimeoutError | LockCompromisedError + export interface Options { + readonly staleMs?: number + readonly timeoutMs?: number + } + // --------------------------------------------------------------------------- - // Timing (baked in — no caller ever overrides these) + // Timing defaults // --------------------------------------------------------------------------- - const STALE_MS = 60_000 - const TIMEOUT_MS = 5 * 60_000 + const DEFAULT_STALE_MS = 60_000 + const DEFAULT_TIMEOUT_MS = 5 * 60_000 const BASE_DELAY_MS = 100 const MAX_DELAY_MS = 2_000 - const HEARTBEAT_MS = Math.max(100, Math.floor(STALE_MS / 3)) - const retrySchedule = Schedule.exponential(BASE_DELAY_MS, 1.7).pipe( - Schedule.either(Schedule.spaced(MAX_DELAY_MS)), + const retrySchedule = (timeoutMs: number) => Schedule.exponential(BASE_DELAY_MS, 1.7).pipe( + Schedule.either(Schedule.spaced(Math.min(MAX_DELAY_MS, Math.max(BASE_DELAY_MS, Math.floor(timeoutMs / 10))))), Schedule.jittered, - Schedule.while((meta) => meta.elapsed < TIMEOUT_MS), + Schedule.while((meta) => meta.elapsed < timeoutMs), ) // --------------------------------------------------------------------------- @@ -73,7 +77,7 @@ export namespace EffectFlock { // --------------------------------------------------------------------------- export interface Interface { - readonly acquire: (key: string, dir?: string) => Effect.Effect + readonly acquire: (key: string, dir?: string, options?: Options) => Effect.Effect readonly withLock: { (key: string, dir?: string): (body: Effect.Effect) => Effect.Effect (body: Effect.Effect, key: string, dir?: string): Effect.Effect @@ -135,9 +139,9 @@ export namespace EffectFlock { ), ) - const cleanStaleBreaker = Effect.fnUntraced(function* (breakerPath: string) { + const cleanStaleBreaker = Effect.fnUntraced(function* (breakerPath: string, staleMs: number) { const bs = yield* safeStat(breakerPath) - if (bs && wall() - mtimeMs(bs) > STALE_MS) yield* forceRemove(breakerPath) + if (bs && wall() - mtimeMs(bs) > staleMs) yield* forceRemove(breakerPath) return false }) @@ -147,26 +151,31 @@ export namespace EffectFlock { ensuredDirs.add(dir) }) - const isStale = Effect.fnUntraced(function* (lockDir: string, heartbeatPath: string, metaPath: string) { + const isStale = Effect.fnUntraced(function* ( + lockDir: string, + heartbeatPath: string, + metaPath: string, + staleMs: number, + ) { const now = wall() const hb = yield* safeStat(heartbeatPath) - if (hb) return now - mtimeMs(hb) > STALE_MS + if (hb) return now - mtimeMs(hb) > staleMs const meta = yield* safeStat(metaPath) - if (meta) return now - mtimeMs(meta) > STALE_MS + if (meta) return now - mtimeMs(meta) > staleMs const dir = yield* safeStat(lockDir) if (!dir) return false - return now - mtimeMs(dir) > STALE_MS + return now - mtimeMs(dir) > staleMs }) // -- single lock attempt -- type Handle = { token: string; metaPath: string; heartbeatPath: string; lockDir: string } - const tryAcquireLockDir = (lockDir: string, key: string) => + const tryAcquireLockDir = (lockDir: string, key: string, staleMs: number) => Effect.gen(function* () { const token = randomUUID() const metaPath = path.join(lockDir, "meta.json") @@ -176,7 +185,7 @@ export namespace EffectFlock { const created = yield* atomicMkdir(lockDir) if (!created) { - if (!(yield* isStale(lockDir, heartbeatPath, metaPath))) return yield* new NotAcquired() + if (!(yield* isStale(lockDir, heartbeatPath, metaPath, staleMs))) return yield* new NotAcquired() // Stale — race for breaker ownership const breakerPath = lockDir + ".breaker" @@ -185,7 +194,7 @@ export namespace EffectFlock { Effect.as(true), Effect.catchIf( (e) => e.reason._tag === "AlreadyExists", - () => cleanStaleBreaker(breakerPath), + () => cleanStaleBreaker(breakerPath, staleMs), ), Effect.catchIf(isPathGone, () => Effect.succeed(false)), Effect.orDie, @@ -195,7 +204,7 @@ export namespace EffectFlock { // We own the breaker — double-check staleness, nuke, recreate const recreated = yield* Effect.gen(function* () { - if (!(yield* isStale(lockDir, heartbeatPath, metaPath))) return false + if (!(yield* isStale(lockDir, heartbeatPath, metaPath, staleMs))) return false yield* forceRemove(lockDir) return yield* atomicMkdir(lockDir) }).pipe(Effect.ensuring(forceRemove(breakerPath))) @@ -218,13 +227,21 @@ export namespace EffectFlock { // -- retry wrapper (preserves Handle type) -- - const acquireHandle = (lockfile: string, key: string): Effect.Effect => - tryAcquireLockDir(lockfile, key).pipe( + const acquireHandle = ( + lockfile: string, + key: string, + options: { staleMs: number; timeoutMs: number }, + ): Effect.Effect => + tryAcquireLockDir(lockfile, key, options.staleMs).pipe( Effect.retry({ while: (err) => err._tag === "NotAcquired", - schedule: retrySchedule, + schedule: retrySchedule(options.timeoutMs), }), Effect.catchTag("NotAcquired", () => Effect.fail(new LockTimeoutError({ key }))), + Effect.timeoutOrElse({ + duration: options.timeoutMs, + orElse: () => Effect.fail(new LockTimeoutError({ key })), + }), ) // -- release -- @@ -250,19 +267,27 @@ export namespace EffectFlock { // -- build service -- - const acquire = Effect.fn("EffectFlock.acquire")(function* (key: string, dir?: string) { + const acquire = Effect.fn("EffectFlock.acquire")(function* (key: string, dir?: string, options: Options = {}) { const lockDir = dir ?? lockRoot + const staleMs = options.staleMs ?? DEFAULT_STALE_MS + const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS 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)) + const handle = yield* Effect.acquireRelease(acquireHandle(lockfile, key, { staleMs, timeoutMs }), (handle) => + release(handle), + ) // Heartbeat fiber — scoped, so it's interrupted before release runs yield* fs .utimes(handle.heartbeatPath, new Date(), new Date()) - .pipe(Effect.ignore, Effect.repeat(Schedule.spaced(HEARTBEAT_MS)), Effect.forkScoped) + .pipe( + Effect.ignore, + Effect.repeat(Schedule.spaced(Math.max(100, Math.floor(staleMs / 3)))), + Effect.forkScoped, + ) }) const withLock: Interface["withLock"] = Function.dual( diff --git a/packages/core/test/util/effect-flock.test.ts b/packages/core/test/util/effect-flock.test.ts index a0f737a998..ca2820737d 100644 --- a/packages/core/test/util/effect-flock.test.ts +++ b/packages/core/test/util/effect-flock.test.ts @@ -134,6 +134,29 @@ describe("util.effect-flock", () => { }), ) + it.live( + "supports an acquisition timeout", + 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") + const key = "eflock:timeout" + + yield* Effect.scoped( + Effect.gen(function* () { + yield* flock.acquire(key, dir) + const started = performance.now() + const error = yield* Effect.scoped( + flock.acquire(key, dir, { staleMs: 10_000, timeoutMs: 300 }), + ).pipe(Effect.flip) + expect(error._tag).toBe("LockTimeoutError") + expect(performance.now() - started).toBeLessThan(1_000) + }), + ) + yield* Effect.promise(() => fs.rm(tmp, { recursive: true, force: true })) + }), + ) + it.live( "withLock data-first", Effect.gen(function* () {