fix(cli): elect one managed daemon

Author: kitlangton

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(() =>

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<boolean>) {
+  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<typeof spawn>) {
+  if (process.exitCode !== null || process.signalCode !== null) return
+  const closed = new Promise<void>((resolve) => process.once("close", () => resolve()))
+  process.kill("SIGTERM")
+  await closed
+}

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<void, LockError, Scope.Scope>
+    readonly tryAcquire: (key: string, dir?: string) => Effect.Effect<boolean, 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>
@@ -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,26 +265,44 @@ 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 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 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
-        yield* fs
-          .utimes(handle.heartbeatPath, new Date(), new Date())
-          .pipe(Effect.ignore, Effect.repeat(Schedule.spaced(HEARTBEAT_MS)), Effect.forkScoped)
+      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(
@@ -276,7 +316,7 @@ export namespace EffectFlock {
           ),
       )
 
-      return Service.of({ acquire, withLock })
+      return Service.of({ acquire, tryAcquire, withLock })
     }),
   )
 

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)