mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-12 12:45:07 -04:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| a44316f9f2 |
@@ -39,6 +39,8 @@ export const run = Effect.fnUntraced(function* (options: Options) {
|
||||
})
|
||||
|
||||
const processEffect = Effect.fnUntraced(function* (options: Options) {
|
||||
const serviceErrorFormat = process.env.OPENCODE_SERVICE_ERROR_FORMAT
|
||||
delete process.env.OPENCODE_SERVICE_ERROR_FORMAT
|
||||
const global = yield* Global.Service
|
||||
if (options.mode === "service") yield* Effect.sync(() => process.chdir(global.home))
|
||||
return yield* Effect.scoped(
|
||||
@@ -127,15 +129,7 @@ const processEffect = Effect.fnUntraced(function* (options: Options) {
|
||||
if (serviceOptions === undefined || port === undefined || !addressInUse(error)) return Effect.fail(error)
|
||||
return recognizeIncumbent(serviceOptions, hostname, port).pipe(
|
||||
Effect.flatMap((found) =>
|
||||
found
|
||||
? Effect.void
|
||||
: Effect.fail(
|
||||
new Error(
|
||||
`Managed service port ${port} on ${hostname} is already in use by another process. ` +
|
||||
"Configure another port with `opencode service set port <port>` and start the service again.",
|
||||
{ cause: error },
|
||||
),
|
||||
),
|
||||
found ? Effect.void : managedPortInUse(hostname, port, error, serviceErrorFormat),
|
||||
),
|
||||
)
|
||||
}),
|
||||
@@ -214,6 +208,17 @@ function serviceURL(hostname: string, port: number) {
|
||||
return `http://${hostname.includes(":") ? `[${hostname}]` : hostname}:${port}`
|
||||
}
|
||||
|
||||
function managedPortInUse(hostname: string, port: number, cause: unknown, format?: string) {
|
||||
const message =
|
||||
`Managed service port ${port} on ${hostname} is already in use by another process. ` +
|
||||
"Configure another port with `opencode service set port <port>` and start the service again."
|
||||
const failure = new Error(message, { cause })
|
||||
if (format !== "plain") return Effect.fail(failure)
|
||||
return Effect.sync(() => process.stderr.write(`OPENCODE_SERVICE_ERROR:${message}\n`)).pipe(
|
||||
Effect.andThen(Effect.fail(failure)),
|
||||
)
|
||||
}
|
||||
|
||||
function truthy(value?: string) {
|
||||
return value === "1" || value?.toLowerCase() === "true"
|
||||
}
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
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 { ServiceProcess } from "../service-process.js"
|
||||
|
||||
export * from "../service.js"
|
||||
/** Contents of the local service registration file. */
|
||||
@@ -17,11 +17,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. */
|
||||
@@ -52,11 +47,12 @@ const discoverLocal = Effect.fnUntraced(function* (options: DiscoverOptions) {
|
||||
// becomes discoverable. A contender is never killed merely for slow startup.
|
||||
/** Ensure a healthy, compatible local service is running. */
|
||||
export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOptions = {}) {
|
||||
const contenders = new Set<Contender>()
|
||||
const contenders = new Set<ServiceProcess.Contender>()
|
||||
let timeouts: { readonly info: Info; readonly count: number } | undefined
|
||||
let announced = false
|
||||
let lastSpawn = 0
|
||||
let spawnDelay = 5_000
|
||||
let lastFailure: Error | undefined
|
||||
const announce = (reason: "missing" | "version-mismatch", previousVersion?: string) =>
|
||||
Effect.sync(() => {
|
||||
if (announced) return
|
||||
@@ -67,15 +63,7 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
|
||||
const [command, ...args] = options.command ?? ["opencode", "serve", "--service"]
|
||||
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 }
|
||||
},
|
||||
try: () => ServiceProcess.start(command, args),
|
||||
catch: (cause) => new Error("Failed to start server", { cause }),
|
||||
})
|
||||
})
|
||||
@@ -108,8 +96,9 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
|
||||
return Option.none<LocalService>()
|
||||
} else if (lastSpawn === 0 && info !== undefined) lastSpawn = Date.now()
|
||||
|
||||
const finished = [...contenders].filter(contenderFinished)
|
||||
const failure = finished.map(contenderFailure).find((error): error is Error => error !== undefined)
|
||||
const finished = [...contenders].filter(ServiceProcess.finished)
|
||||
const failure = finished.map(ServiceProcess.failure).find((error): error is Error => error !== undefined)
|
||||
if (failure !== undefined) lastFailure = failure
|
||||
if (finished.some((item) => item.child.exitCode === 0)) {
|
||||
spawnDelay = Math.min(spawnDelay * 2, 30_000)
|
||||
}
|
||||
@@ -129,24 +118,10 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
|
||||
}),
|
||||
)
|
||||
if (Option.isNone(found))
|
||||
return yield* Effect.fail(new Error("Timed out waiting for the background service to start"))
|
||||
return yield* Effect.fail(lastFailure ?? 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,8 @@
|
||||
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 { ServiceProcess } from "../service-process.js"
|
||||
import type { ServiceHealth, ServiceStopResponse } from "./generated/types.js"
|
||||
|
||||
export * from "../service.js"
|
||||
@@ -13,11 +13,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
|
||||
@@ -33,11 +28,12 @@ async function discoverLocal(options: DiscoverOptions) {
|
||||
/** Ensure a healthy, compatible local service is running. */
|
||||
export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
|
||||
const deadline = Date.now() + 120_000
|
||||
const contenders = new Set<Contender>()
|
||||
const contenders = new Set<ServiceProcess.Contender>()
|
||||
let timeouts: { readonly info: Info; readonly count: number } | undefined
|
||||
let announced = false
|
||||
let lastSpawn = 0
|
||||
let spawnDelay = 5_000
|
||||
let lastFailure: Error | undefined
|
||||
|
||||
const announce = (reason: "missing" | "version-mismatch", previousVersion?: string) => {
|
||||
if (announced) return
|
||||
@@ -47,21 +43,11 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
|
||||
const spawnContender = () => {
|
||||
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 }
|
||||
} catch (cause) {
|
||||
throw new Error("Failed to start server", { cause })
|
||||
}
|
||||
return ServiceProcess.start(command, args)
|
||||
}
|
||||
|
||||
while (true) {
|
||||
if (Date.now() >= deadline) throw new Error("Timed out waiting for the background service to start")
|
||||
if (Date.now() >= deadline) throw lastFailure ?? new Error("Timed out waiting for the background service to start")
|
||||
const registration = await registered(options.file, true)
|
||||
if (registration.timedOut && registration.info !== undefined) {
|
||||
timeouts = {
|
||||
@@ -89,8 +75,9 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
|
||||
}
|
||||
} else {
|
||||
if (lastSpawn === 0 && registration.info !== undefined) lastSpawn = Date.now()
|
||||
const finished = [...contenders].filter(contenderFinished)
|
||||
const failure = finished.map(contenderFailure).find((error) => error !== undefined)
|
||||
const finished = [...contenders].filter(ServiceProcess.finished)
|
||||
const failure = finished.map(ServiceProcess.failure).find((error) => error !== undefined)
|
||||
if (failure !== undefined) lastFailure = failure
|
||||
if (finished.some((item) => item.child.exitCode === 0)) {
|
||||
spawnDelay = Math.min(spawnDelay * 2, 30_000)
|
||||
}
|
||||
@@ -107,20 +94,6 @@ export async function ensure(options: EnsureOptions = {}): Promise<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 async function stop(options: StopOptions = {}) {
|
||||
const existing = await find(options)
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
export * as ServiceProcess from "./service-process"
|
||||
|
||||
import { spawn, type ChildProcess } from "node:child_process"
|
||||
|
||||
const errorPrefix = "OPENCODE_SERVICE_ERROR:"
|
||||
|
||||
export type Contender = {
|
||||
readonly child: ChildProcess
|
||||
readonly error: () => Error | undefined
|
||||
readonly startupError: () => string
|
||||
}
|
||||
|
||||
export function start(command: string, args: ReadonlyArray<string>) {
|
||||
try {
|
||||
const child = spawn(command, args, {
|
||||
detached: true,
|
||||
stdio: ["ignore", "ignore", "pipe"],
|
||||
env: { ...process.env, OPENCODE_SERVICE_ERROR_FORMAT: "plain" },
|
||||
})
|
||||
let error: Error | undefined
|
||||
let pending = ""
|
||||
let startupError = ""
|
||||
child.once("error", (cause) => {
|
||||
error = new Error("Failed to start server", { cause })
|
||||
})
|
||||
child.stderr?.on("data", (chunk) => {
|
||||
const lines = (pending + chunk.toString()).split(/\r?\n/)
|
||||
pending = lines.pop()?.slice(-64 * 1024) ?? ""
|
||||
const message = lines.findLast((line) => line.startsWith(errorPrefix))
|
||||
if (message !== undefined) startupError = message.slice(errorPrefix.length)
|
||||
})
|
||||
unref(child.stderr)
|
||||
child.unref()
|
||||
return { child, error: () => error, startupError: () => startupError } satisfies Contender
|
||||
} catch (cause) {
|
||||
throw new Error("Failed to start server", { cause })
|
||||
}
|
||||
}
|
||||
|
||||
export function failure(contender: Contender) {
|
||||
const error = contender.error()
|
||||
if (error !== undefined) return error
|
||||
if (contender.child.exitCode !== null && contender.child.exitCode !== 0)
|
||||
return new Error(contender.startupError() || `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
|
||||
}
|
||||
|
||||
export function finished(contender: Contender) {
|
||||
return contender.error() !== undefined || contender.child.exitCode !== null || contender.child.signalCode !== null
|
||||
}
|
||||
|
||||
function unref(stream: ChildProcess["stderr"]) {
|
||||
if (!stream || !("unref" in stream) || typeof stream.unref !== "function") return
|
||||
stream.unref()
|
||||
}
|
||||
@@ -3,6 +3,11 @@ 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 === "failed-message") {
|
||||
console.error("sensitive startup detail")
|
||||
console.error("OPENCODE_SERVICE_ERROR:Managed service port is already in use")
|
||||
process.exit(1)
|
||||
}
|
||||
if (mode === "record-start") {
|
||||
await writeFile(registration + ".started", "")
|
||||
process.exit(1)
|
||||
|
||||
@@ -70,6 +70,19 @@ test("reports a failed registered service", async () => {
|
||||
)
|
||||
})
|
||||
|
||||
test("reports the native contender's startup error", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
|
||||
await expect(
|
||||
Service.ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "failed-message"],
|
||||
}),
|
||||
).rejects.toThrow(/^Managed service port is already in use$/)
|
||||
}, 10_000)
|
||||
|
||||
test("evicts an unresponsive registered service before starting its replacement", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
|
||||
@@ -197,6 +197,20 @@ test("reports a contender that fails to start", async () => {
|
||||
).rejects.toThrow("Server process exited with code 1")
|
||||
}, 10_000)
|
||||
|
||||
test("reports the contender's startup error", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
await expect(
|
||||
run(
|
||||
Service.ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "failed-message"],
|
||||
}),
|
||||
),
|
||||
).rejects.toThrow(/^Managed service port is already in use$/)
|
||||
}, 10_000)
|
||||
|
||||
test("reports a contender terminated by a signal", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
|
||||
Reference in New Issue
Block a user