fix(client): preserve compatible background service (#36583)
This commit is contained in:
parent
c0ed0106b1
commit
397993e898
4 changed files with 142 additions and 6 deletions
5
.changeset/calm-services-start.md
Normal file
5
.changeset/calm-services-start.md
Normal file
|
|
@ -0,0 +1,5 @@
|
||||||
|
---
|
||||||
|
"@opencode-ai/client": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
Reuse a same-version background service when a repeated health probe succeeds instead of replacing an endpoint another client may already be using.
|
||||||
|
|
@ -59,11 +59,11 @@ const discoverLocal = Effect.fnUntraced(function* (options: Options) {
|
||||||
export const start = Effect.fn("service.start")(function* (options: StartOptions = {}) {
|
export const start = Effect.fn("service.start")(function* (options: StartOptions = {}) {
|
||||||
const compatible = yield* discover(options)
|
const compatible = yield* discover(options)
|
||||||
if (compatible !== undefined) return compatible
|
if (compatible !== undefined) return compatible
|
||||||
const mismatched = yield* find(options)
|
const existing = yield* find(options)
|
||||||
yield* Effect.sync(() =>
|
if (existing?.version !== undefined && (options.version === undefined || existing.version === options.version))
|
||||||
options.onStart?.(mismatched === undefined ? "missing" : "version-mismatch", mismatched?.info),
|
return existing.endpoint
|
||||||
)
|
yield* Effect.sync(() => options.onStart?.(existing === undefined ? "missing" : "version-mismatch", existing?.info))
|
||||||
if (mismatched !== undefined) yield* kill(mismatched.info, options).pipe(Effect.ignore)
|
if (existing !== undefined) yield* kill(existing.info, options).pipe(Effect.ignore)
|
||||||
|
|
||||||
const [command, ...args] = options.command ?? ["opencode", "serve", "--service"]
|
const [command, ...args] = options.command ?? ["opencode", "serve", "--service"]
|
||||||
if (command === undefined) return yield* Effect.fail(new Error("Missing service command"))
|
if (command === undefined) return yield* Effect.fail(new Error("Missing service command"))
|
||||||
|
|
@ -138,6 +138,7 @@ const read = Effect.fnUntraced(function* (file?: string) {
|
||||||
type LocalService = {
|
type LocalService = {
|
||||||
readonly info: Info
|
readonly info: Info
|
||||||
readonly endpoint: Endpoint
|
readonly endpoint: Endpoint
|
||||||
|
readonly version?: string
|
||||||
}
|
}
|
||||||
|
|
||||||
const probe = Effect.fnUntraced(function* (info: Info, version?: string, allowLegacy = false) {
|
const probe = Effect.fnUntraced(function* (info: Info, version?: string, allowLegacy = false) {
|
||||||
|
|
@ -161,7 +162,7 @@ const probe = Effect.fnUntraced(function* (info: Info, version?: string, allowLe
|
||||||
if (health.value.pid !== info.pid) return undefined
|
if (health.value.pid !== info.pid) return undefined
|
||||||
if (info.version !== undefined && health.value.version !== info.version) return undefined
|
if (info.version !== undefined && health.value.version !== info.version) return undefined
|
||||||
if (version !== undefined && health.value.version !== version) return undefined
|
if (version !== undefined && health.value.version !== version) return undefined
|
||||||
return { info, endpoint } satisfies LocalService
|
return { info, endpoint, version: health.value.version } satisfies LocalService
|
||||||
}
|
}
|
||||||
if (
|
if (
|
||||||
!allowLegacy ||
|
!allowLegacy ||
|
||||||
|
|
|
||||||
39
packages/client/test/fixture/service.ts
Normal file
39
packages/client/test/fixture/service.ts
Normal file
|
|
@ -0,0 +1,39 @@
|
||||||
|
import { rename, writeFile } from "node:fs/promises"
|
||||||
|
|
||||||
|
const [registration, mode] = process.argv.slice(2)
|
||||||
|
if (registration === undefined || mode === undefined) throw new Error("Missing service fixture arguments")
|
||||||
|
|
||||||
|
let requests = 0
|
||||||
|
const server = Bun.serve({
|
||||||
|
port: 0,
|
||||||
|
async fetch(request) {
|
||||||
|
if (new URL(request.url).pathname !== "/api/health") return new Response(null, { status: 404 })
|
||||||
|
requests += 1
|
||||||
|
if (mode === "modern" && requests === 1) {
|
||||||
|
await writeFile(registration + ".first-request", "")
|
||||||
|
while (!(await Bun.file(registration + ".release").exists())) await Bun.sleep(5)
|
||||||
|
return new Response(null, { status: 503 })
|
||||||
|
}
|
||||||
|
if (mode === "legacy") return Response.json({ healthy: true })
|
||||||
|
return Response.json({ healthy: true, version: "test", pid: process.pid })
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
await writeFile(
|
||||||
|
registration + ".tmp",
|
||||||
|
JSON.stringify({
|
||||||
|
id: crypto.randomUUID(),
|
||||||
|
version: mode === "legacy" ? undefined : "test",
|
||||||
|
url: server.url.toString(),
|
||||||
|
pid: process.pid,
|
||||||
|
}),
|
||||||
|
{ mode: 0o600 },
|
||||||
|
)
|
||||||
|
await rename(registration + ".tmp", registration)
|
||||||
|
|
||||||
|
const shutdown = () => {
|
||||||
|
server.stop(true)
|
||||||
|
process.exit()
|
||||||
|
}
|
||||||
|
process.on("SIGTERM", shutdown)
|
||||||
|
process.on("SIGINT", shutdown)
|
||||||
91
packages/client/test/service.test.ts
Normal file
91
packages/client/test/service.test.ts
Normal file
|
|
@ -0,0 +1,91 @@
|
||||||
|
import { NodeFileSystem } from "@effect/platform-node"
|
||||||
|
import { afterEach, expect, test } from "bun:test"
|
||||||
|
import { Effect } from "effect"
|
||||||
|
import { mkdtemp, rm, writeFile } from "node:fs/promises"
|
||||||
|
import { tmpdir } from "node:os"
|
||||||
|
import { join } from "node:path"
|
||||||
|
import { Service } from "../src/effect/index"
|
||||||
|
|
||||||
|
const fixture = join(import.meta.dir, "fixture/service.ts")
|
||||||
|
const processes: Bun.Subprocess[] = []
|
||||||
|
const directories: string[] = []
|
||||||
|
|
||||||
|
afterEach(async () => {
|
||||||
|
processes.forEach((process) => process.kill("SIGTERM"))
|
||||||
|
await Promise.all(processes.splice(0).map((process) => process.exited))
|
||||||
|
await Promise.all(directories.splice(0).map((directory) => rm(directory, { recursive: true, force: true })))
|
||||||
|
})
|
||||||
|
|
||||||
|
test("a concurrent same-version start cannot invalidate a resolved endpoint", async () => {
|
||||||
|
const directory = await temp()
|
||||||
|
const registration = join(directory, "service.json")
|
||||||
|
spawn(registration, "modern")
|
||||||
|
await waitForFile(registration)
|
||||||
|
const original = await Bun.file(registration).json()
|
||||||
|
|
||||||
|
const starts: Service.StartReason[] = []
|
||||||
|
const first = run(
|
||||||
|
Service.start({
|
||||||
|
file: registration,
|
||||||
|
version: "test",
|
||||||
|
command: [],
|
||||||
|
onStart: (reason) => starts.push(reason),
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
await waitForFile(registration + ".first-request")
|
||||||
|
|
||||||
|
const resolved = await run(Service.start({ file: registration, version: "test" }))
|
||||||
|
expect(resolved.url).toBe(original.url)
|
||||||
|
|
||||||
|
await writeFile(registration + ".release", "")
|
||||||
|
await first
|
||||||
|
|
||||||
|
expect(starts).toEqual([])
|
||||||
|
expect(await Bun.file(registration).json()).toEqual(original)
|
||||||
|
expect(await health(resolved.url)).toEqual({ healthy: true, version: "test", pid: original.pid })
|
||||||
|
})
|
||||||
|
|
||||||
|
test("a legacy health response is still replaced", async () => {
|
||||||
|
const directory = await temp()
|
||||||
|
const registration = join(directory, "service.json")
|
||||||
|
const existing = spawn(registration, "legacy")
|
||||||
|
await waitForFile(registration)
|
||||||
|
|
||||||
|
const starts: Service.StartReason[] = []
|
||||||
|
const result = run(Service.start({ file: registration, command: [], onStart: (reason) => starts.push(reason) }))
|
||||||
|
|
||||||
|
await expect(result).rejects.toThrow("Missing service command")
|
||||||
|
expect(starts).toEqual(["version-mismatch"])
|
||||||
|
await existing.exited
|
||||||
|
})
|
||||||
|
|
||||||
|
function run<A, E>(effect: Effect.Effect<A, E, never>) {
|
||||||
|
return Effect.runPromise(effect.pipe(Effect.provide(NodeFileSystem.layer)))
|
||||||
|
}
|
||||||
|
|
||||||
|
function spawn(registration: string, mode: string, ...args: string[]) {
|
||||||
|
const subprocess = Bun.spawn([process.execPath, fixture, registration, mode, ...args], {
|
||||||
|
stdout: "ignore",
|
||||||
|
stderr: "inherit",
|
||||||
|
})
|
||||||
|
processes.push(subprocess)
|
||||||
|
return subprocess
|
||||||
|
}
|
||||||
|
|
||||||
|
async function temp() {
|
||||||
|
const directory = await mkdtemp(join(tmpdir(), "opencode-client-service-"))
|
||||||
|
directories.push(directory)
|
||||||
|
return directory
|
||||||
|
}
|
||||||
|
|
||||||
|
async function waitForFile(file: string) {
|
||||||
|
for (let attempt = 0; attempt < 600; attempt++) {
|
||||||
|
if (await Bun.file(file).exists()) return
|
||||||
|
await Bun.sleep(5)
|
||||||
|
}
|
||||||
|
throw new Error(`Timed out waiting for ${file}`)
|
||||||
|
}
|
||||||
|
|
||||||
|
async function health(url: string) {
|
||||||
|
return fetch(new URL("/api/health", url), { signal: AbortSignal.timeout(1_000) }).then((response) => response.json())
|
||||||
|
}
|
||||||
Loading…
Add table
Add a link
Reference in a new issue