mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-17 06:03:25 +00:00
test(client): accelerate service lifecycle tests (#41879)
This commit is contained in:
Kit Langton
GitHub
parent
c217ebe2ad
commit
0df3070d6d
7 changed files with 176 additions and 97 deletions
|
|
@ -4,6 +4,7 @@ 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 { defaultEnsureTiming, ensureTiming, type EnsureTiming } from "../service-timing.js"
|
||||
|
||||
export * from "../service.js"
|
||||
/** Contents of the local service registration file. */
|
||||
|
|
@ -52,11 +53,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 timing = ensureTiming(options)
|
||||
const contenders = new Set<Contender>()
|
||||
let timeouts: { readonly info: Info; readonly count: number } | undefined
|
||||
let announced = false
|
||||
let lastSpawn = 0
|
||||
let spawnDelay = 5_000
|
||||
let spawnDelay = timing.spawnDelay
|
||||
const announce = (reason: "missing" | "version-mismatch", previousVersion?: string) =>
|
||||
Effect.sync(() => {
|
||||
if (announced) return
|
||||
|
|
@ -80,7 +82,7 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
|
|||
})
|
||||
})
|
||||
const found = yield* Effect.gen(function* () {
|
||||
const registration = yield* registered(options.file, true)
|
||||
const registration = yield* registered(options.file, true, timing.requestTimeout)
|
||||
const info = registration.info
|
||||
const service = registration.service
|
||||
if (registration.timedOut && info !== undefined) {
|
||||
|
|
@ -90,20 +92,20 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
|
|||
}
|
||||
if (timeouts.count >= 3) {
|
||||
yield* announce("missing")
|
||||
yield* evict(info, options)
|
||||
yield* evict(info, options, timing)
|
||||
timeouts = undefined
|
||||
lastSpawn = Date.now() - spawnDelay
|
||||
}
|
||||
} else timeouts = undefined
|
||||
if (service !== undefined) {
|
||||
spawnDelay = 5_000
|
||||
spawnDelay = timing.spawnDelay
|
||||
const compatible = !service.legacy && (options.version === undefined || service.version === options.version)
|
||||
if (compatible && service.state === "ready") return Option.some(service)
|
||||
if (compatible && service.state === "failed")
|
||||
return yield* Effect.fail(new Error("Background service failed to start"))
|
||||
if (compatible) return Option.none<LocalService>()
|
||||
yield* announce("version-mismatch", service.version)
|
||||
yield* kill(service, options).pipe(Effect.ignore)
|
||||
yield* kill(service, options, timing).pipe(Effect.ignore)
|
||||
lastSpawn = 0
|
||||
return Option.none<LocalService>()
|
||||
} else if (lastSpawn === 0 && info !== undefined) lastSpawn = Date.now()
|
||||
|
|
@ -111,7 +113,7 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
|
|||
const finished = [...contenders].filter(contenderFinished)
|
||||
const failure = finished.map(contenderFailure).find((error): error is Error => error !== undefined)
|
||||
if (finished.some((item) => item.child.exitCode === 0)) {
|
||||
spawnDelay = Math.min(spawnDelay * 2, 30_000)
|
||||
spawnDelay = Math.min(spawnDelay * 2, timing.maxSpawnDelay)
|
||||
}
|
||||
finished.forEach((item) => contenders.delete(item))
|
||||
if (failure !== undefined && contenders.size === 0) return yield* Effect.fail(failure)
|
||||
|
|
@ -125,7 +127,7 @@ export const ensure = Effect.fn("service.ensure")(function* (options: EnsureOpti
|
|||
}).pipe(
|
||||
Effect.repeat({
|
||||
until: Option.isSome,
|
||||
schedule: Schedule.max([Schedule.spaced("1 second"), Schedule.recurs(120)]),
|
||||
schedule: Schedule.max([Schedule.spaced(timing.pollInterval), Schedule.recurs(timing.attempts)]),
|
||||
}),
|
||||
)
|
||||
if (Option.isNone(found))
|
||||
|
|
@ -150,7 +152,7 @@ function contenderFinished(contender: Contender) {
|
|||
/** 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)
|
||||
if (existing !== undefined) yield* kill(existing, options, defaultEnsureTiming)
|
||||
})
|
||||
|
||||
function fallback() {
|
||||
|
|
@ -198,7 +200,11 @@ const probe = Effect.fnUntraced(function* (info: Info, allowLegacy = false) {
|
|||
return (yield* probeResult(info, allowLegacy)).service
|
||||
})
|
||||
|
||||
const probeResult = Effect.fnUntraced(function* (info: Info, allowLegacy = false) {
|
||||
const probeResult = Effect.fnUntraced(function* (
|
||||
info: Info,
|
||||
allowLegacy = false,
|
||||
timeout = defaultEnsureTiming.requestTimeout,
|
||||
) {
|
||||
const endpoint = {
|
||||
url: info.url,
|
||||
auth:
|
||||
|
|
@ -206,7 +212,7 @@ const probeResult = Effect.fnUntraced(function* (info: Info, allowLegacy = false
|
|||
? undefined
|
||||
: { type: "basic" as const, username: "opencode", password: info.password },
|
||||
} satisfies Endpoint
|
||||
const signal = AbortSignal.timeout(2_000)
|
||||
const signal = AbortSignal.timeout(timeout)
|
||||
const result = yield* Effect.promise(() =>
|
||||
fetch(new URL("/api/health", info.url), {
|
||||
headers: headers(endpoint),
|
||||
|
|
@ -249,10 +255,10 @@ const probeResult = Effect.fnUntraced(function* (info: Info, allowLegacy = false
|
|||
}
|
||||
})
|
||||
|
||||
const registered = Effect.fnUntraced(function* (file?: string, allowLegacy = false) {
|
||||
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)) }
|
||||
return { info, ...(yield* probeResult(info, allowLegacy, timeout)) }
|
||||
})
|
||||
|
||||
// Health-checked lookup without the version gate: lifecycle operations must be
|
||||
|
|
@ -263,7 +269,8 @@ const find = Effect.fnUntraced(function* (options: { readonly file?: string }) {
|
|||
|
||||
// 50ms cadence bounded at ~5s, shared by stop escalation and each ensure
|
||||
// discovery window.
|
||||
const poll = Schedule.max([Schedule.spaced("50 millis"), Schedule.recurs(100)])
|
||||
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)
|
||||
|
|
@ -280,21 +287,25 @@ 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 }) {
|
||||
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), Effect.option)
|
||||
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))
|
||||
yield* stopped(info.pid).pipe(Effect.retry(poll(timing)))
|
||||
})
|
||||
|
||||
const kill = Effect.fnUntraced(function* (service: LocalService, options: { readonly file?: string }) {
|
||||
const requested = yield* requestStop(service)
|
||||
const kill = Effect.fnUntraced(function* (
|
||||
service: LocalService,
|
||||
options: { readonly file?: string },
|
||||
timing: EnsureTiming,
|
||||
) {
|
||||
const requested = yield* requestStop(service, timing.requestTimeout)
|
||||
if (requested === "rejected") return
|
||||
if (requested === "unsupported") {
|
||||
// A stale registration may point at a reused PID. Authenticate again
|
||||
|
|
@ -303,25 +314,25 @@ const kill = Effect.fnUntraced(function* (service: LocalService, options: { read
|
|||
if (current === undefined || !same(current.info, service.info)) return
|
||||
yield* signal(service.info.pid, "SIGTERM")
|
||||
}
|
||||
const done = yield* stopped(service.info.pid).pipe(Effect.retry(poll), Effect.option)
|
||||
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))
|
||||
yield* stopped(service.info.pid).pipe(Effect.retry(poll(timing)))
|
||||
})
|
||||
|
||||
const decodeStopResponse = Schema.decodeUnknownOption(ServiceStatus.StopResponse)
|
||||
|
||||
const requestStop = Effect.fnUntraced(function* (service: LocalService) {
|
||||
const requestStop = Effect.fnUntraced(function* (service: LocalService, timeout = defaultEnsureTiming.requestTimeout) {
|
||||
if (service.info.id === undefined || service.legacy) return "unsupported" as const
|
||||
const response = yield* Effect.tryPromise(() =>
|
||||
fetch(new URL("/api/service/stop", service.info.url), {
|
||||
method: "POST",
|
||||
headers: { ...headers(service.endpoint), "content-type": "application/json" },
|
||||
body: JSON.stringify({ instanceID: service.info.id }),
|
||||
signal: AbortSignal.timeout(2_000),
|
||||
signal: AbortSignal.timeout(timeout),
|
||||
}),
|
||||
).pipe(Effect.option, Effect.map(Option.getOrUndefined))
|
||||
if (response === undefined || response.status === 404 || response.status === 405) return "unsupported" as const
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ 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 { defaultEnsureTiming, ensureTiming, type EnsureTiming } from "../service-timing.js"
|
||||
import type { ServiceHealth, ServiceStopResponse } from "./generated/types.js"
|
||||
|
||||
export * from "../service.js"
|
||||
|
|
@ -32,12 +33,13 @@ 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 timing = ensureTiming(options)
|
||||
const deadline = Date.now() + timing.promiseTimeout
|
||||
const contenders = new Set<Contender>()
|
||||
let timeouts: { readonly info: Info; readonly count: number } | undefined
|
||||
let announced = false
|
||||
let lastSpawn = 0
|
||||
let spawnDelay = 5_000
|
||||
let spawnDelay = timing.spawnDelay
|
||||
|
||||
const announce = (reason: "missing" | "version-mismatch", previousVersion?: string) => {
|
||||
if (announced) return
|
||||
|
|
@ -62,7 +64,7 @@ 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)
|
||||
const registration = await registered(options.file, true, timing.requestTimeout)
|
||||
if (registration.timedOut && registration.info !== undefined) {
|
||||
timeouts = {
|
||||
info: registration.info,
|
||||
|
|
@ -70,21 +72,21 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
|
|||
}
|
||||
if (timeouts.count >= 3) {
|
||||
announce("missing")
|
||||
await evict(registration.info, options)
|
||||
await evict(registration.info, options, timing)
|
||||
timeouts = undefined
|
||||
lastSpawn = Date.now() - spawnDelay
|
||||
}
|
||||
} else timeouts = undefined
|
||||
|
||||
if (registration.service !== undefined) {
|
||||
spawnDelay = 5_000
|
||||
spawnDelay = timing.spawnDelay
|
||||
const service = registration.service
|
||||
const compatible = !service.legacy && (options.version === undefined || service.version === options.version)
|
||||
if (compatible && service.state === "ready") return service.endpoint
|
||||
if (compatible && service.state === "failed") throw new Error("Background service failed to start")
|
||||
if (!compatible) {
|
||||
announce("version-mismatch", service.version)
|
||||
await kill(service, options).catch(() => undefined)
|
||||
await kill(service, options, timing).catch(() => undefined)
|
||||
lastSpawn = 0
|
||||
}
|
||||
} else {
|
||||
|
|
@ -92,7 +94,7 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
|
|||
const finished = [...contenders].filter(contenderFinished)
|
||||
const failure = finished.map(contenderFailure).find((error) => error !== undefined)
|
||||
if (finished.some((item) => item.child.exitCode === 0)) {
|
||||
spawnDelay = Math.min(spawnDelay * 2, 30_000)
|
||||
spawnDelay = Math.min(spawnDelay * 2, timing.maxSpawnDelay)
|
||||
}
|
||||
finished.forEach((item) => contenders.delete(item))
|
||||
if (failure !== undefined && contenders.size === 0) throw failure
|
||||
|
|
@ -103,7 +105,7 @@ export async function ensure(options: EnsureOptions = {}): Promise<Endpoint> {
|
|||
lastSpawn = Date.now()
|
||||
}
|
||||
}
|
||||
await delay(1_000)
|
||||
await delay(timing.pollInterval)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -124,7 +126,7 @@ function contenderFinished(contender: Contender) {
|
|||
/** Stop the registered local service. */
|
||||
export async function stop(options: StopOptions = {}) {
|
||||
const existing = await find(options)
|
||||
if (existing !== undefined) await kill(existing, options)
|
||||
if (existing !== undefined) await kill(existing, options, defaultEnsureTiming)
|
||||
}
|
||||
|
||||
function fallback() {
|
||||
|
|
@ -161,7 +163,7 @@ async function probe(info: Info, allowLegacy = false): Promise<LocalService | un
|
|||
return (await probeResult(info, allowLegacy)).service
|
||||
}
|
||||
|
||||
async function probeResult(info: Info, allowLegacy = false) {
|
||||
async function probeResult(info: Info, allowLegacy = false, timeout = defaultEnsureTiming.requestTimeout) {
|
||||
const endpoint = {
|
||||
url: info.url,
|
||||
auth:
|
||||
|
|
@ -169,7 +171,7 @@ async function probeResult(info: Info, allowLegacy = false) {
|
|||
? undefined
|
||||
: { type: "basic" as const, username: "opencode", password: info.password },
|
||||
} satisfies Endpoint
|
||||
const signal = AbortSignal.timeout(2_000)
|
||||
const signal = AbortSignal.timeout(timeout)
|
||||
const result = await fetch(new URL("/api/health", info.url), {
|
||||
headers: headers(endpoint),
|
||||
signal,
|
||||
|
|
@ -206,10 +208,10 @@ async function probeResult(info: Info, allowLegacy = false) {
|
|||
}
|
||||
}
|
||||
|
||||
async function registered(file?: string, allowLegacy = false) {
|
||||
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)) }
|
||||
return { info, ...(await probeResult(info, allowLegacy, timeout)) }
|
||||
}
|
||||
|
||||
async function find(options: { readonly file?: string }) {
|
||||
|
|
@ -231,10 +233,10 @@ function stopped(pid: number) {
|
|||
}
|
||||
}
|
||||
|
||||
async function waitUntilStopped(pid: number) {
|
||||
for (let attempt = 0; attempt <= 100; attempt++) {
|
||||
async function waitUntilStopped(pid: number, timing: EnsureTiming) {
|
||||
for (let attempt = 0; attempt <= timing.stopPollAttempts; attempt++) {
|
||||
if (stopped(pid)) return true
|
||||
if (attempt < 100) await delay(50)
|
||||
if (attempt < timing.stopPollAttempts) await delay(timing.stopPollInterval)
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
|
@ -243,42 +245,42 @@ 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 }) {
|
||||
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)) return
|
||||
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))) throw new Error(`Server process ${info.pid} is still running`)
|
||||
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 }) {
|
||||
const requested = await requestStop(service)
|
||||
async function kill(service: LocalService, options: { readonly file?: string }, 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 (await waitUntilStopped(service.info.pid)) return
|
||||
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)))
|
||||
if (!(await waitUntilStopped(service.info.pid, timing)))
|
||||
throw new Error(`Server process ${service.info.pid} is still running`)
|
||||
}
|
||||
|
||||
async function requestStop(service: LocalService) {
|
||||
async function requestStop(service: LocalService, timeout = defaultEnsureTiming.requestTimeout) {
|
||||
if (service.info.id === undefined || service.legacy) return "unsupported" as const
|
||||
const response = await fetch(new URL("/api/service/stop", service.info.url), {
|
||||
method: "POST",
|
||||
headers: { ...headers(service.endpoint), "content-type": "application/json" },
|
||||
body: JSON.stringify({ instanceID: service.info.id }),
|
||||
signal: AbortSignal.timeout(2_000),
|
||||
signal: AbortSignal.timeout(timeout),
|
||||
}).catch(() => undefined)
|
||||
if (response === undefined || response.status === 404 || response.status === 405) return "unsupported" as const
|
||||
const body = (await response.json().catch(() => undefined)) as ServiceStopResponse | undefined
|
||||
|
|
|
|||
33
packages/client/src/service-timing.ts
Normal file
33
packages/client/src/service-timing.ts
Normal file
|
|
@ -0,0 +1,33 @@
|
|||
export type EnsureTiming = {
|
||||
readonly pollInterval: number
|
||||
readonly attempts: number
|
||||
readonly requestTimeout: number
|
||||
readonly spawnDelay: number
|
||||
readonly maxSpawnDelay: number
|
||||
readonly promiseTimeout: number
|
||||
readonly stopPollInterval: number
|
||||
readonly stopPollAttempts: number
|
||||
}
|
||||
|
||||
const timings = new WeakMap<object, EnsureTiming>()
|
||||
|
||||
export const defaultEnsureTiming: EnsureTiming = {
|
||||
pollInterval: 1_000,
|
||||
attempts: 120,
|
||||
requestTimeout: 2_000,
|
||||
spawnDelay: 5_000,
|
||||
maxSpawnDelay: 30_000,
|
||||
promiseTimeout: 120_000,
|
||||
stopPollInterval: 50,
|
||||
stopPollAttempts: 100,
|
||||
}
|
||||
|
||||
export function ensureTiming(options: object) {
|
||||
return timings.get(options) ?? defaultEnsureTiming
|
||||
}
|
||||
|
||||
// Keep test timing out of the public lifecycle option types.
|
||||
export function withEnsureTiming<A extends object>(options: A, overrides: Partial<EnsureTiming>): A {
|
||||
timings.set(options, { ...defaultEnsureTiming, ...overrides })
|
||||
return options
|
||||
}
|
||||
26
packages/client/test/fixture/service-timing.ts
Normal file
26
packages/client/test/fixture/service-timing.ts
Normal file
|
|
@ -0,0 +1,26 @@
|
|||
import { withEnsureTiming } from "../../src/service-timing"
|
||||
|
||||
const timing = {
|
||||
pollInterval: 20,
|
||||
requestTimeout: 100,
|
||||
spawnDelay: 200,
|
||||
maxSpawnDelay: 1_200,
|
||||
promiseTimeout: 3_000,
|
||||
stopPollInterval: 5,
|
||||
}
|
||||
|
||||
export function accelerate<A extends object, B>(ensure: (options: A) => B) {
|
||||
return (options: A) => ensure(withEnsureTiming(options, timing))
|
||||
}
|
||||
|
||||
export async function waitForExit(pid: number) {
|
||||
for (let attempt = 0; attempt < 600; attempt++) {
|
||||
try {
|
||||
process.kill(pid, 0)
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
await Bun.sleep(5)
|
||||
}
|
||||
throw new Error(`Timed out waiting for process ${pid}`)
|
||||
}
|
||||
|
|
@ -17,7 +17,7 @@ if (mode === "delayed" || mode === "delayed-failed" || mode === "coordinated" ||
|
|||
if (!owner) process.exit(mode === "coordinated-failed-loser" ? 1 : 0)
|
||||
if (mode === "coordinated" || mode === "coordinated-failed-loser") {
|
||||
while ((await Bun.file(registration + ".starts").text()).trim().split("\n").length < 2) await Bun.sleep(10)
|
||||
if (mode === "coordinated-failed-loser") await Bun.sleep(1_500)
|
||||
if (mode === "coordinated-failed-loser") await Bun.sleep(Number(delay ?? 1_500))
|
||||
} else await Bun.sleep(Number(delay))
|
||||
if (mode === "delayed-failed") process.exit(1)
|
||||
}
|
||||
|
|
@ -30,7 +30,7 @@ const server = Bun.serve({
|
|||
async fetch(request) {
|
||||
const pathname = new URL(request.url).pathname
|
||||
if (pathname === "/api/service/stop" && mode === "reject-stop") {
|
||||
await writeFile(registration + ".stop-attempt", "")
|
||||
await appendFile(registration + ".stop-attempts", process.pid + "\n")
|
||||
return Response.json({ accepted: false })
|
||||
}
|
||||
if (pathname === "/api/service/stop" && mode === "graceful") {
|
||||
|
|
@ -42,6 +42,7 @@ const server = Bun.serve({
|
|||
}
|
||||
if (pathname !== "/api/health") return new Response(null, { status: 404 })
|
||||
requests += 1
|
||||
if (mode === "starting") await writeFile(registration + ".health-request", "")
|
||||
if (mode === "hanging") {
|
||||
await appendFile(registration + ".requests", process.pid + "\n")
|
||||
return new Promise<Response>(() => {})
|
||||
|
|
|
|||
|
|
@ -3,8 +3,10 @@ import { mkdtemp, rm } from "node:fs/promises"
|
|||
import { tmpdir } from "node:os"
|
||||
import { join } from "node:path"
|
||||
import { Service, type EnsureReason } from "../src/promise/service"
|
||||
import { accelerate, waitForExit } from "./fixture/service-timing"
|
||||
|
||||
const fixture = join(import.meta.dir, "fixture/service.ts")
|
||||
const ensure = accelerate(Service.ensure)
|
||||
const processes: Bun.Subprocess[] = []
|
||||
const directories: string[] = []
|
||||
|
||||
|
|
@ -28,7 +30,7 @@ test("ensures a missing service with native promises", async () => {
|
|||
const registration = join(directory, "service.json")
|
||||
const starts: EnsureReason[] = []
|
||||
|
||||
const endpoint = await Service.ensure({
|
||||
const endpoint = await ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "coordinated"],
|
||||
|
|
@ -42,16 +44,16 @@ test("ensures a missing service with native promises", async () => {
|
|||
process.kill(info.pid, "SIGTERM")
|
||||
await waitForExit(info.pid)
|
||||
}
|
||||
}, 15_000)
|
||||
})
|
||||
|
||||
test("waits for a live contender when another native contender fails", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
|
||||
const endpoint = await Service.ensure({
|
||||
const endpoint = await ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "coordinated-failed-loser"],
|
||||
command: [process.execPath, fixture, registration, "coordinated-failed-loser", "300"],
|
||||
})
|
||||
const info = await Bun.file(registration).json()
|
||||
try {
|
||||
|
|
@ -60,12 +62,12 @@ test("waits for a live contender when another native contender fails", async ()
|
|||
process.kill(info.pid, "SIGTERM")
|
||||
await waitForExit(info.pid)
|
||||
}
|
||||
}, 15_000)
|
||||
})
|
||||
|
||||
test("reports a failed registered service", async () => {
|
||||
const registration = await setup("failed-owner")
|
||||
|
||||
await expect(Service.ensure({ file: registration, version: "test", command: [] })).rejects.toThrow(
|
||||
await expect(ensure({ file: registration, version: "test", command: [] })).rejects.toThrow(
|
||||
"Background service failed to start",
|
||||
)
|
||||
})
|
||||
|
|
@ -81,7 +83,7 @@ test("evicts an unresponsive registered service before starting its replacement"
|
|||
await waitForFile(registration)
|
||||
const original = await Bun.file(registration).json()
|
||||
|
||||
const endpoint = await Service.ensure({
|
||||
const endpoint = await ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "delayed", "10"],
|
||||
|
|
@ -94,7 +96,7 @@ test("evicts an unresponsive registered service before starting its replacement"
|
|||
expect(endpoint.url).toBe(replacement.url)
|
||||
process.kill(replacement.pid, "SIGTERM")
|
||||
await waitForExit(replacement.pid)
|
||||
}, 20_000)
|
||||
})
|
||||
|
||||
test("requests graceful stop of the exact service instance", async () => {
|
||||
const registration = await setup("graceful")
|
||||
|
|
@ -126,15 +128,3 @@ async function waitForFile(file: string) {
|
|||
}
|
||||
throw new Error(`Timed out waiting for ${file}`)
|
||||
}
|
||||
|
||||
async function waitForExit(pid: number) {
|
||||
for (let attempt = 0; attempt < 600; attempt++) {
|
||||
try {
|
||||
process.kill(pid, 0)
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
await Bun.sleep(5)
|
||||
}
|
||||
throw new Error(`Timed out waiting for process ${pid}`)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -5,8 +5,10 @@ import { mkdtemp, rm, writeFile } from "node:fs/promises"
|
|||
import { tmpdir } from "node:os"
|
||||
import { join } from "node:path"
|
||||
import { Service, type EnsureReason } from "../src/effect/service"
|
||||
import { accelerate, waitForExit } from "./fixture/service-timing"
|
||||
|
||||
const fixture = join(import.meta.dir, "fixture/service.ts")
|
||||
const ensure = accelerate(Service.ensure)
|
||||
const processes: Bun.Subprocess[] = []
|
||||
const directories: string[] = []
|
||||
|
||||
|
|
@ -25,7 +27,7 @@ test("a concurrent same-version start cannot invalidate a resolved endpoint", as
|
|||
|
||||
const starts: EnsureReason[] = []
|
||||
const first = run(
|
||||
Service.ensure({
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [],
|
||||
|
|
@ -34,7 +36,7 @@ test("a concurrent same-version start cannot invalidate a resolved endpoint", as
|
|||
)
|
||||
await waitForFile(registration + ".first-request")
|
||||
|
||||
const resolved = await run(Service.ensure({ file: registration, version: "test" }))
|
||||
const resolved = await run(ensure({ file: registration, version: "test" }))
|
||||
expect(resolved.url).toBe(original.url)
|
||||
|
||||
await writeFile(registration + ".release", "")
|
||||
|
|
@ -50,9 +52,9 @@ test("waits for a registered service to finish starting", async () => {
|
|||
const registration = join(directory, "service.json")
|
||||
const process = spawn(registration, "starting")
|
||||
await waitForFile(registration)
|
||||
const result = run(Service.ensure({ file: registration, version: "test", command: [] }))
|
||||
const result = run(ensure({ file: registration, version: "test", command: [] }))
|
||||
|
||||
await Bun.sleep(500)
|
||||
await waitForFile(registration + ".health-request")
|
||||
expect(process.exitCode).toBe(null)
|
||||
await writeFile(registration + ".release", "")
|
||||
expect((await result).url).toBe((await Bun.file(registration).json()).url)
|
||||
|
|
@ -64,7 +66,7 @@ test("reports a failed registered service without spawning", async () => {
|
|||
const process = spawn(registration, "failed-owner")
|
||||
await waitForFile(registration)
|
||||
|
||||
await expect(run(Service.ensure({ file: registration, version: "test", command: [] }))).rejects.toThrow(
|
||||
await expect(run(ensure({ file: registration, version: "test", command: [] }))).rejects.toThrow(
|
||||
"Background service failed to start",
|
||||
)
|
||||
expect(process.exitCode).toBe(null)
|
||||
|
|
@ -78,7 +80,7 @@ test("evicts an unresponsive registered service before starting its replacement"
|
|||
const original = await Bun.file(registration).json()
|
||||
|
||||
const endpoint = await run(
|
||||
Service.ensure({
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "delayed", "10"],
|
||||
|
|
@ -92,7 +94,8 @@ test("evicts an unresponsive registered service before starting its replacement"
|
|||
expect(endpoint.url).toBe(replacement.url)
|
||||
expect(await health(endpoint.url)).toEqual({ healthy: true, version: "test", pid: replacement.pid })
|
||||
process.kill(replacement.pid, "SIGTERM")
|
||||
}, 20_000)
|
||||
await waitForExit(replacement.pid)
|
||||
})
|
||||
|
||||
test("requests graceful stop of the exact service instance", async () => {
|
||||
const directory = await temp()
|
||||
|
|
@ -114,7 +117,7 @@ test("does not spawn contenders while an incompatible service rejects replacemen
|
|||
await waitForFile(registration)
|
||||
const controller = new AbortController()
|
||||
const starting = Effect.runPromise(
|
||||
Service.ensure({
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, contender, "record-start"],
|
||||
|
|
@ -122,8 +125,7 @@ test("does not spawn contenders while an incompatible service rejects replacemen
|
|||
{ signal: controller.signal },
|
||||
)
|
||||
|
||||
await waitForFile(registration + ".stop-attempt")
|
||||
await Bun.sleep(500)
|
||||
await waitForLines(registration + ".stop-attempts", 2)
|
||||
controller.abort()
|
||||
await starting.catch(() => undefined)
|
||||
|
||||
|
|
@ -138,18 +140,18 @@ test("a legacy health response is still replaced", async () => {
|
|||
await waitForFile(registration)
|
||||
|
||||
const starts: EnsureReason[] = []
|
||||
const result = run(Service.ensure({ file: registration, command: [], onStart: (reason) => starts.push(reason) }))
|
||||
const result = run(ensure({ file: registration, command: [], onStart: (reason) => starts.push(reason) }))
|
||||
|
||||
await expect(result).rejects.toThrow("Missing service command")
|
||||
expect(starts).toEqual(["version-mismatch"])
|
||||
await existing.exited
|
||||
}, 10_000)
|
||||
})
|
||||
|
||||
test("waits for a slow winner while bounding lock probes", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const endpoint = await run(
|
||||
Service.ensure({
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "coordinated"],
|
||||
|
|
@ -162,17 +164,18 @@ test("waits for a slow winner while bounding lock probes", async () => {
|
|||
expect((await Bun.file(registration + ".starts").text()).trim().split("\n")).toHaveLength(2)
|
||||
} finally {
|
||||
process.kill(info.pid, "SIGTERM")
|
||||
await waitForExit(info.pid)
|
||||
}
|
||||
}, 15_000)
|
||||
})
|
||||
|
||||
test("waits for a live contender when another contender fails", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const endpoint = await run(
|
||||
Service.ensure({
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "coordinated-failed-loser"],
|
||||
command: [process.execPath, fixture, registration, "coordinated-failed-loser", "300"],
|
||||
}),
|
||||
)
|
||||
const info = await Bun.file(registration).json()
|
||||
|
|
@ -180,62 +183,63 @@ test("waits for a live contender when another contender fails", async () => {
|
|||
expect(endpoint.url).toBe(info.url)
|
||||
} finally {
|
||||
process.kill(info.pid, "SIGTERM")
|
||||
await waitForExit(info.pid)
|
||||
}
|
||||
}, 15_000)
|
||||
})
|
||||
|
||||
test("reports a contender that fails to start", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
await expect(
|
||||
run(
|
||||
Service.ensure({
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "failed"],
|
||||
}),
|
||||
),
|
||||
).rejects.toThrow("Server process exited with code 1")
|
||||
}, 10_000)
|
||||
})
|
||||
|
||||
test("reports a contender terminated by a signal", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
await expect(
|
||||
run(
|
||||
Service.ensure({
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "signal"],
|
||||
}),
|
||||
),
|
||||
).rejects.toThrow(/Server process (terminated by|exited with code)/)
|
||||
}, 10_000)
|
||||
})
|
||||
|
||||
test("reports a slow contender that eventually fails", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
await expect(
|
||||
run(
|
||||
Service.ensure({
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "delayed-failed", "8000"],
|
||||
command: [process.execPath, fixture, registration, "delayed-failed", "500"],
|
||||
}),
|
||||
),
|
||||
).rejects.toThrow("Server process exited with code 1")
|
||||
}, 15_000)
|
||||
})
|
||||
|
||||
test("replaces an incompatible owner that appears during startup", async () => {
|
||||
const directory = await temp()
|
||||
const registration = join(directory, "service.json")
|
||||
const starting = run(
|
||||
Service.ensure({
|
||||
ensure({
|
||||
file: registration,
|
||||
version: "test",
|
||||
command: [process.execPath, fixture, registration, "delayed", "8000"],
|
||||
command: [process.execPath, fixture, registration, "delayed", "500"],
|
||||
}),
|
||||
)
|
||||
await Bun.sleep(1_000)
|
||||
await waitForFile(registration + ".starts")
|
||||
const old = spawn(registration, "old")
|
||||
await waitForFile(registration)
|
||||
const endpoint = await starting
|
||||
|
|
@ -246,8 +250,9 @@ test("replaces an incompatible owner that appears during startup", async () => {
|
|||
await old.exited
|
||||
} finally {
|
||||
process.kill(info.pid, "SIGTERM")
|
||||
await waitForExit(info.pid)
|
||||
}
|
||||
}, 20_000)
|
||||
})
|
||||
|
||||
function run<A, E>(effect: Effect.Effect<A, E>) {
|
||||
return Effect.runPromise(effect.pipe(Effect.provide(NodeFileSystem.layer)))
|
||||
|
|
@ -276,6 +281,17 @@ 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}`)
|
||||
}
|
||||
|
||||
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