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

fix(cli): elect one service process

Browse source
This commit is contained in:
Dax Raad 2026-07-08 11:45:38 -04:00
commit dd842f9862
4 changed files with 145 additions and 28 deletions
Download patch file Download diff file Expand all files Collapse all files

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

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

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

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

73
packages/core/src/util/effect-flock.ts
View file

@ -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<void, LockError, Scope.Scope>
readonly acquire: (key: string, dir?: string, options?: Options) => Effect.Effect<void, LockError, Scope.Scope>
readonly withLock: {
(key: string, dir?: string): <A, E, R>(body: Effect.Effect<A, E, R>) => Effect.Effect<A, E | LockError, R>
<A, E, R>(body: Effect.Effect<A, E, R>, key: string, dir?: string): Effect.Effect<A, E | LockError, R>
@ -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<Handle, LockError> =>
tryAcquireLockDir(lockfile, key).pipe(
const acquireHandle = (
lockfile: string,
key: string,
options: { staleMs: number; timeoutMs: number },
): Effect.Effect<Handle, LockError> =>
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(

23
packages/core/test/util/effect-flock.test.ts
View file

@ -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* () {