mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-17 04:03:26 +00:00
fix(client): surface managed startup stderr (#41793)
This commit is contained in:
parent
d6ed520c25
commit
9322f5d2c9
6 changed files with 161 additions and 97 deletions
|
|
@ -1,9 +1,14 @@
|
|||
import { ServiceStatus } from "@opencode-ai/protocol/groups/health"
|
||||
import { Effect, FileSystem, Option, Schedule, Schema } from "effect"
|
||||
import { spawn, type ChildProcess } from "node:child_process"
|
||||
import { homedir } from "node:os"
|
||||
import { join } from "node:path"
|
||||
import type { DiscoverOptions, Endpoint, EnsureOptions, StopOptions } from "../service.js"
|
||||
import {
|
||||
contenderFailure,
|
||||
contenderFinished,
|
||||
type ServiceContender,
|
||||
spawnServiceContender,
|
||||
} from "../service-contender.js"
|
||||
import { defaultEnsureTiming, ensureTiming, type EnsureTiming } from "../service-timing.js"
|
||||
|
||||
export * from "../service.js"
|
||||
|
|
@ -18,11 +23,6 @@ export type Info = import("../service.js").Info
|
|||
// is all a client needs to connect. The daemon's own configuration (port,
|
||||
// persisted password) is CLI-owned and never read here.
|
||||
|
||||
type Contender = {
|
||||
readonly child: ChildProcess
|
||||
readonly error: () => Error | undefined
|
||||
}
|
||||
|
||||
// Read-only lookup: registration file plus health check and version gate.
|
||||
// Never spawns; escalation to ensure() is the caller's policy.
|
||||
/** Discover a healthy, compatible local service without starting one. */
|
||||
|
|
@ -54,7 +54,7 @@ const discoverLocal = Effect.fnUntraced(function* (options: DiscoverOptions) {
|
|||
/** Ensure a healthy, compatible local service is running. */
|
||||
export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOptions = {}) {
|
||||
const timing = ensureTiming(options)
|
||||
const contenders = new Set<Contender>()
|
||||
const contenders = new Set<ServiceContender>()
|
||||
let timeouts: { readonly info: Info; readonly count: number } | undefined
|
||||
let announced = false
|
||||
let lastSpawn = 0
|
||||
|
|
@ -70,13 +70,7 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
|
|||
if (command === undefined) return yield* Effect.fail(new Error("Missing service command"))
|
||||
return yield* Effect.try({
|
||||
try: () => {
|
||||
const child = spawn(command, args, { detached: true, stdio: "ignore" })
|
||||
let error: Error | undefined
|
||||
child.once("error", (cause) => {
|
||||
error = new Error("Failed to start server", { cause })
|
||||
})
|
||||
child.unref()
|
||||
return { child, error: () => error }
|
||||
return spawnServiceContender(command, args)
|
||||
},
|
||||
catch: (cause) => new Error("Failed to start server", { cause }),
|
||||
})
|
||||
|
|
@ -129,26 +123,13 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
|
|||
until: Option.isSome,
|
||||
schedule: Schedule.max([Schedule.spaced(timing.pollInterval), Schedule.recurs(timing.attempts)]),
|
||||
}),
|
||||
Effect.ensuring(Effect.sync(() => contenders.forEach((contender) => contender.release()))),
|
||||
)
|
||||
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) {
|
||||
const error = contender.error()
|
||||
if (error !== undefined) return error
|
||||
if (contender.child.exitCode !== null && contender.child.exitCode !== 0)
|
||||
return new Error(`Server process exited with code ${contender.child.exitCode}`)
|
||||
if (contender.child.signalCode !== null)
|
||||
return new Error(`Server process terminated by ${contender.child.signalCode}`)
|
||||
return undefined
|
||||
}
|
||||
|
||||
function contenderFinished(contender: Contender) {
|
||||
return contender.error() !== undefined || contender.child.exitCode !== null || contender.child.signalCode !== null
|
||||
}
|
||||
|
||||
/** Stop the registered local service. */
|
||||
export const stop = Effect.fn("service.stop")(function* (options: StopOptions = {}) {
|
||||
const existing = yield* find(options)
|
||||
|
|
|
|||
|
|
@ -1,8 +1,13 @@
|
|||
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 {
|
||||
contenderFailure,
|
||||
contenderFinished,
|
||||
type ServiceContender,
|
||||
spawnServiceContender,
|
||||
} from "../service-contender.js"
|
||||
import { defaultEnsureTiming, ensureTiming, type EnsureTiming } from "../service-timing.js"
|
||||
import type { ServiceHealth, ServiceStopResponse } from "./generated/types.js"
|
||||
|
||||
|
|
@ -14,11 +19,6 @@ export * from "../service.js"
|
|||
// intentionally implemented with Node APIs so Promise clients do not need
|
||||
// Effect or @effect/platform-node at runtime.
|
||||
|
||||
type Contender = {
|
||||
readonly child: ChildProcess
|
||||
readonly error: () => Error | undefined
|
||||
}
|
||||
|
||||
/** Discover a healthy, compatible local service without starting one. */
|
||||
export async function discover(options: DiscoverOptions = {}) {
|
||||
return (await discoverLocal(options))?.endpoint
|
||||
|
|
@ -35,7 +35,7 @@ async function discoverLocal(options: DiscoverOptions) {
|
|||
export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
|
||||
const timing = ensureTiming(options)
|
||||
const deadline = Date.now() + timing.promiseTimeout
|
||||
const contenders = new Set<Contender>()
|
||||
const contenders = new Set<ServiceContender>()
|
||||
let timeouts: { readonly info: Info; readonly count: number } | undefined
|
||||
let announced = false
|
||||
let lastSpawn = 0
|
||||
|
|
@ -50,79 +50,63 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
|
|||
const [command, ...args] = options.command ?? ["opencode", "serve", "--service"]
|
||||
if (command === undefined) throw new Error("Missing service command")
|
||||
try {
|
||||
const child = spawn(command, args, { detached: true, stdio: "ignore" })
|
||||
let error: Error | undefined
|
||||
child.once("error", (cause) => {
|
||||
error = new Error("Failed to start server", { cause })
|
||||
})
|
||||
child.unref()
|
||||
return { child, error: () => error }
|
||||
return spawnServiceContender(command, args)
|
||||
} catch (cause) {
|
||||
throw new Error("Failed to start server", { cause })
|
||||
}
|
||||
}
|
||||
|
||||
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, timing.requestTimeout)
|
||||
if (registration.timedOut && registration.info !== undefined) {
|
||||
timeouts = {
|
||||
info: registration.info,
|
||||
count: timeouts !== undefined && same(timeouts.info, registration.info) ? timeouts.count + 1 : 1,
|
||||
}
|
||||
if (timeouts.count >= 3) {
|
||||
announce("missing")
|
||||
await evict(registration.info, options, timing)
|
||||
timeouts = undefined
|
||||
lastSpawn = Date.now() - spawnDelay
|
||||
}
|
||||
} else timeouts = undefined
|
||||
try {
|
||||
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, timing.requestTimeout)
|
||||
if (registration.timedOut && registration.info !== undefined) {
|
||||
timeouts = {
|
||||
info: registration.info,
|
||||
count: timeouts !== undefined && same(timeouts.info, registration.info) ? timeouts.count + 1 : 1,
|
||||
}
|
||||
if (timeouts.count >= 3) {
|
||||
announce("missing")
|
||||
await evict(registration.info, options, timing)
|
||||
timeouts = undefined
|
||||
lastSpawn = Date.now() - spawnDelay
|
||||
}
|
||||
} else timeouts = undefined
|
||||
|
||||
if (registration.service !== undefined) {
|
||||
spawnDelay = timing.spawnDelay
|
||||
const service = registration.service
|
||||
const compatible = !service.legacy && (options.version === undefined || service.version === options.version)
|
||||
if (compatible && service.state === "ready") return service.endpoint
|
||||
if (compatible && service.state === "failed") throw new Error("Background service failed to start")
|
||||
if (!compatible) {
|
||||
announce("version-mismatch", service.version)
|
||||
await kill(service, options, timing).catch(() => undefined)
|
||||
lastSpawn = 0
|
||||
}
|
||||
} else {
|
||||
if (lastSpawn === 0 && registration.info !== undefined) lastSpawn = Date.now()
|
||||
const finished = [...contenders].filter(contenderFinished)
|
||||
const failure = finished.map(contenderFailure).find((error) => error !== undefined)
|
||||
if (finished.some((item) => item.child.exitCode === 0)) {
|
||||
spawnDelay = Math.min(spawnDelay * 2, timing.maxSpawnDelay)
|
||||
}
|
||||
finished.forEach((item) => contenders.delete(item))
|
||||
if (failure !== undefined && contenders.size === 0) throw failure
|
||||
// Keep one candidate plus one lock probe so a pre-lock stall cannot block recovery.
|
||||
if (contenders.size < 2 && Date.now() - lastSpawn >= spawnDelay) {
|
||||
announce("missing")
|
||||
contenders.add(spawnContender())
|
||||
lastSpawn = Date.now()
|
||||
if (registration.service !== undefined) {
|
||||
spawnDelay = timing.spawnDelay
|
||||
const service = registration.service
|
||||
const compatible = !service.legacy && (options.version === undefined || service.version === options.version)
|
||||
if (compatible && service.state === "ready") return service.endpoint
|
||||
if (compatible && service.state === "failed") throw new Error("Background service failed to start")
|
||||
if (!compatible) {
|
||||
announce("version-mismatch", service.version)
|
||||
await kill(service, options, timing).catch(() => undefined)
|
||||
lastSpawn = 0
|
||||
}
|
||||
} else {
|
||||
if (lastSpawn === 0 && registration.info !== undefined) lastSpawn = Date.now()
|
||||
const finished = [...contenders].filter(contenderFinished)
|
||||
const failure = finished.map(contenderFailure).find((error) => error !== undefined)
|
||||
if (finished.some((item) => item.child.exitCode === 0)) {
|
||||
spawnDelay = Math.min(spawnDelay * 2, timing.maxSpawnDelay)
|
||||
}
|
||||
finished.forEach((item) => contenders.delete(item))
|
||||
if (failure !== undefined && contenders.size === 0) throw failure
|
||||
// Keep one candidate plus one lock probe so a pre-lock stall cannot block recovery.
|
||||
if (contenders.size < 2 && Date.now() - lastSpawn >= spawnDelay) {
|
||||
announce("missing")
|
||||
contenders.add(spawnContender())
|
||||
lastSpawn = Date.now()
|
||||
}
|
||||
}
|
||||
await delay(timing.pollInterval)
|
||||
}
|
||||
await delay(timing.pollInterval)
|
||||
} finally {
|
||||
contenders.forEach((contender) => contender.release())
|
||||
}
|
||||
}
|
||||
|
||||
function contenderFailure(contender: Contender) {
|
||||
const error = contender.error()
|
||||
if (error !== undefined) return error
|
||||
if (contender.child.exitCode !== null && contender.child.exitCode !== 0)
|
||||
return new Error(`Server process exited with code ${contender.child.exitCode}`)
|
||||
if (contender.child.signalCode !== null)
|
||||
return new Error(`Server process terminated by ${contender.child.signalCode}`)
|
||||
return undefined
|
||||
}
|
||||
|
||||
function contenderFinished(contender: Contender) {
|
||||
return contender.error() !== undefined || contender.child.exitCode !== null || contender.child.signalCode !== null
|
||||
}
|
||||
|
||||
/** Stop the registered local service. */
|
||||
export async function stop(options: StopOptions = {}) {
|
||||
const existing = await find(options)
|
||||
|
|
|
|||
63
packages/client/src/service-contender.ts
Normal file
63
packages/client/src/service-contender.ts
Normal file
|
|
@ -0,0 +1,63 @@
|
|||
import { spawn, type ChildProcess } from "node:child_process"
|
||||
|
||||
export type ServiceContender = {
|
||||
readonly child: ChildProcess
|
||||
readonly error: () => Error | undefined
|
||||
readonly closed: () => boolean
|
||||
readonly stderr: () => string
|
||||
readonly release: () => void
|
||||
}
|
||||
|
||||
const stderrLimit = 8 * 1024
|
||||
|
||||
export function spawnServiceContender(command: string, args: ReadonlyArray<string>): ServiceContender {
|
||||
const child = spawn(command, args, { detached: true, stdio: ["ignore", "ignore", "pipe"] })
|
||||
let error: Error | undefined
|
||||
let closed = false
|
||||
let stderr = Buffer.alloc(0)
|
||||
const onStderr = (chunk: Buffer) => {
|
||||
const tail = chunk.subarray(-stderrLimit)
|
||||
stderr =
|
||||
tail.length === stderrLimit
|
||||
? Buffer.from(tail)
|
||||
: Buffer.concat([stderr.subarray(-(stderrLimit - tail.length)), tail])
|
||||
}
|
||||
child.stderr?.on("data", onStderr)
|
||||
if (child.stderr !== null && "unref" in child.stderr && typeof child.stderr.unref === "function") child.stderr.unref()
|
||||
child.once("error", (cause) => {
|
||||
error = new Error("Failed to start server", { cause })
|
||||
})
|
||||
child.once("close", () => {
|
||||
closed = true
|
||||
})
|
||||
child.unref()
|
||||
return {
|
||||
child,
|
||||
error: () => error,
|
||||
closed: () => closed,
|
||||
stderr: () => stderr.toString("utf8").trim(),
|
||||
release: () => {
|
||||
child.stderr?.off("data", onStderr)
|
||||
child.stderr?.resume()
|
||||
stderr = Buffer.alloc(0)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
export function contenderFailure(contender: ServiceContender) {
|
||||
const error = contender.error()
|
||||
if (error !== undefined) return error
|
||||
if (contender.child.exitCode !== null && contender.child.exitCode !== 0)
|
||||
return startupError(`Server process exited with code ${contender.child.exitCode}`, contender.stderr())
|
||||
if (contender.child.signalCode !== null)
|
||||
return startupError(`Server process terminated by ${contender.child.signalCode}`, contender.stderr())
|
||||
return undefined
|
||||
}
|
||||
|
||||
export function contenderFinished(contender: ServiceContender) {
|
||||
return contender.error() !== undefined || contender.closed()
|
||||
}
|
||||
|
||||
function startupError(message: string, stderr: string) {
|
||||
return new Error(stderr ? `${message}\n${stderr}` : message)
|
||||
}
|
||||
|
|
@ -3,6 +3,10 @@ import { appendFile, rename, writeFile } from "node:fs/promises"
|
|||
const [registration, mode, delay] = process.argv.slice(2)
|
||||
if (registration === undefined || mode === undefined) throw new Error("Missing service fixture arguments")
|
||||
if (mode === "failed") process.exit(1)
|
||||
if (mode === "stderr-failed") {
|
||||
process.stderr.write("x".repeat(16_384) + "\nactionable startup failure\n")
|
||||
process.exit(1)
|
||||
}
|
||||
if (mode === "record-start") {
|
||||
await writeFile(registration + ".started", "")
|
||||
process.exit(1)
|
||||
|
|
|
|||
|
|
@ -72,6 +72,21 @@ test("reports a failed registered service", async () => {
|
|||
)
|
||||
})
|
||||
|
||||
test("reports a bounded contender stderr tail with native promises", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const error = await Service.ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "stderr-failed"],
|
||||
}).catch((error: unknown) => error)
|
||||
|
||||
expect(error).toBeInstanceOf(Error)
|
||||
if (!(error instanceof Error)) throw error
|
||||
expect(error.message).toContain("actionable startup failure")
|
||||
expect(error.message.length).toBeLessThan(9_000)
|
||||
}, 10_000)
|
||||
|
||||
test("evicts an unresponsive registered service before starting its replacement", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
|
|
|
|||
|
|
@ -201,6 +201,23 @@ test("reports a contender that fails to start", async () => {
|
|||
).rejects.toThrow("Server process exited with code 1")
|
||||
})
|
||||
|
||||
test("reports a bounded contender stderr tail", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const error = await run(
|
||||
Service.ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "stderr-failed"],
|
||||
}),
|
||||
).catch((error: unknown) => error)
|
||||
|
||||
expect(error).toBeInstanceOf(Error)
|
||||
if (!(error instanceof Error)) throw error
|
||||
expect(error.message).toContain("actionable startup failure")
|
||||
expect(error.message.length).toBeLessThan(9_000)
|
||||
}, 10_000)
|
||||
|
||||
test("reports a contender terminated by a signal", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue