mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-12 12:45:07 -04:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 3a8961a90a | |||
| 01b17c4d53 |
@@ -188,12 +188,14 @@
|
||||
"dependencies": {
|
||||
"@opencode-ai/protocol": "workspace:*",
|
||||
"@opencode-ai/schema": "workspace:*",
|
||||
"semver": "catalog:",
|
||||
},
|
||||
"devDependencies": {
|
||||
"@effect/platform-node": "catalog:",
|
||||
"@opencode-ai/httpapi-codegen": "workspace:*",
|
||||
"@tsconfig/bun": "catalog:",
|
||||
"@types/bun": "catalog:",
|
||||
"@types/semver": "catalog:",
|
||||
"@typescript/native-preview": "catalog:",
|
||||
"effect": "catalog:",
|
||||
},
|
||||
@@ -6351,6 +6353,8 @@
|
||||
|
||||
"@opencode-ai/cli/semver": ["semver@7.7.4", "", { "bin": { "semver": "bin/semver.js" } }, "sha512-vFKC2IEtQnVhpT78h1Yp8wzwrf8CM+MzKMHGJZfBtzhZNycRFnXsHk6E5TxIkkMsgNS7mdX3AGB7x2QM2di4lA=="],
|
||||
|
||||
"@opencode-ai/client/semver": ["semver@7.7.4", "", { "bin": { "semver": "bin/semver.js" } }, "sha512-vFKC2IEtQnVhpT78h1Yp8wzwrf8CM+MzKMHGJZfBtzhZNycRFnXsHk6E5TxIkkMsgNS7mdX3AGB7x2QM2di4lA=="],
|
||||
|
||||
"@opencode-ai/console-app/@smithy/eventstream-codec": ["@smithy/eventstream-codec@4.2.7", "", { "dependencies": { "@aws-crypto/crc32": "5.2.0", "@smithy/types": "^4.11.0", "@smithy/util-hex-encoding": "^4.2.0", "tslib": "^2.6.2" } }, "sha512-DrpkEoM3j9cBBWhufqBwnbbn+3nf1N9FP6xuVJ+e220jbactKuQgaZwjwP5CP1t+O94brm2JgVMD2atMGX3xIQ=="],
|
||||
|
||||
"@opencode-ai/console-app/@smithy/util-utf8": ["@smithy/util-utf8@4.2.0", "", { "dependencies": { "@smithy/util-buffer-from": "^4.2.0", "tslib": "^2.6.2" } }, "sha512-zBPfuzoI8xyBtR2P6WQj63Rz8i3AmfAaJLuNG8dWsfvPe8lO4aCPYLn879mEgHndZH1zQ2oXmG8O1GGzzaoZiw=="],
|
||||
|
||||
@@ -23,7 +23,14 @@ export default Runtime.handler(Commands, (input) =>
|
||||
server: Option.getOrUndefined(input.server),
|
||||
standalone: input.standalone,
|
||||
mismatch: "replace",
|
||||
confirmDowngrade: (serverVersion) =>
|
||||
Effect.promise(async () => {
|
||||
const confirmed = await preflight.confirmDowngrade(serverVersion)
|
||||
if (confirmed) preflight.begin(serverVersion)
|
||||
return confirmed
|
||||
}),
|
||||
onStart: (reason, previousVersion) => {
|
||||
if (preflight.active()) return
|
||||
if (reason === "version-mismatch" && preflight.begin(previousVersion)) return
|
||||
process.stderr.write(
|
||||
reason === "version-mismatch"
|
||||
|
||||
@@ -8,7 +8,7 @@ import { ServiceConfig } from "../../../services/service-config"
|
||||
export default Runtime.handler(
|
||||
Commands.commands.service.commands.restart,
|
||||
Effect.fn("cli.service.restart")(function* () {
|
||||
const options = yield* ServiceConfig.options()
|
||||
const options = yield* ServiceConfig.options({ checkVersion: true })
|
||||
yield* Service.stop(options)
|
||||
const transport = yield* Service.ensure(options)
|
||||
process.stdout.write(transport.url + EOL)
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { Service, type Endpoint, type EnsureOptions } from "@opencode-ai/client/effect/service"
|
||||
import { Service, VersionMismatchError, type Endpoint, type EnsureOptions } from "@opencode-ai/client/effect/service"
|
||||
import { ClientError, isUnauthorizedError, OpenCode } from "@opencode-ai/client/promise"
|
||||
import { OPENCODE_VERSION } from "../version"
|
||||
import { Effect, Redacted } from "effect"
|
||||
import { Effect, Redacted, Result } from "effect"
|
||||
import { Env } from "../env"
|
||||
import { ServiceConfig } from "./service-config"
|
||||
import { Standalone } from "./standalone"
|
||||
@@ -11,6 +11,7 @@ export type Args = {
|
||||
readonly standalone?: boolean
|
||||
readonly mismatch?: "replace" | "ignore" | "error"
|
||||
readonly onStart?: EnsureOptions["onStart"]
|
||||
readonly confirmDowngrade?: (serverVersion: string, clientVersion: string) => Effect.Effect<boolean>
|
||||
}
|
||||
|
||||
export type Resolved = {
|
||||
@@ -45,7 +46,7 @@ export const resolve = Effect.fn("cli.server-connection.resolve")(function* (arg
|
||||
const mismatch = args.mismatch ?? "ignore"
|
||||
const options = yield* ServiceConfig.options({ checkVersion: mismatch !== "ignore" })
|
||||
return {
|
||||
endpoint: yield* resolveManaged({ ...options, onStart: args.onStart }, mismatch),
|
||||
endpoint: yield* resolveManaged({ ...options, onStart: args.onStart }, mismatch, args.confirmDowngrade),
|
||||
service: managedService(options),
|
||||
} satisfies Resolved
|
||||
})
|
||||
@@ -57,13 +58,21 @@ function managedService(options: EnsureOptions) {
|
||||
restart: () =>
|
||||
Effect.gen(function* () {
|
||||
yield* Service.stop(options)
|
||||
yield* Service.ensure(reconnectOptions)
|
||||
yield* Service.ensure(options)
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
const resolveManaged = Effect.fnUntraced(function* (options: EnsureOptions, mismatch: NonNullable<Args["mismatch"]>) {
|
||||
if (mismatch === "replace") return yield* Service.ensure(options)
|
||||
const resolveManaged = Effect.fnUntraced(function* (
|
||||
options: EnsureOptions,
|
||||
mismatch: NonNullable<Args["mismatch"]>,
|
||||
confirmDowngrade?: Args["confirmDowngrade"],
|
||||
) {
|
||||
if (mismatch === "replace") {
|
||||
const result = yield* Effect.result(Service.ensure(options))
|
||||
if (Result.isSuccess(result)) return result.success
|
||||
return yield* confirmManagedDowngrade(options, result.failure, confirmDowngrade)
|
||||
}
|
||||
if (mismatch === "ignore") return yield* Service.ensure({ ...options, version: undefined })
|
||||
|
||||
const compatible = yield* Service.discover(options)
|
||||
@@ -74,6 +83,29 @@ const resolveManaged = Effect.fnUntraced(function* (options: EnsureOptions, mism
|
||||
return yield* Service.ensure(options)
|
||||
})
|
||||
|
||||
export const confirmManagedDowngrade = Effect.fnUntraced(function* (
|
||||
options: EnsureOptions,
|
||||
error: unknown,
|
||||
confirm?: Args["confirmDowngrade"],
|
||||
) {
|
||||
if (
|
||||
!(error instanceof VersionMismatchError) ||
|
||||
error.serverVersion === undefined ||
|
||||
error.clientVersion === undefined ||
|
||||
!Service.canReplaceVersion(error.clientVersion, error.serverVersion) ||
|
||||
confirm === undefined
|
||||
)
|
||||
return yield* Effect.fail(error)
|
||||
if (!(yield* confirm(error.serverVersion, error.clientVersion)))
|
||||
return yield* Effect.fail(
|
||||
new Error(`${error.message}. Run \`opencode2 service restart\` to activate this installed version.`, {
|
||||
cause: error,
|
||||
}),
|
||||
)
|
||||
yield* Service.stop(options)
|
||||
return yield* Service.ensure(options)
|
||||
})
|
||||
|
||||
function connectError(endpoint: Endpoint, cause: unknown) {
|
||||
if (isUnauthorizedError(cause)) {
|
||||
return new Error(
|
||||
|
||||
@@ -104,10 +104,15 @@ export const options = Effect.fnUntraced(function* (input: { readonly checkVersi
|
||||
return {
|
||||
file,
|
||||
version: input.checkVersion ? OPENCODE_VERSION : undefined,
|
||||
canReplace: (version: string | undefined) => Service.canReplaceVersion(version, OPENCODE_VERSION),
|
||||
command: [...selfCommand(), "serve", "--service"],
|
||||
}
|
||||
})
|
||||
|
||||
export function canReplaceVersion(serverVersion: string | undefined, clientVersion = OPENCODE_VERSION) {
|
||||
return Service.canReplaceVersion(serverVersion, clientVersion)
|
||||
}
|
||||
|
||||
export const read = Effect.fn("cli.service-config.read")(function* () {
|
||||
const { fs, configFile, legacyConfigFile } = yield* paths
|
||||
if (legacyConfigFile) yield* migrateConfig(legacyConfigFile, configFile)
|
||||
|
||||
@@ -28,7 +28,9 @@ const transitionDuration = 420
|
||||
const completionHold = 650
|
||||
|
||||
export type Handle = {
|
||||
readonly active: () => boolean
|
||||
readonly begin: (from?: string) => boolean
|
||||
readonly confirmDowngrade: (from: string) => Promise<boolean>
|
||||
readonly loading: () => void
|
||||
readonly finish: () => Promise<Handoff | undefined>
|
||||
readonly fail: (message: string) => Promise<void>
|
||||
@@ -44,6 +46,7 @@ export type Handoff = {
|
||||
export const make = (): Handle => {
|
||||
let session: Promise<Session | undefined> | undefined
|
||||
return {
|
||||
active: () => session !== undefined,
|
||||
begin: (from) => {
|
||||
if (!process.stdout.isTTY || !process.stdin.isTTY) return false
|
||||
session ??= open(from).catch(() => {
|
||||
@@ -52,6 +55,7 @@ export const make = (): Handle => {
|
||||
})
|
||||
return true
|
||||
},
|
||||
confirmDowngrade: (from) => confirmDowngrade(from),
|
||||
loading: () => {
|
||||
void session?.then((active) => active?.loading())
|
||||
},
|
||||
@@ -197,6 +201,40 @@ async function open(from?: string): Promise<Session> {
|
||||
}
|
||||
}
|
||||
|
||||
async function confirmDowngrade(from: string) {
|
||||
if (!process.stdout.isTTY || !process.stdin.isTTY) return false
|
||||
const renderer = await createCliRenderer({
|
||||
stdin: process.stdin,
|
||||
useMouse: false,
|
||||
autoFocus: true,
|
||||
openConsoleOnError: false,
|
||||
exitOnCtrlC: false,
|
||||
screenMode: "split-footer",
|
||||
footerHeight: 4,
|
||||
targetFps: 30,
|
||||
useKittyKeyboard: {},
|
||||
externalOutputMode: "capture-stdout",
|
||||
consoleMode: "disabled",
|
||||
})
|
||||
const result = Promise.withResolvers<boolean>()
|
||||
const onKeypress = (event: { readonly name: string; readonly ctrl: boolean }) => {
|
||||
if (event.name === "return") return finish(true)
|
||||
if (event.name === "escape" || (event.ctrl && event.name === "c")) finish(false)
|
||||
}
|
||||
const finish = (confirmed: boolean) => {
|
||||
renderer.keyInput.off("keypress", onKeypress)
|
||||
if (!renderer.isDestroyed) renderer.destroy()
|
||||
result.resolve(confirmed)
|
||||
}
|
||||
renderer.keyInput.on("keypress", onKeypress)
|
||||
await render(() => <DowngradeFooter from={from} />, renderer).catch((error) => {
|
||||
renderer.keyInput.off("keypress", onKeypress)
|
||||
if (!renderer.isDestroyed) renderer.destroy()
|
||||
throw error
|
||||
})
|
||||
return result.promise
|
||||
}
|
||||
|
||||
const colors = {
|
||||
accent: RGBA.fromHex("#a6b8ff"),
|
||||
accentBright: RGBA.fromHex("#eef1ff"),
|
||||
@@ -447,7 +485,7 @@ function UpdateFooter(props: {
|
||||
})
|
||||
|
||||
return (
|
||||
<box width="100%" height={4} flexDirection="row" gap={1} live={props.animating()}>
|
||||
<box width="100%" height={4} flexDirection="row" gap={1} paddingLeft={1} live={props.animating()}>
|
||||
<Monogram ink={monogramInk} />
|
||||
<box flexDirection="column" flexGrow={1} overflow="hidden">
|
||||
<CellLine cells={header()} />
|
||||
@@ -471,6 +509,20 @@ function UpdateFooter(props: {
|
||||
)
|
||||
}
|
||||
|
||||
function DowngradeFooter(props: { from: string }) {
|
||||
return (
|
||||
<box width="100%" height={4} flexDirection="column" paddingLeft={1}>
|
||||
<text fg={colors.text}>
|
||||
<span style={{ fg: colors.muted }}>Background service </span>
|
||||
<span style={{ fg: colors.accent }}>{props.from}</span>
|
||||
<span style={{ fg: colors.muted }}> is newer than installed </span>
|
||||
<span style={{ fg: colors.accent }}>{OPENCODE_VERSION}</span>
|
||||
</text>
|
||||
<text fg={colors.muted}>Press Enter to downgrade and restart · Esc to cancel</text>
|
||||
</box>
|
||||
)
|
||||
}
|
||||
|
||||
function CellLine(props: { cells: ReadonlyArray<Cell> }) {
|
||||
return (
|
||||
<text truncate>
|
||||
|
||||
@@ -8,6 +8,7 @@ import os from "node:os"
|
||||
import path from "node:path"
|
||||
import { ServerConnection } from "../src/services/server-connection"
|
||||
import { ServiceConfig } from "../src/services/service-config"
|
||||
import { VersionMismatchError } from "@opencode-ai/client/effect/service"
|
||||
|
||||
test("resolution groups Effect-native lifecycle operations only for the managed service", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-server-resolution-"))
|
||||
@@ -69,3 +70,35 @@ test("service options only require a matching version when requested", async ()
|
||||
await fs.rm(root, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
test("downgrade confirmation is offered only when the running service is newer", async () => {
|
||||
const options = { version: "0.0.0-next-17271" }
|
||||
const offered: string[] = []
|
||||
const confirm = (serverVersion: string) =>
|
||||
Effect.sync(() => {
|
||||
offered.push(serverVersion)
|
||||
return false
|
||||
})
|
||||
|
||||
await expect(
|
||||
Effect.runPromise(
|
||||
ServerConnection.confirmManagedDowngrade(
|
||||
options,
|
||||
new VersionMismatchError(options.version, "0.0.0-next-17272"),
|
||||
confirm,
|
||||
).pipe(Effect.provide(NodeFileSystem.layer)),
|
||||
),
|
||||
).rejects.toThrow("Run `opencode2 service restart`")
|
||||
expect(offered).toEqual(["0.0.0-next-17272"])
|
||||
|
||||
await expect(
|
||||
Effect.runPromise(
|
||||
ServerConnection.confirmManagedDowngrade(
|
||||
{ version: "0.0.0-next-17272" },
|
||||
new VersionMismatchError("0.0.0-next-17272", "0.0.0-next-17271"),
|
||||
confirm,
|
||||
).pipe(Effect.provide(NodeFileSystem.layer)),
|
||||
),
|
||||
).rejects.toThrow("does not match")
|
||||
expect(offered).toEqual(["0.0.0-next-17272"])
|
||||
})
|
||||
|
||||
@@ -47,6 +47,35 @@ test("service filenames share release channels and identify preview channels", (
|
||||
expect(ServiceConfig.versionBelongsToChannel("1.2.3", "preview-a")).toBe(false)
|
||||
})
|
||||
|
||||
test("only newer clients replace managed service versions", () => {
|
||||
expect(ServiceConfig.canReplaceVersion("0.0.0-next-17271", "0.0.0-next-17272")).toBe(true)
|
||||
expect(ServiceConfig.canReplaceVersion("0.0.0-next-17272", "0.0.0-next-17271")).toBe(false)
|
||||
expect(ServiceConfig.canReplaceVersion("0.0.0-next-17272", "0.0.0-next-17272")).toBe(false)
|
||||
expect(ServiceConfig.canReplaceVersion(undefined, "0.0.0-next-17272")).toBe(false)
|
||||
expect(ServiceConfig.canReplaceVersion("development-a", "development-b")).toBe(false)
|
||||
expect(ServiceConfig.canReplaceVersion("development-b", "development-a")).toBe(false)
|
||||
})
|
||||
|
||||
test("managed version replacement can never be mutual", () => {
|
||||
const versions = [
|
||||
undefined,
|
||||
"1.0.0",
|
||||
"1.0.1",
|
||||
"0.0.0-next-9999",
|
||||
"0.0.0-next-15000",
|
||||
"0.0.0-next-15000.1",
|
||||
"0.0.0-next-15000.2",
|
||||
"development-a",
|
||||
"development-b",
|
||||
]
|
||||
for (const left of versions) {
|
||||
for (const right of versions) {
|
||||
if (left === undefined || right === undefined) continue
|
||||
expect(ServiceConfig.canReplaceVersion(left, right) && ServiceConfig.canReplaceVersion(right, left)).toBe(false)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
test("service config migrates from the hashed channel filename", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-config-migration-"))
|
||||
const legacy = path.join(root, ServiceConfig.legacyFilename("preview-a")!)
|
||||
|
||||
@@ -33,7 +33,8 @@
|
||||
},
|
||||
"dependencies": {
|
||||
"@opencode-ai/schema": "workspace:*",
|
||||
"@opencode-ai/protocol": "workspace:*"
|
||||
"@opencode-ai/protocol": "workspace:*",
|
||||
"semver": "catalog:"
|
||||
},
|
||||
"peerDependencies": {
|
||||
"effect": "4.0.0-beta.101"
|
||||
@@ -49,6 +50,7 @@
|
||||
"@tsconfig/bun": "catalog:",
|
||||
"@types/bun": "catalog:",
|
||||
"@typescript/native-preview": "catalog:",
|
||||
"@types/semver": "catalog:",
|
||||
"effect": "catalog:"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,7 +2,14 @@ import { ServiceStatus } from "@opencode-ai/protocol/groups/health"
|
||||
import { Effect, FileSystem, Option, Schedule, Schema } from "effect"
|
||||
import { homedir } from "node:os"
|
||||
import { join } from "node:path"
|
||||
import type { DiscoverOptions, Endpoint, EnsureOptions, StopOptions } from "../service.js"
|
||||
import {
|
||||
canReplaceVersion,
|
||||
VersionMismatchError,
|
||||
type DiscoverOptions,
|
||||
type Endpoint,
|
||||
type EnsureOptions,
|
||||
type StopOptions,
|
||||
} from "../service.js"
|
||||
import {
|
||||
contenderFailure,
|
||||
contenderFinished,
|
||||
@@ -55,7 +62,6 @@ const discoverLocal = Effect.fnUntraced(function* (options: DiscoverOptions) {
|
||||
export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOptions = {}) {
|
||||
const timing = ensureTiming(options)
|
||||
const contenders = new Set<ServiceContender>()
|
||||
let timeouts: { readonly info: Info; readonly count: number } | undefined
|
||||
let announced = false
|
||||
let lastSpawn = 0
|
||||
let spawnDelay = timing.spawnDelay
|
||||
@@ -79,18 +85,6 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
|
||||
const registration = yield* registered(options.file, true, timing.requestTimeout)
|
||||
const info = registration.info
|
||||
const service = registration.service
|
||||
if (registration.timedOut && info !== undefined) {
|
||||
timeouts = {
|
||||
info,
|
||||
count: timeouts !== undefined && same(timeouts.info, info) ? timeouts.count + 1 : 1,
|
||||
}
|
||||
if (timeouts.count >= 3) {
|
||||
yield* announce("missing")
|
||||
yield* evict(info, options, timing)
|
||||
timeouts = undefined
|
||||
lastSpawn = Date.now() - spawnDelay
|
||||
}
|
||||
} else timeouts = undefined
|
||||
if (service !== undefined) {
|
||||
spawnDelay = timing.spawnDelay
|
||||
const compatible = !service.legacy && (options.version === undefined || service.version === options.version)
|
||||
@@ -98,8 +92,10 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
|
||||
if (compatible && service.state === "failed")
|
||||
return yield* Effect.fail(new Error("Background service failed to start"))
|
||||
if (compatible) return Option.none<LocalService>()
|
||||
if (options.canReplace?.(service.version) !== true)
|
||||
return yield* Effect.fail(new VersionMismatchError(options.version, service.version))
|
||||
yield* kill(service, timing)
|
||||
yield* announce("version-mismatch", service.version)
|
||||
yield* kill(service, options, timing).pipe(Effect.ignore)
|
||||
lastSpawn = 0
|
||||
return Option.none<LocalService>()
|
||||
} else if (lastSpawn === 0 && info !== undefined) lastSpawn = Date.now()
|
||||
@@ -133,7 +129,11 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
|
||||
/** Stop the registered local service. */
|
||||
export const stop = Effect.fn("service.stop")(function* (options: StopOptions = {}) {
|
||||
const existing = yield* find(options)
|
||||
if (existing !== undefined) yield* kill(existing, options, defaultEnsureTiming)
|
||||
if (existing !== undefined) yield* kill(existing, defaultEnsureTiming)
|
||||
if (existing === undefined && (yield* read(options.file)) !== undefined)
|
||||
return yield* Effect.fail(
|
||||
new Error("Background service is not responding; stop its process manually and try again"),
|
||||
)
|
||||
})
|
||||
|
||||
function fallback() {
|
||||
@@ -178,7 +178,7 @@ type LocalService = {
|
||||
}
|
||||
|
||||
const probe = Effect.fnUntraced(function* (info: Info, allowLegacy = false) {
|
||||
return (yield* probeResult(info, allowLegacy)).service
|
||||
return yield* probeResult(info, allowLegacy)
|
||||
})
|
||||
|
||||
const probeResult = Effect.fnUntraced(function* (
|
||||
@@ -205,41 +205,34 @@ const probeResult = Effect.fnUntraced(function* (
|
||||
(cause: unknown) => ({ cause }),
|
||||
),
|
||||
)
|
||||
if ("cause" in result) return { service: undefined, timedOut: signal.aborted }
|
||||
if ("cause" in result) return undefined
|
||||
const response = result.value.response
|
||||
const body = result.value.body
|
||||
const health = decodeHealth(body)
|
||||
if (Option.isSome(health)) {
|
||||
if (health.value.pid !== info.pid) return { service: undefined, timedOut: false }
|
||||
if (info.version !== undefined && health.value.version !== info.version)
|
||||
return { service: undefined, timedOut: false }
|
||||
if (health.value.pid !== info.pid) return undefined
|
||||
if (info.version !== undefined && health.value.version !== info.version) return undefined
|
||||
return {
|
||||
service: {
|
||||
info,
|
||||
endpoint,
|
||||
version: health.value.version,
|
||||
state: response.ok ? "ready" : response.status === 500 ? "failed" : "waiting",
|
||||
legacy: false,
|
||||
} satisfies LocalService,
|
||||
timedOut: false,
|
||||
}
|
||||
info,
|
||||
endpoint,
|
||||
version: health.value.version,
|
||||
state: response.ok ? "ready" : response.status === 500 ? "failed" : "waiting",
|
||||
legacy: false,
|
||||
} satisfies LocalService
|
||||
}
|
||||
if (
|
||||
!allowLegacy ||
|
||||
Option.isNone(decodeLegacyHealth(body)) ||
|
||||
(typeof body === "object" && body !== null && ("version" in body || "pid" in body))
|
||||
)
|
||||
return { service: undefined, timedOut: false }
|
||||
return {
|
||||
service: { info, endpoint, state: "ready", legacy: true } satisfies LocalService,
|
||||
timedOut: false,
|
||||
}
|
||||
return undefined
|
||||
return { info, endpoint, state: "ready", legacy: true } satisfies LocalService
|
||||
})
|
||||
|
||||
const registered = Effect.fnUntraced(function* (file?: string, allowLegacy = false, timeout?: number) {
|
||||
const info = yield* read(file)
|
||||
if (info === undefined) return { info: undefined, service: undefined, timedOut: false }
|
||||
return { info, ...(yield* probeResult(info, allowLegacy, timeout)) }
|
||||
if (info === undefined) return { info: undefined, service: undefined }
|
||||
return { info, service: yield* probeResult(info, allowLegacy, timeout) }
|
||||
})
|
||||
|
||||
// Health-checked lookup without the version gate: lifecycle operations must be
|
||||
@@ -248,14 +241,10 @@ const find = Effect.fnUntraced(function* (options: { readonly file?: string }) {
|
||||
return (yield* registered(options.file, true)).service
|
||||
})
|
||||
|
||||
// 50ms cadence bounded at ~5s, shared by stop escalation and each ensure
|
||||
// discovery window.
|
||||
// Poll until an authenticated stop exits, bounded by the configured stop window.
|
||||
const poll = (timing: EnsureTiming) =>
|
||||
Schedule.max([Schedule.spaced(timing.stopPollInterval), Schedule.recurs(timing.stopPollAttempts)])
|
||||
|
||||
const signal = (pid: number, name: NodeJS.Signals) =>
|
||||
Effect.try({ try: () => process.kill(pid, name), catch: (cause) => cause }).pipe(Effect.ignore)
|
||||
|
||||
const stopped = Effect.fnUntraced(function* (pid: number) {
|
||||
const running = yield* Effect.try({ try: () => process.kill(pid, 0), catch: () => false }).pipe(
|
||||
Effect.orElseSucceed(() => false),
|
||||
@@ -264,44 +253,16 @@ const stopped = Effect.fnUntraced(function* (pid: number) {
|
||||
return yield* Effect.fail(new Error(`Server process ${pid} is still running`))
|
||||
})
|
||||
|
||||
function same(left: Info, right: Info) {
|
||||
return left.id === right.id && left.version === right.version && left.url === right.url && left.pid === right.pid
|
||||
}
|
||||
|
||||
const evict = Effect.fnUntraced(function* (info: Info, options: { readonly file?: string }, timing: EnsureTiming) {
|
||||
const current = yield* read(options.file)
|
||||
if (current === undefined || !same(current, info)) return
|
||||
yield* signal(info.pid, "SIGTERM")
|
||||
const done = yield* stopped(info.pid).pipe(Effect.retry(poll(timing)), Effect.option)
|
||||
if (Option.isSome(done)) return
|
||||
|
||||
const latest = yield* read(options.file)
|
||||
if (latest === undefined || !same(latest, info)) return
|
||||
yield* signal(info.pid, "SIGKILL")
|
||||
yield* stopped(info.pid).pipe(Effect.retry(poll(timing)))
|
||||
})
|
||||
|
||||
const kill = Effect.fnUntraced(function* (
|
||||
service: LocalService,
|
||||
options: { readonly file?: string },
|
||||
timing: EnsureTiming,
|
||||
) {
|
||||
const kill = Effect.fnUntraced(function* (service: LocalService, timing: EnsureTiming) {
|
||||
const requested = yield* requestStop(service, timing.requestTimeout)
|
||||
if (requested === "rejected") return
|
||||
if (requested === "rejected") return yield* Effect.fail(new Error("Background service rejected the stop request"))
|
||||
if (requested === "unsupported") {
|
||||
// A stale registration may point at a reused PID. Authenticate again
|
||||
// immediately before the legacy signal fallback.
|
||||
const current = yield* find(options)
|
||||
if (current === undefined || !same(current.info, service.info)) return
|
||||
yield* signal(service.info.pid, "SIGTERM")
|
||||
return yield* Effect.fail(new Error("Background service does not support authenticated stop requests"))
|
||||
}
|
||||
const done = yield* stopped(service.info.pid).pipe(Effect.retry(poll(timing)), Effect.option)
|
||||
if (Option.isSome(done)) return
|
||||
|
||||
const latest = yield* find(options)
|
||||
if (latest === undefined || !same(latest.info, service.info)) return
|
||||
yield* signal(service.info.pid, "SIGKILL")
|
||||
yield* stopped(service.info.pid).pipe(Effect.retry(poll(timing)))
|
||||
return yield* Effect.fail(new Error("Background service accepted the stop request but did not exit"))
|
||||
})
|
||||
|
||||
const decodeStopResponse = Schema.decodeUnknownOption(ServiceStatus.StopResponse)
|
||||
@@ -316,7 +277,8 @@ const requestStop = Effect.fnUntraced(function* (service: LocalService, timeout
|
||||
signal: AbortSignal.timeout(timeout),
|
||||
}),
|
||||
).pipe(Effect.option, Effect.map(Option.getOrUndefined))
|
||||
if (response === undefined || response.status === 404 || response.status === 405) return "unsupported" as const
|
||||
if (response === undefined) return "rejected" as const
|
||||
if (response.status === 404 || response.status === 405) return "unsupported" as const
|
||||
const body = yield* Effect.tryPromise(() => response.json()).pipe(Effect.option, Effect.map(Option.getOrUndefined))
|
||||
const decoded = decodeStopResponse(body)
|
||||
if (!response.ok || Option.isNone(decoded) || !decoded.value.accepted) return "rejected" as const
|
||||
@@ -324,4 +286,4 @@ const requestStop = Effect.fnUntraced(function* (service: LocalService, timeout
|
||||
})
|
||||
|
||||
/** Effect-based local service lifecycle operations. */
|
||||
export const Service = { discover, incumbent, ensure, stop, headers, Info }
|
||||
export const Service = { discover, incumbent, ensure, stop, headers, canReplaceVersion, Info }
|
||||
|
||||
@@ -1,7 +1,15 @@
|
||||
import { readFile } from "node:fs/promises"
|
||||
import { homedir } from "node:os"
|
||||
import { join } from "node:path"
|
||||
import type { DiscoverOptions, Endpoint, Info, EnsureOptions, StopOptions } from "../service.js"
|
||||
import {
|
||||
canReplaceVersion,
|
||||
VersionMismatchError,
|
||||
type DiscoverOptions,
|
||||
type Endpoint,
|
||||
type Info,
|
||||
type EnsureOptions,
|
||||
type StopOptions,
|
||||
} from "../service.js"
|
||||
import {
|
||||
contenderFailure,
|
||||
contenderFinished,
|
||||
@@ -9,7 +17,7 @@ import {
|
||||
spawnServiceContender,
|
||||
} from "../service-contender.js"
|
||||
import { defaultEnsureTiming, ensureTiming, type EnsureTiming } from "../service-timing.js"
|
||||
import type { ServiceHealth, ServiceStopResponse } from "./generated/types.js"
|
||||
import type { ServiceStopResponse } from "./generated/types.js"
|
||||
|
||||
export * from "../service.js"
|
||||
|
||||
@@ -36,7 +44,6 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
|
||||
const timing = ensureTiming(options)
|
||||
const deadline = Date.now() + timing.promiseTimeout
|
||||
const contenders = new Set<ServiceContender>()
|
||||
let timeouts: { readonly info: Info; readonly count: number } | undefined
|
||||
let announced = false
|
||||
let lastSpawn = 0
|
||||
let spawnDelay = timing.spawnDelay
|
||||
@@ -60,19 +67,6 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
|
||||
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
|
||||
@@ -80,8 +74,10 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
|
||||
if (compatible && service.state === "ready") return service.endpoint
|
||||
if (compatible && service.state === "failed") throw new Error("Background service failed to start")
|
||||
if (!compatible) {
|
||||
if (options.canReplace?.(service.version) !== true)
|
||||
throw new VersionMismatchError(options.version, service.version)
|
||||
await kill(service, timing)
|
||||
announce("version-mismatch", service.version)
|
||||
await kill(service, options, timing).catch(() => undefined)
|
||||
lastSpawn = 0
|
||||
}
|
||||
} else {
|
||||
@@ -110,7 +106,9 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
|
||||
/** Stop the registered local service. */
|
||||
export async function stop(options: StopOptions = {}) {
|
||||
const existing = await find(options)
|
||||
if (existing !== undefined) await kill(existing, options, defaultEnsureTiming)
|
||||
if (existing !== undefined) await kill(existing, defaultEnsureTiming)
|
||||
if (existing === undefined && (await read(options.file)) !== undefined)
|
||||
throw new Error("Background service is not responding; stop its process manually and try again")
|
||||
}
|
||||
|
||||
function fallback() {
|
||||
@@ -129,7 +127,15 @@ async function read(file?: string) {
|
||||
const text = await readFile(file ?? fallback(), "utf8").catch(() => undefined)
|
||||
if (text === undefined) return undefined
|
||||
try {
|
||||
return JSON.parse(text) as Info
|
||||
const value: unknown = JSON.parse(text)
|
||||
if (typeof value !== "object" || value === null) return undefined
|
||||
if (!("url" in value) || typeof value.url !== "string") return undefined
|
||||
if (!("pid" in value) || !Number.isInteger(value.pid) || typeof value.pid !== "number" || value.pid <= 0)
|
||||
return undefined
|
||||
if ("id" in value && value.id !== undefined && typeof value.id !== "string") return undefined
|
||||
if ("version" in value && value.version !== undefined && typeof value.version !== "string") return undefined
|
||||
if ("password" in value && value.password !== undefined && typeof value.password !== "string") return undefined
|
||||
return value as Info
|
||||
} catch {
|
||||
return undefined
|
||||
}
|
||||
@@ -143,10 +149,6 @@ type LocalService = {
|
||||
readonly legacy: boolean
|
||||
}
|
||||
|
||||
async function probe(info: Info, allowLegacy = false): Promise<LocalService | undefined> {
|
||||
return (await probeResult(info, allowLegacy)).service
|
||||
}
|
||||
|
||||
async function probeResult(info: Info, allowLegacy = false, timeout = defaultEnsureTiming.requestTimeout) {
|
||||
const endpoint = {
|
||||
url: info.url,
|
||||
@@ -162,52 +164,60 @@ async function probeResult(info: Info, allowLegacy = false, timeout = defaultEns
|
||||
})
|
||||
.then(async (response) => ({
|
||||
response,
|
||||
body: (await response.json()) as ServiceHealth | { readonly healthy: true },
|
||||
body: (await response.json()) as unknown,
|
||||
}))
|
||||
.then(
|
||||
(value) => ({ value }),
|
||||
(cause: unknown) => ({ cause }),
|
||||
)
|
||||
if ("cause" in result) return { service: undefined, timedOut: signal.aborted }
|
||||
if ("cause" in result) return undefined
|
||||
const response = result.value.response
|
||||
const body = result.value.body
|
||||
if (body !== undefined && "version" in body && "pid" in body) {
|
||||
if (body.pid !== info.pid) return { service: undefined, timedOut: false }
|
||||
if (info.version !== undefined && body.version !== info.version) return { service: undefined, timedOut: false }
|
||||
if (
|
||||
typeof body === "object" &&
|
||||
body !== null &&
|
||||
"healthy" in body &&
|
||||
body.healthy === true &&
|
||||
"version" in body &&
|
||||
typeof body.version === "string" &&
|
||||
"pid" in body &&
|
||||
typeof body.pid === "number" &&
|
||||
Number.isInteger(body.pid) &&
|
||||
body.pid > 0
|
||||
) {
|
||||
if (body.pid !== info.pid) return undefined
|
||||
if (info.version !== undefined && body.version !== info.version) return undefined
|
||||
return {
|
||||
service: {
|
||||
info,
|
||||
endpoint,
|
||||
version: body.version,
|
||||
state: response.ok ? "ready" : response.status === 500 ? "failed" : "waiting",
|
||||
legacy: false,
|
||||
} satisfies LocalService,
|
||||
timedOut: false,
|
||||
}
|
||||
}
|
||||
if (!allowLegacy || body?.healthy !== true) return { service: undefined, timedOut: false }
|
||||
return {
|
||||
service: { info, endpoint, state: "ready", legacy: true } satisfies LocalService,
|
||||
timedOut: false,
|
||||
info,
|
||||
endpoint,
|
||||
version: body.version,
|
||||
state: response.ok ? "ready" : response.status === 500 ? "failed" : "waiting",
|
||||
legacy: false,
|
||||
} satisfies LocalService
|
||||
}
|
||||
if (
|
||||
!allowLegacy ||
|
||||
typeof body !== "object" ||
|
||||
body === null ||
|
||||
!("healthy" in body) ||
|
||||
body.healthy !== true ||
|
||||
"version" in body ||
|
||||
"pid" in body
|
||||
)
|
||||
return undefined
|
||||
return { info, endpoint, state: "ready", legacy: true } satisfies LocalService
|
||||
}
|
||||
|
||||
async function registered(file?: string, allowLegacy = false, timeout?: number) {
|
||||
const info = await read(file)
|
||||
if (info === undefined) return { info: undefined, service: undefined, timedOut: false }
|
||||
return { info, ...(await probeResult(info, allowLegacy, timeout)) }
|
||||
if (info === undefined) return { info: undefined, service: undefined }
|
||||
return { info, service: await probeResult(info, allowLegacy, timeout) }
|
||||
}
|
||||
|
||||
async function find(options: { readonly file?: string }) {
|
||||
return (await registered(options.file, true)).service
|
||||
}
|
||||
|
||||
function signal(pid: number, name: NodeJS.Signals) {
|
||||
try {
|
||||
process.kill(pid, name)
|
||||
} catch {}
|
||||
}
|
||||
|
||||
function stopped(pid: number) {
|
||||
try {
|
||||
process.kill(pid, 0)
|
||||
@@ -225,37 +235,12 @@ async function waitUntilStopped(pid: number, timing: EnsureTiming) {
|
||||
return false
|
||||
}
|
||||
|
||||
function same(left: Info, right: Info) {
|
||||
return left.id === right.id && left.version === right.version && left.url === right.url && left.pid === right.pid
|
||||
}
|
||||
|
||||
async function evict(info: Info, options: { readonly file?: string }, timing: EnsureTiming) {
|
||||
const current = await read(options.file)
|
||||
if (current === undefined || !same(current, info)) return
|
||||
signal(info.pid, "SIGTERM")
|
||||
if (await waitUntilStopped(info.pid, timing)) return
|
||||
|
||||
const latest = await read(options.file)
|
||||
if (latest === undefined || !same(latest, info)) return
|
||||
signal(info.pid, "SIGKILL")
|
||||
if (!(await waitUntilStopped(info.pid, timing))) throw new Error(`Server process ${info.pid} is still running`)
|
||||
}
|
||||
|
||||
async function kill(service: LocalService, options: { readonly file?: string }, timing: EnsureTiming) {
|
||||
async function kill(service: LocalService, timing: EnsureTiming) {
|
||||
const requested = await requestStop(service, timing.requestTimeout)
|
||||
if (requested === "rejected") return
|
||||
if (requested === "unsupported") {
|
||||
const current = await find(options)
|
||||
if (current === undefined || !same(current.info, service.info)) return
|
||||
signal(service.info.pid, "SIGTERM")
|
||||
}
|
||||
if (requested === "rejected") throw new Error("Background service rejected the stop request")
|
||||
if (requested === "unsupported") throw new Error("Background service does not support authenticated stop requests")
|
||||
if (await waitUntilStopped(service.info.pid, timing)) return
|
||||
|
||||
const latest = await find(options)
|
||||
if (latest === undefined || !same(latest.info, service.info)) return
|
||||
signal(service.info.pid, "SIGKILL")
|
||||
if (!(await waitUntilStopped(service.info.pid, timing)))
|
||||
throw new Error(`Server process ${service.info.pid} is still running`)
|
||||
throw new Error("Background service accepted the stop request but did not exit")
|
||||
}
|
||||
|
||||
async function requestStop(service: LocalService, timeout = defaultEnsureTiming.requestTimeout) {
|
||||
@@ -266,7 +251,8 @@ async function requestStop(service: LocalService, timeout = defaultEnsureTiming.
|
||||
body: JSON.stringify({ instanceID: service.info.id }),
|
||||
signal: AbortSignal.timeout(timeout),
|
||||
}).catch(() => undefined)
|
||||
if (response === undefined || response.status === 404 || response.status === 405) return "unsupported" as const
|
||||
if (response === undefined) return "rejected" as const
|
||||
if (response.status === 404 || response.status === 405) return "unsupported" as const
|
||||
const body = (await response.json().catch(() => undefined)) as ServiceStopResponse | undefined
|
||||
if (!response.ok || body?.accepted !== true) return "rejected" as const
|
||||
return "accepted" as const
|
||||
@@ -277,4 +263,4 @@ function delay(milliseconds: number) {
|
||||
}
|
||||
|
||||
/** Promise-based local service lifecycle operations. */
|
||||
export const Service = { discover, ensure, stop, headers }
|
||||
export const Service = { discover, ensure, stop, headers, canReplaceVersion }
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import semver from "semver"
|
||||
|
||||
/** Connection details for a local OpenCode service. */
|
||||
export type Endpoint = {
|
||||
/** Base URL of the service. */
|
||||
@@ -28,10 +30,34 @@ export type EnsureReason = "missing" | "version-mismatch"
|
||||
export type EnsureOptions = DiscoverOptions & {
|
||||
/** Service command and arguments. Defaults to `opencode serve --service`. */
|
||||
readonly command?: ReadonlyArray<string>
|
||||
/** Decide whether a version-mismatched service may be replaced. Defaults to false. */
|
||||
readonly canReplace?: (version: string | undefined) => boolean
|
||||
/** Called once before spawning a new service process. */
|
||||
readonly onStart?: (reason: EnsureReason, previousVersion?: string) => void
|
||||
}
|
||||
|
||||
/** A healthy service exists, but the caller's replacement policy protects it. */
|
||||
export class VersionMismatchError extends Error {
|
||||
override readonly name = "VersionMismatchError"
|
||||
|
||||
constructor(
|
||||
readonly clientVersion: string | undefined,
|
||||
readonly serverVersion: string | undefined,
|
||||
) {
|
||||
super(`Background service ${serverVersion ?? "unknown"} does not match client ${clientVersion ?? "unknown"}`)
|
||||
}
|
||||
}
|
||||
|
||||
/** Whether a client version is strictly newer than a service version. */
|
||||
export function canReplaceVersion(serverVersion: string | undefined, clientVersion: string) {
|
||||
if (serverVersion === undefined) return false
|
||||
// Compare preview build numbers numerically rather than as semver prerelease strings.
|
||||
const server = serverVersion.replace(/-(\d+)(?=(?:\.\d+)?$)/, ".$1")
|
||||
const client = clientVersion.replace(/-(\d+)(?=(?:\.\d+)?$)/, ".$1")
|
||||
if (!semver.valid(server) || !semver.valid(client)) return false
|
||||
return semver.lt(server, client)
|
||||
}
|
||||
|
||||
/** Options used to stop the local OpenCode service. */
|
||||
export type StopOptions = {
|
||||
/** Absolute registration file path. Defaults to the XDG state directory. */
|
||||
|
||||
@@ -27,7 +27,7 @@ if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated" ||
|
||||
}
|
||||
|
||||
let requests = 0
|
||||
const version = mode === "old" || mode === "reject-stop" ? "old" : "test"
|
||||
const version = mode === "old" || mode === "reject-stop" || mode === "stop-hanging" ? "old" : "test"
|
||||
const id = crypto.randomUUID()
|
||||
const server = Bun.serve({
|
||||
port: 0,
|
||||
@@ -37,7 +37,11 @@ const server = Bun.serve({
|
||||
await appendFile(registration + ".stop-attempts", process.pid + "\n")
|
||||
return Response.json({ accepted: false })
|
||||
}
|
||||
if (pathname === "/api/service/stop" && mode === "graceful") {
|
||||
if (pathname === "/api/service/stop" && mode === "stop-hanging") {
|
||||
await appendFile(registration + ".stop-attempts", process.pid + "\n")
|
||||
return new Promise<Response>(() => {})
|
||||
}
|
||||
if (pathname === "/api/service/stop" && (mode === "graceful" || mode === "old")) {
|
||||
const body = await request.json()
|
||||
if (typeof body !== "object" || body === null || body.instanceID !== id) return Response.json({ accepted: false })
|
||||
await writeFile(registration + ".stop", JSON.stringify(body))
|
||||
@@ -60,7 +64,7 @@ const server = Bun.serve({
|
||||
if (mode === "starting" && !(await Bun.file(registration + ".release").exists()))
|
||||
return Response.json({ healthy: true, version, pid: process.pid }, { status: 503 })
|
||||
if (mode === "failed-owner") return Response.json({ healthy: true, version, pid: process.pid }, { status: 500 })
|
||||
if (mode === "starting" || mode === "graceful" || mode === "reject-stop")
|
||||
if (mode === "starting" || mode === "graceful" || mode === "reject-stop" || mode === "stop-hanging")
|
||||
return Response.json({ healthy: true, version, pid: process.pid })
|
||||
return Response.json({ healthy: true, version, pid: process.pid })
|
||||
},
|
||||
|
||||
@@ -25,6 +25,50 @@ test("discovers a registered service", async () => {
|
||||
expect(await Service.discover({ file: registration, version: "other" })).toBeUndefined()
|
||||
})
|
||||
|
||||
test("rejects malformed registrations without probing or signaling", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const malformed = [
|
||||
null,
|
||||
[],
|
||||
{},
|
||||
{ url: "http://127.0.0.1:1" },
|
||||
{ url: "http://127.0.0.1:1", pid: 0 },
|
||||
{ url: "http://127.0.0.1:1", pid: -1 },
|
||||
{ url: "http://127.0.0.1:1", pid: 1.5 },
|
||||
{ url: "http://127.0.0.1:1", pid: "1" },
|
||||
{ url: "http://127.0.0.1:1", pid: 1, id: 1 },
|
||||
]
|
||||
|
||||
for (const value of malformed) {
|
||||
await Bun.write(registration, JSON.stringify(value))
|
||||
expect(await Service.discover({ file: registration })).toBeUndefined()
|
||||
}
|
||||
})
|
||||
|
||||
test("rejects primitive and partial modern health responses", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const bodies = [
|
||||
null,
|
||||
1,
|
||||
"healthy",
|
||||
[],
|
||||
{},
|
||||
{ healthy: false, version: "test", pid: process.pid },
|
||||
{ healthy: true, version: null, pid: process.pid },
|
||||
{ healthy: true, version: "test", pid: "1" },
|
||||
{ healthy: true, version: "test" },
|
||||
{ healthy: true, pid: process.pid },
|
||||
]
|
||||
|
||||
for (const body of bodies) {
|
||||
using server = Bun.serve({ port: 0, fetch: () => Response.json(body) })
|
||||
await Bun.write(registration, JSON.stringify({ url: server.url.toString(), pid: process.pid }))
|
||||
expect(await Service.discover({ file: registration })).toBeUndefined()
|
||||
}
|
||||
})
|
||||
|
||||
test("ensures a missing service with native promises", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
@@ -87,7 +131,7 @@ test("reports a bounded contender stderr tail with native promises", async () =>
|
||||
expect(error.message.length).toBeLessThan(9_000)
|
||||
}, 10_000)
|
||||
|
||||
test("evicts an unresponsive registered service before starting its replacement", async () => {
|
||||
test("never evicts an unresponsive registered service automatically", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const existing = Bun.spawn([process.execPath, fixture, registration, "hanging"], {
|
||||
@@ -98,19 +142,45 @@ test("evicts an unresponsive registered service before starting its replacement"
|
||||
await waitForFile(registration)
|
||||
const original = await Bun.file(registration).json()
|
||||
|
||||
const endpoint = await ensure({
|
||||
const options = {
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "delayed", "10"],
|
||||
})
|
||||
const replacement = await Bun.file(registration).json()
|
||||
command: [process.execPath, fixture, registration, "record-start"],
|
||||
}
|
||||
const result = ensure(options)
|
||||
await waitForLines(registration + ".requests", 3)
|
||||
|
||||
expect((await Bun.file(registration + ".requests").text()).trim().split("\n")).toHaveLength(3)
|
||||
expect(await existing.exited).toBe(0)
|
||||
expect(replacement.pid).not.toBe(original.pid)
|
||||
expect(endpoint.url).toBe(replacement.url)
|
||||
process.kill(replacement.pid, "SIGTERM")
|
||||
await waitForExit(replacement.pid)
|
||||
expect(existing.exitCode).toBe(null)
|
||||
expect(await Bun.file(registration).json()).toEqual(original)
|
||||
await expect(result).rejects.toThrow()
|
||||
})
|
||||
|
||||
test("explicit native stop refuses to signal an unidentified unresponsive PID", async () => {
|
||||
const registration = await setup("hanging")
|
||||
const info = await Bun.file(registration).json()
|
||||
|
||||
await expect(Service.stop({ file: registration })).rejects.toThrow("stop its process manually")
|
||||
|
||||
expect(process.kill(info.pid, 0)).toBe(true)
|
||||
})
|
||||
|
||||
test("a stale native client refuses to replace a newer service", async () => {
|
||||
const registration = await setup("graceful")
|
||||
const directory = await temp()
|
||||
const contender = join(directory, "contender.json")
|
||||
const info = await Bun.file(registration).json()
|
||||
|
||||
await expect(
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "old",
|
||||
command: [process.execPath, fixture, contender, "record-start"],
|
||||
}),
|
||||
).rejects.toThrow("Background service test does not match client old")
|
||||
|
||||
expect(await Bun.file(contender + ".started").exists()).toBe(false)
|
||||
expect(process.kill(info.pid, 0)).toBe(true)
|
||||
expect(await Bun.file(registration).json()).toEqual(info)
|
||||
})
|
||||
|
||||
test("requests graceful stop of the exact service instance", async () => {
|
||||
@@ -143,3 +213,14 @@ async function waitForFile(file: string) {
|
||||
}
|
||||
throw new Error(`Timed out waiting for ${file}`)
|
||||
}
|
||||
|
||||
async function waitForLines(file: string, count: number) {
|
||||
for (let attempt = 0; attempt < 600; attempt++) {
|
||||
const text = await Bun.file(file)
|
||||
.text()
|
||||
.catch(() => "")
|
||||
if (text.trim().split("\n").length >= count) return
|
||||
await Bun.sleep(5)
|
||||
}
|
||||
throw new Error(`Timed out waiting for ${count} lines in ${file}`)
|
||||
}
|
||||
|
||||
@@ -72,29 +72,39 @@ test("reports a failed registered service without spawning", async () => {
|
||||
expect(process.exitCode).toBe(null)
|
||||
})
|
||||
|
||||
test("evicts an unresponsive registered service before starting its replacement", async () => {
|
||||
test("never evicts an unresponsive registered service automatically", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const existing = spawn(registration, "hanging")
|
||||
await waitForFile(registration)
|
||||
const original = await Bun.file(registration).json()
|
||||
|
||||
const endpoint = await run(
|
||||
const controller = new AbortController()
|
||||
const result = Effect.runPromise(
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "delayed", "10"],
|
||||
}),
|
||||
command: [process.execPath, fixture, registration, "record-start"],
|
||||
}).pipe(Effect.provide(NodeFileSystem.layer)),
|
||||
{ signal: controller.signal },
|
||||
)
|
||||
const replacement = await Bun.file(registration).json()
|
||||
await waitForLines(registration + ".requests", 3)
|
||||
controller.abort()
|
||||
await result.catch(() => undefined)
|
||||
|
||||
expect((await Bun.file(registration + ".requests").text()).trim().split("\n")).toHaveLength(3)
|
||||
expect(await existing.exited).toBe(0)
|
||||
expect(replacement.pid).not.toBe(original.pid)
|
||||
expect(endpoint.url).toBe(replacement.url)
|
||||
expect(await health(endpoint.url)).toEqual({ healthy: true, version: "test", pid: replacement.pid })
|
||||
process.kill(replacement.pid, "SIGTERM")
|
||||
await waitForExit(replacement.pid)
|
||||
expect(existing.exitCode).toBe(null)
|
||||
expect(await Bun.file(registration).json()).toEqual(original)
|
||||
})
|
||||
|
||||
test("explicit stop refuses to signal an unidentified unresponsive PID", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const existing = spawn(registration, "hanging")
|
||||
await waitForFile(registration)
|
||||
|
||||
await expect(run(Service.stop({ file: registration }))).rejects.toThrow("stop its process manually")
|
||||
|
||||
expect(existing.exitCode).toBe(null)
|
||||
})
|
||||
|
||||
test("requests graceful stop of the exact service instance", async () => {
|
||||
@@ -115,36 +125,124 @@ test("does not spawn contenders while an incompatible service rejects replacemen
|
||||
const contender = join(directory, "contender.json")
|
||||
const existing = spawn(registration, "reject-stop")
|
||||
await waitForFile(registration)
|
||||
const controller = new AbortController()
|
||||
const starting = Effect.runPromise(
|
||||
const starting = run(
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
canReplace: () => true,
|
||||
command: [process.execPath, fixture, contender, "record-start"],
|
||||
}).pipe(Effect.provide(NodeFileSystem.layer)),
|
||||
{ signal: controller.signal },
|
||||
}),
|
||||
)
|
||||
|
||||
await waitForLines(registration + ".stop-attempts", 2)
|
||||
controller.abort()
|
||||
await starting.catch(() => undefined)
|
||||
await expect(starting).rejects.toThrow("Background service rejected the stop request")
|
||||
|
||||
expect(await Bun.file(contender + ".started").exists()).toBe(false)
|
||||
expect((await Bun.file(registration + ".stop-attempts").text()).trim().split("\n")).toHaveLength(1)
|
||||
expect(existing.exitCode).toBe(null)
|
||||
})
|
||||
|
||||
test("a legacy health response is still replaced", async () => {
|
||||
test("does not signal a modern service when its stop request times out", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const contender = join(directory, "contender.json")
|
||||
const existing = spawn(registration, "stop-hanging")
|
||||
await waitForFile(registration)
|
||||
const starting = run(
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
canReplace: () => true,
|
||||
command: [process.execPath, fixture, contender, "record-start"],
|
||||
}),
|
||||
)
|
||||
|
||||
await expect(starting).rejects.toThrow("Background service rejected the stop request")
|
||||
|
||||
expect(await Bun.file(contender + ".started").exists()).toBe(false)
|
||||
expect((await Bun.file(registration + ".stop-attempts").text()).trim().split("\n")).toHaveLength(1)
|
||||
expect(existing.exitCode).toBe(null)
|
||||
})
|
||||
|
||||
test("explicit stop refuses to signal when a modern stop request times out", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const existing = spawn(registration, "stop-hanging")
|
||||
await waitForFile(registration)
|
||||
|
||||
await expect(run(Service.stop({ file: registration }))).rejects.toThrow(
|
||||
"Background service rejected the stop request",
|
||||
)
|
||||
|
||||
expect(existing.exitCode).toBe(null)
|
||||
})
|
||||
|
||||
test("a stale client refuses to replace a newer service", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const contender = join(directory, "contender.json")
|
||||
const existing = spawn(registration, "graceful")
|
||||
await waitForFile(registration)
|
||||
const info = await Bun.file(registration).json()
|
||||
|
||||
await expect(
|
||||
run(
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "old",
|
||||
command: [process.execPath, fixture, contender, "record-start"],
|
||||
}),
|
||||
),
|
||||
).rejects.toThrow("Background service test does not match client old")
|
||||
|
||||
expect(await Bun.file(contender + ".started").exists()).toBe(false)
|
||||
expect(existing.exitCode).toBe(null)
|
||||
expect(await Bun.file(registration).json()).toEqual(info)
|
||||
})
|
||||
|
||||
test("explicit restart can activate an installed downgrade", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const current = spawn(registration, "graceful")
|
||||
await waitForFile(registration)
|
||||
const before = await Bun.file(registration).json()
|
||||
const options = {
|
||||
file: registration,
|
||||
version: "old",
|
||||
canReplace: () => false,
|
||||
command: [process.execPath, fixture, registration, "old"],
|
||||
}
|
||||
|
||||
await expect(run(ensure(options))).rejects.toThrow("Background service test does not match client old")
|
||||
expect(current.exitCode).toBe(null)
|
||||
|
||||
await run(Service.stop({ file: registration }))
|
||||
const endpoint = await run(ensure(options))
|
||||
const after = await Bun.file(registration).json()
|
||||
|
||||
try {
|
||||
expect(after.pid).not.toBe(before.pid)
|
||||
expect(after.version).toBe("old")
|
||||
expect(endpoint.url).toBe(after.url)
|
||||
} finally {
|
||||
process.kill(after.pid, "SIGTERM")
|
||||
await waitForExit(after.pid)
|
||||
}
|
||||
})
|
||||
|
||||
test("refuses to signal a legacy service without authenticated stop", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const existing = spawn(registration, "legacy")
|
||||
await waitForFile(registration)
|
||||
|
||||
const starts: EnsureReason[] = []
|
||||
const result = run(ensure({ file: registration, command: [], onStart: (reason) => starts.push(reason) }))
|
||||
const result = run(
|
||||
ensure({ file: registration, command: [], canReplace: () => true, onStart: (reason) => starts.push(reason) }),
|
||||
)
|
||||
|
||||
await expect(result).rejects.toThrow("Missing service command")
|
||||
expect(starts).toEqual(["version-mismatch"])
|
||||
await existing.exited
|
||||
await expect(result).rejects.toThrow("does not support authenticated stop requests")
|
||||
expect(starts).toEqual([])
|
||||
expect(existing.exitCode).toBe(null)
|
||||
})
|
||||
|
||||
test("waits for a slow winner while bounding lock probes", async () => {
|
||||
@@ -253,6 +351,7 @@ test("replaces an incompatible owner that appears during startup", async () => {
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
canReplace: () => true,
|
||||
command: [process.execPath, fixture, registration, "delayed", "500"],
|
||||
}),
|
||||
)
|
||||
@@ -271,6 +370,38 @@ test("replaces an incompatible owner that appears during startup", async () => {
|
||||
}
|
||||
})
|
||||
|
||||
test("concurrent current-version launchers converge on one replacement", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const old = spawn(registration, "old")
|
||||
await waitForFile(registration)
|
||||
|
||||
const endpoints = await Promise.all(
|
||||
Array.from({ length: 20 }, (_, index) => {
|
||||
const options = {
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "coordinated"],
|
||||
canReplace: (version: string | undefined) => version === "old",
|
||||
}
|
||||
return index % 2 === 0
|
||||
? run(ensure(options))
|
||||
: import("../src/promise/service").then((mod) => mod.Service.ensure(options))
|
||||
}),
|
||||
)
|
||||
const info = await Bun.file(registration).json()
|
||||
|
||||
try {
|
||||
expect(new Set(endpoints.map((endpoint) => endpoint.url))).toEqual(new Set([info.url]))
|
||||
expect(info.version).toBe("test")
|
||||
expect(old.exitCode).not.toBe(null)
|
||||
expect(await health(info.url)).toEqual({ healthy: true, version: "test", pid: info.pid })
|
||||
} finally {
|
||||
process.kill(info.pid, "SIGTERM")
|
||||
await waitForExit(info.pid)
|
||||
}
|
||||
})
|
||||
|
||||
function run<A, E>(effect: Effect.Effect<A, E>) {
|
||||
return Effect.runPromise(effect.pipe(Effect.provide(NodeFileSystem.layer)))
|
||||
}
|
||||
|
||||
@@ -26,6 +26,7 @@ export async function startBackgroundCli(logger: Logger) {
|
||||
if (isolated) process.env.XDG_STATE_HOME = app.getPath("userData")
|
||||
const service = await Service.ensure({
|
||||
version,
|
||||
canReplace: (serverVersion) => Service.canReplaceVersion(serverVersion, version),
|
||||
command: [binary, "serve", "--service"],
|
||||
onStart: (reason, previousVersion) => logger.log("v2 CLI background service starting", { reason, previousVersion }),
|
||||
})
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { ClientError } from "@opencode-ai/client"
|
||||
import { createSignal, onCleanup, onMount, Show } from "solid-js"
|
||||
import { useClient } from "../context/client"
|
||||
import { useTheme } from "../context/theme"
|
||||
@@ -16,9 +17,23 @@ export function MigrationOverlay() {
|
||||
|
||||
onMount(async () => {
|
||||
await Bun.sleep(1_000)
|
||||
if (abort.signal.aborted) return
|
||||
void (async () => {
|
||||
while (true) {
|
||||
const status = await client.api.migration.v1.status({ signal: abort.signal })
|
||||
const result = await client.api.migration.v1.status({ signal: abort.signal }).then(
|
||||
(status) => ({ status }),
|
||||
(error: unknown) => ({ error }),
|
||||
)
|
||||
if ("error" in result) {
|
||||
if (result.error instanceof ClientError && result.error.reason === "Transport") {
|
||||
if (abort.signal.aborted) return
|
||||
await Bun.sleep(1_000)
|
||||
if (abort.signal.aborted) return
|
||||
continue
|
||||
}
|
||||
throw result.error
|
||||
}
|
||||
const status = result.status
|
||||
setProgress(status.status === "running" ? status.progress : undefined)
|
||||
if (status.status === "completed") return
|
||||
if (status.status === "error") throw new Error(status.error)
|
||||
|
||||
Reference in New Issue
Block a user