mirror of
https://gitee.com/zhangsken/opencode.git
synced 2026-10-16 10:30:01 +08:00
fix(cli): retry managed service binds after temporary conflicts
This commit is contained in:
@@ -1,13 +1,13 @@
|
||||
export * as ServerProcess from "./server-process"
|
||||
|
||||
import { NodeServices } from "@effect/platform-node"
|
||||
import { Service, type DiscoverOptions } from "@opencode/client/effect/service"
|
||||
import { Service } from "@opencode/client/effect/service"
|
||||
import { LayerNode } from "@opencode/util/effect/layer-node"
|
||||
import { Global } from "@opencode/util/global"
|
||||
import { OPENCODE_ARTIFACT, OPENCODE_CHANNEL, OPENCODE_VERSION } from "./version"
|
||||
import { AppProcess } from "@opencode/util/process"
|
||||
import { randomBytes, randomUUID } from "node:crypto"
|
||||
import { Effect, Option, Redacted, Schedule, Schema } from "effect"
|
||||
import { Effect, Option, Redacted, Schema } from "effect"
|
||||
import { PersistentPty } from "@opencode/schema/persistent-pty"
|
||||
import { HttpServer } from "effect/unstable/http"
|
||||
import { Env } from "./env"
|
||||
@@ -84,7 +84,7 @@ const processEffect = Effect.fnUntraced(function* (options: Options) {
|
||||
if (!password) return yield* Effect.fail(new Error("Missing server password"))
|
||||
const instanceID = randomUUID()
|
||||
const transform = yield* WebUi.handler()
|
||||
const server = yield* start(
|
||||
const launch = start(
|
||||
{
|
||||
app: {
|
||||
name: process.env.OPENCODE_CLIENT ?? OPENCODE_ARTIFACT,
|
||||
@@ -140,22 +140,33 @@ const processEffect = Effect.fnUntraced(function* (options: Options) {
|
||||
}),
|
||||
},
|
||||
transform,
|
||||
).pipe(
|
||||
)
|
||||
const server = yield* launch.pipe(
|
||||
Effect.catch((error) => {
|
||||
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 },
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
return Effect.gen(function* () {
|
||||
const deadline = Date.now() + 15_000
|
||||
while (Date.now() < deadline) {
|
||||
const incumbent = yield* Service.incumbent({ ...serviceOptions, url: serviceURL(hostname, port) }).pipe(
|
||||
Effect.timeoutOption(deadline - Date.now()),
|
||||
)
|
||||
if (Option.isSome(incumbent) && incumbent.value !== undefined) return
|
||||
yield* Effect.sleep("100 millis")
|
||||
if (Date.now() >= deadline) break
|
||||
// Failed binds close their scope; a successful bind may take longer than this window to boot.
|
||||
const server = yield* launch.pipe(
|
||||
Effect.catch((error) => (addressInUse(error) ? Effect.void : Effect.fail(error))),
|
||||
)
|
||||
if (server !== undefined) return server
|
||||
}
|
||||
return yield* 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 },
|
||||
),
|
||||
)
|
||||
})
|
||||
}),
|
||||
)
|
||||
if (server === undefined) return
|
||||
@@ -171,15 +182,6 @@ const processEffect = Effect.fnUntraced(function* (options: Options) {
|
||||
)
|
||||
})
|
||||
|
||||
const recognizeIncumbent = Effect.fnUntraced(function* (options: DiscoverOptions, hostname: string, port: number) {
|
||||
const found = yield* Service.incumbent({ ...options, url: serviceURL(hostname, port) }).pipe(
|
||||
Effect.filterOrFail((value) => value !== undefined),
|
||||
Effect.retry(Schedule.spaced("100 millis")),
|
||||
Effect.timeoutOption("15 seconds"),
|
||||
)
|
||||
return Option.isSome(found)
|
||||
})
|
||||
|
||||
function serviceURL(hostname: string, port: number) {
|
||||
return `http://${hostname.includes(":") ? `[${hostname}]` : hostname}:${port}`
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@ import { Service, type Info } from "@opencode/client/effect/service"
|
||||
import { Global } from "@opencode/util/global"
|
||||
import { OPENCODE_VERSION } from "../src/version"
|
||||
import { expect, test } from "bun:test"
|
||||
import { Effect, Schema } from "effect"
|
||||
import { Effect, Schedule, Schema } from "effect"
|
||||
import fs from "node:fs/promises"
|
||||
import os from "node:os"
|
||||
import path from "node:path"
|
||||
@@ -384,6 +384,77 @@ test("unrelated managed port occupancy reports an actionable conflict", async ()
|
||||
}
|
||||
}, 30_000)
|
||||
|
||||
test("the original managed service contender binds when the occupied port is released", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-bind-retry-"))
|
||||
const recognizing = Promise.withResolvers<void>()
|
||||
const requests: string[] = []
|
||||
using listener = Bun.serve({
|
||||
hostname: "127.0.0.1",
|
||||
port: 0,
|
||||
fetch(request) {
|
||||
requests.push(new URL(request.url).pathname)
|
||||
if (requests.length === 2) recognizing.resolve()
|
||||
return Response.json({ unrelated: true })
|
||||
},
|
||||
})
|
||||
const port = listener.port
|
||||
if (port === undefined) throw new Error("Server did not bind a port")
|
||||
const registration = path.join(root, "state", "opencode", "service-local.json")
|
||||
await fs.mkdir(path.join(root, "config"), { recursive: true })
|
||||
await fs.mkdir(path.dirname(registration), { recursive: true })
|
||||
await fs.writeFile(path.join(root, "config", "service-local.json"), JSON.stringify({ port }))
|
||||
await fs.writeFile(
|
||||
registration,
|
||||
JSON.stringify({
|
||||
id: "stale",
|
||||
version: OPENCODE_VERSION,
|
||||
url: "http://127.0.0.1:1",
|
||||
pid: 2_147_483_647,
|
||||
password: "stale",
|
||||
}),
|
||||
)
|
||||
const contender = Bun.spawn([process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"], {
|
||||
env: isolatedEnv(root),
|
||||
stderr: "pipe",
|
||||
stdout: "ignore",
|
||||
})
|
||||
const stderr = new Response(contender.stderr).text()
|
||||
try {
|
||||
// The first probe is preflight; the second only happens after start() fails to bind.
|
||||
expect(await Promise.race([recognizing.promise.then(() => true), Bun.sleep(20_000).then(() => false)])).toBe(true)
|
||||
expect(requests).toEqual(["/api/info", "/api/info"])
|
||||
await listener.stop(true)
|
||||
|
||||
const info = await Promise.race([
|
||||
waitForInfo(registration, (info) => info.pid === contender.pid),
|
||||
contender.exited.then(() => undefined),
|
||||
])
|
||||
expect(info?.pid, contender.exitCode === null ? undefined : await stderr).toBe(contender.pid)
|
||||
const endpoint = await Effect.runPromise(
|
||||
Service.discover({ file: registration }).pipe(
|
||||
Effect.filterOrFail((value) => value !== undefined),
|
||||
Effect.retry({ times: 400, schedule: Schedule.spaced("50 millis") }),
|
||||
Effect.provide(NodeFileSystem.layer),
|
||||
),
|
||||
)
|
||||
expect(new URL(endpoint.url).port).toBe(String(port))
|
||||
expect(
|
||||
await fetch(new URL("/api/info", endpoint.url), { headers: Service.headers(endpoint) }).then((response) =>
|
||||
response.json(),
|
||||
),
|
||||
).toMatchObject({ pid: contender.pid, version: OPENCODE_VERSION, urls: [endpoint.url] })
|
||||
expect(contender.exitCode).toBe(null)
|
||||
await Effect.runPromise(Service.stop({ file: registration }).pipe(Effect.provide(NodeFileSystem.layer)))
|
||||
expect(await waitForExit(contender)).toBe(true)
|
||||
expect(await Bun.file(registration).exists()).toBe(false)
|
||||
await expectPortAvailable(port)
|
||||
} finally {
|
||||
contender.kill("SIGTERM")
|
||||
await contender.exited
|
||||
await fs.rm(root, { recursive: true, force: true })
|
||||
}
|
||||
}, 45_000)
|
||||
|
||||
test("unresponsive managed port occupancy reports a bounded conflict", async () => {
|
||||
const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-unresponsive-conflict-"))
|
||||
const recognizing = Promise.withResolvers<void>()
|
||||
|
||||
Reference in New Issue
Block a user