mirror of
https://github.com/anomalyco/opencode.git
synced 2026-11-08 13:08:56 +00:00
fix(core): release location holders on shutdown (#59399)
This commit is contained in:
Aiden Cline
GitHub
parent
e469e85a29
commit
1908fcb8b7
10 changed files with 357 additions and 15 deletions
|
|
@ -6,6 +6,7 @@ import { Bus } from "./bus.js"
|
|||
import { Config } from "./config.js"
|
||||
import { Job } from "./job.js"
|
||||
import { Location } from "./location.js"
|
||||
import { LocationLifecycle } from "./location-lifecycle.js"
|
||||
import { LocationServiceMap } from "./location-service-map.js"
|
||||
import { SessionEvent } from "./session/event.js"
|
||||
import { SessionExecution } from "./session/execution.js"
|
||||
|
|
@ -50,6 +51,22 @@ export function layer(options: { readonly timeToLive?: Duration.Input; readonly
|
|||
Effect.orElseSucceed(() => timeToLive),
|
||||
)
|
||||
|
||||
// An open terminal or client MCP socket keeps its Location loaded: eviction would end the shell under the user,
|
||||
// and the client would reconnect and boot a fresh Location right away.
|
||||
const retained = (ref: Location.Ref) =>
|
||||
locations
|
||||
.contextEffectOption(ref)
|
||||
.pipe(
|
||||
Effect.map(
|
||||
Option.exists((context) =>
|
||||
Option.exists(Context.getOption(context, LocationLifecycle.Service), (lifecycle) =>
|
||||
lifecycle.isRetained(),
|
||||
),
|
||||
),
|
||||
),
|
||||
Effect.scoped,
|
||||
)
|
||||
|
||||
const unsubscribe = yield* bus.listen((event) => {
|
||||
if (!isSessionEvent(event) || !event.location) return Effect.void
|
||||
const location = LocationServiceMap.canonical(event.location)
|
||||
|
|
@ -87,14 +104,16 @@ export function layer(options: { readonly timeToLive?: Duration.Input; readonly
|
|||
)
|
||||
yield* Effect.forEach(
|
||||
expired,
|
||||
(entry) => {
|
||||
if (occupied.has(key(entry.ref))) return touch(entry.ref)
|
||||
entries.delete(key(entry.ref))
|
||||
return Effect.logInfo("location services evicted", {
|
||||
directory: entry.ref.directory,
|
||||
workspaceID: entry.ref.workspaceID,
|
||||
}).pipe(Effect.andThen(locations.invalidate(entry.ref)))
|
||||
},
|
||||
(entry) =>
|
||||
Effect.gen(function* () {
|
||||
if (occupied.has(key(entry.ref)) || (yield* retained(entry.ref))) return yield* touch(entry.ref)
|
||||
entries.delete(key(entry.ref))
|
||||
yield* Effect.logInfo("location services evicted", {
|
||||
directory: entry.ref.directory,
|
||||
workspaceID: entry.ref.workspaceID,
|
||||
})
|
||||
yield* locations.invalidate(entry.ref)
|
||||
}),
|
||||
{ discard: true },
|
||||
)
|
||||
}).pipe(Effect.forever, Effect.forkScoped)
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
export * as LocationLifecycle from "./location-lifecycle.js"
|
||||
|
||||
import { Context, Deferred, Effect, Layer, Schema } from "effect"
|
||||
import { Context, Deferred, Effect, Layer, Schema, Scope } from "effect"
|
||||
import { makeLocationNode } from "@opencode/util/effect/app-node"
|
||||
import { LocationEvent } from "@opencode/schema/location-event"
|
||||
import { Bus } from "./bus.js"
|
||||
|
|
@ -16,7 +16,15 @@ const isSessionEvent = Schema.is(SessionEvent.Durable)
|
|||
|
||||
export class Service extends Context.Service<
|
||||
Service,
|
||||
{ readonly isClosed: () => boolean; readonly shutdown: Effect.Effect<void> }
|
||||
{
|
||||
readonly isClosed: () => boolean
|
||||
readonly shutdown: Effect.Effect<void>
|
||||
/** Completes once shutdown has run. Long-lived holders release the Location on it so its scope can close. */
|
||||
readonly awaitShutdown: Effect.Effect<void>
|
||||
/** Keeps the idle sweep from evicting the Location for the calling scope, for open client connections. */
|
||||
readonly retain: Effect.Effect<void, never, Scope.Scope>
|
||||
readonly isRetained: () => boolean
|
||||
}
|
||||
>()("@opencode/LocationLifecycle") {}
|
||||
|
||||
const layer = Layer.effect(
|
||||
|
|
@ -39,6 +47,7 @@ const layer = Layer.effect(
|
|||
})
|
||||
yield* Effect.addFinalizer(() => unsubscribe)
|
||||
let closed = false
|
||||
let retained = 0
|
||||
const done = yield* Deferred.make<void>()
|
||||
const shutdown = Effect.suspend(() => {
|
||||
if (closed) return Deferred.await(done)
|
||||
|
|
@ -59,6 +68,17 @@ const layer = Layer.effect(
|
|||
return Service.of({
|
||||
isClosed: () => closed,
|
||||
shutdown,
|
||||
awaitShutdown: Deferred.await(done),
|
||||
retain: Effect.acquireRelease(
|
||||
Effect.sync(() => {
|
||||
retained++
|
||||
}),
|
||||
() =>
|
||||
Effect.sync(() => {
|
||||
retained--
|
||||
}),
|
||||
),
|
||||
isRetained: () => retained > 0,
|
||||
})
|
||||
}),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -9,6 +9,8 @@ export { LocationServiceMap } from "./location-service-map.js"
|
|||
|
||||
export type LocationServices = Instance.Services
|
||||
|
||||
const HELD_WARNING_AFTER = "30 seconds"
|
||||
|
||||
export function buildLocationServiceMap(
|
||||
replacements: LayerNode.Replacements = [],
|
||||
): Layer.Layer<LocationServiceMap.Service> {
|
||||
|
|
@ -77,10 +79,34 @@ export function buildLocationServiceMap(
|
|||
const key = LocationServiceMap.canonical(ref)
|
||||
const build = Option.getOrUndefined(MutableHashMap.get(builds, key))
|
||||
MutableHashMap.remove(builds, key)
|
||||
const state = inner.rcMap.state
|
||||
const held = state._tag === "Open" ? Option.getOrUndefined(MutableHashMap.get(state.map, key)) : undefined
|
||||
// Detach routing first, then cancel interactions and notify clients. Running
|
||||
// steps retain their borrowed graph until they can hand off at a boundary.
|
||||
// Do not await a boot here: failed/in-flight builds have their own cleanup path.
|
||||
return inner.invalidate(key).pipe(Effect.andThen(build?.close ?? Effect.void))
|
||||
return inner.invalidate(key).pipe(
|
||||
Effect.andThen(build?.close ?? Effect.void),
|
||||
Effect.andThen(
|
||||
// Borrowers release on shutdown; one that does not keeps every service of this Location running.
|
||||
// The timer lives in the Location's scope, so it stops as soon as the last borrower lets go
|
||||
// (forking into an already closed scope never starts it).
|
||||
held
|
||||
? Effect.sleep(HELD_WARNING_AFTER).pipe(
|
||||
Effect.andThen(
|
||||
Effect.suspend(() =>
|
||||
Effect.logWarning("location still held after shutdown", {
|
||||
directory: key.directory,
|
||||
workspaceID: key.workspaceID,
|
||||
holders: held.refCount,
|
||||
}),
|
||||
),
|
||||
),
|
||||
Effect.forkIn(held.scope),
|
||||
Effect.asVoid,
|
||||
)
|
||||
: Effect.void,
|
||||
),
|
||||
)
|
||||
}).pipe(Effect.uninterruptible),
|
||||
}
|
||||
// Cached instances borrow their owner instead of retaining its Layer scope.
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import { describe, expect } from "bun:test"
|
|||
import { Event } from "@opencode/schema/config"
|
||||
import { Money } from "@opencode/schema/money"
|
||||
import {
|
||||
Context,
|
||||
DateTime,
|
||||
Deferred,
|
||||
Duration,
|
||||
|
|
@ -15,6 +16,7 @@ import {
|
|||
Hash,
|
||||
Layer,
|
||||
LayerMap,
|
||||
Logger,
|
||||
Option,
|
||||
RcMap,
|
||||
Schema,
|
||||
|
|
@ -29,6 +31,7 @@ import { Global } from "@opencode/util/global"
|
|||
import { LocationServiceMap, type LocationServices } from "@opencode/core/location-services"
|
||||
import { LocationActivity } from "@opencode/core/location-activity"
|
||||
import { Location } from "@opencode/core/location"
|
||||
import { LocationLifecycle } from "@opencode/core/location-lifecycle"
|
||||
import { Config } from "@opencode/core/config"
|
||||
import { LocationWatcher } from "@opencode/core/filesystem/location-watcher"
|
||||
import { Plugin } from "@opencode/core/plugin"
|
||||
|
|
@ -316,6 +319,39 @@ describe("LocationServiceMap", () => {
|
|||
}),
|
||||
)
|
||||
|
||||
for (const holder of ["stays", "releases on shutdown"] as const)
|
||||
it.effect(
|
||||
`warns when a borrower ${holder === "stays" ? "still holds" : "lets go of"} an invalidated location`,
|
||||
() => {
|
||||
const warnings: unknown[] = []
|
||||
const logger = Logger.map(Logger.formatStructured, (entry) => {
|
||||
if (entry.level === "WARN") warnings.push(entry.message)
|
||||
})
|
||||
return Effect.gen(function* () {
|
||||
const dir = yield* tmpdirScoped()
|
||||
const ref = Location.Ref.make({ directory: AbsolutePath.make(dir.path) })
|
||||
const locations = yield* LocationServiceMap.Service
|
||||
const borrow = yield* Scope.make()
|
||||
const context = yield* locations.contextEffect(ref).pipe(Scope.provide(borrow))
|
||||
if (holder === "releases on shutdown")
|
||||
yield* Context.get(context, LocationLifecycle.Service).awaitShutdown.pipe(
|
||||
Effect.andThen(Scope.close(borrow, Exit.void)),
|
||||
Effect.forkScoped,
|
||||
)
|
||||
|
||||
yield* locations.invalidate(ref)
|
||||
yield* Effect.yieldNow
|
||||
yield* TestClock.adjust("30 seconds")
|
||||
expect(warnings).toEqual(
|
||||
holder === "stays"
|
||||
? [["location still held after shutdown", { directory: dir.path, workspaceID: undefined, holders: 1 }]]
|
||||
: [],
|
||||
)
|
||||
yield* Scope.close(borrow, Exit.void)
|
||||
}).pipe(Effect.provide(Logger.layer([logger])))
|
||||
},
|
||||
)
|
||||
|
||||
it.live("retries a location after its missing directory is recreated", () =>
|
||||
Effect.acquireRelease(
|
||||
Effect.promise(() => tmpdir()),
|
||||
|
|
|
|||
98
packages/core/test/location-retain.test.ts
Normal file
98
packages/core/test/location-retain.test.ts
Normal file
|
|
@ -0,0 +1,98 @@
|
|||
import { expect } from "bun:test"
|
||||
import { Clock, Context, Duration, Effect, Exit, Layer, RcMap, Schedule, Scope } from "effect"
|
||||
import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"
|
||||
import { LayerNode } from "@opencode/util/effect/layer-node"
|
||||
import { makeGlobalNode } from "@opencode/util/effect/app-node"
|
||||
import { Global } from "@opencode/util/global"
|
||||
import { Bus } from "@opencode/core/bus"
|
||||
import { Database } from "@opencode/core/database/database"
|
||||
import { Job } from "@opencode/core/job"
|
||||
import { Location } from "@opencode/core/location"
|
||||
import { LocationActivity } from "@opencode/core/location-activity"
|
||||
import { LocationLifecycle } from "@opencode/core/location-lifecycle"
|
||||
import { LocationServiceMap } from "@opencode/core/location-services"
|
||||
import { AbsolutePath } from "@opencode/core/schema"
|
||||
import { Session } from "@opencode/core/session"
|
||||
import { SessionExecution } from "@opencode/core/session/execution"
|
||||
import { SessionStore } from "@opencode/core/session/store"
|
||||
import { tempGlobalLayer } from "./fixture/global"
|
||||
import { offlineModels } from "./fixture/models"
|
||||
import { tmpdirScoped } from "./fixture/tmpdir"
|
||||
import { testEffect } from "./lib/effect"
|
||||
|
||||
// LocationActivity measures idleness on this clock; only the test moves it past the TTL (see browser-idle.test.ts).
|
||||
const timeToLive = Duration.minutes(1)
|
||||
const idle = { millis: 0 }
|
||||
const live = Clock.Clock.defaultValue()
|
||||
const activityClock: Clock.Clock = {
|
||||
currentTimeMillisUnsafe: () => live.currentTimeMillisUnsafe() + idle.millis,
|
||||
currentTimeMillis: Effect.sync(() => live.currentTimeMillisUnsafe() + idle.millis),
|
||||
currentTimeNanosUnsafe: () => live.currentTimeNanosUnsafe() + BigInt(idle.millis) * 1_000_000n,
|
||||
currentTimeNanos: Effect.sync(() => live.currentTimeNanosUnsafe() + BigInt(idle.millis) * 1_000_000n),
|
||||
monotonicTimeNanosUnsafe: () => live.monotonicTimeNanosUnsafe(),
|
||||
monotonicTimeNanos: live.monotonicTimeNanos,
|
||||
sleep: (duration) => live.sleep(duration),
|
||||
}
|
||||
|
||||
const it = testEffect(
|
||||
AppNodeBuilder.build(
|
||||
LayerNode.group([Database.node, Bus.node, Session.node, LocationServiceMap.node, LocationActivity.node]),
|
||||
[
|
||||
Global.node.replace(tempGlobalLayer),
|
||||
offlineModels,
|
||||
LocationActivity.node.replace(
|
||||
makeGlobalNode({
|
||||
service: LocationActivity.Service,
|
||||
layer: LocationActivity.layer({ timeToLive, sweepInterval: "100 millis" }).pipe(
|
||||
Layer.provide(Layer.succeed(Clock.Clock, activityClock)),
|
||||
),
|
||||
deps: [Bus.node, Job.node, LocationServiceMap.node, SessionExecution.node, SessionStore.node],
|
||||
}),
|
||||
),
|
||||
],
|
||||
),
|
||||
)
|
||||
|
||||
it.live(
|
||||
"idle sweep keeps a retained location and evicts it once released",
|
||||
() =>
|
||||
Effect.gen(function* () {
|
||||
const locations = yield* LocationServiceMap.Service
|
||||
const [retained, sibling] = yield* Effect.forEach([0, 1], () =>
|
||||
tmpdirScoped().pipe(
|
||||
Effect.map((dir) =>
|
||||
LocationServiceMap.canonical(Location.Ref.make({ directory: AbsolutePath.make(dir.path) })),
|
||||
),
|
||||
),
|
||||
)
|
||||
const lifecycle = Context.get(
|
||||
yield* locations.contextEffect(retained).pipe(Effect.scoped),
|
||||
LocationLifecycle.Service,
|
||||
)
|
||||
yield* locations.contextEffect(sibling).pipe(Effect.scoped)
|
||||
const hold = yield* Scope.make()
|
||||
yield* lifecycle.retain.pipe(Scope.provide(hold))
|
||||
const cached = (ref: Location.Ref) => RcMap.has(locations.rcMap, ref)
|
||||
// Each poll idles every location for a full TTL, whenever the sweep first saw it.
|
||||
const until = (check: Effect.Effect<boolean>) =>
|
||||
Effect.sync(() => {
|
||||
idle.millis += Duration.toMillis(timeToLive)
|
||||
}).pipe(
|
||||
Effect.andThen(check),
|
||||
Effect.flatMap((done) => (done ? Effect.void : Effect.fail("pending"))),
|
||||
Effect.retry(Schedule.spaced("20 millis")),
|
||||
Effect.timeout("10 seconds"),
|
||||
Effect.orDie,
|
||||
)
|
||||
|
||||
// Both expired; the sweep that evicts the sibling skips the retained location.
|
||||
yield* until(cached(sibling).pipe(Effect.map((has) => !has)))
|
||||
expect(yield* cached(retained)).toBe(true)
|
||||
expect(lifecycle.isClosed()).toBe(false)
|
||||
|
||||
yield* Scope.close(hold, Exit.void)
|
||||
yield* until(cached(retained).pipe(Effect.map((has) => !has)))
|
||||
expect(lifecycle.isClosed()).toBe(true)
|
||||
}),
|
||||
30_000,
|
||||
)
|
||||
|
|
@ -11,7 +11,7 @@ import { Socket } from "effect/socket"
|
|||
import { Api } from "../api"
|
||||
import { CorsConfig, isAllowedRequestOrigin } from "../cors"
|
||||
import { locationErrors, requestRef, response } from "../location"
|
||||
import { type Outbound, runPtySocket } from "./pty-socket"
|
||||
import { holdLocation, type Outbound, runPtySocket } from "./pty-socket"
|
||||
|
||||
const notFound = <A, R>(effect: Effect.Effect<A, Mcp.NotFoundError, R>) =>
|
||||
effect.pipe(Effect.mapError((error) => new McpServerNotFoundError({ server: error.server, message: error.message })))
|
||||
|
|
@ -112,6 +112,7 @@ export const McpHandler = HttpApiBuilder.group(Api, "server.mcp", (handlers) =>
|
|||
const { McpSocketClient } = yield* Effect.promise(() => import("@opencode/core/mcp/socket-client"))
|
||||
const socket = yield* Effect.orDie(ctx.request.upgrade)
|
||||
const outbox = yield* Queue.unbounded<Outbound>()
|
||||
yield* holdLocation(outbox)
|
||||
const state = { attached: false, disconnected: false }
|
||||
const transport = McpSocketClient.make({
|
||||
send: (text) => Queue.offerUnsafe(outbox, text),
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import { LocationLifecycle } from "@opencode/core/location-lifecycle"
|
||||
import { Effect, Queue } from "effect"
|
||||
import { Socket } from "effect/socket"
|
||||
|
||||
|
|
@ -31,3 +32,16 @@ export function runPtySocket<R>(input: {
|
|||
Effect.orDie,
|
||||
)
|
||||
}
|
||||
|
||||
// A socket borrows its Location for as long as it stays open. Its client reconnects right away when the socket
|
||||
// closes, so it keeps the idle sweep from evicting the Location (eviction would only boot a fresh one). Any other
|
||||
// shutdown (reload, VCS init, debug eviction) closes the socket with 1012, which releases the borrow so the
|
||||
// Location's scope can close, and the client reconnects to the replacement.
|
||||
export const holdLocation = Effect.fnUntraced(function* (outbox: Queue.Enqueue<Outbound>) {
|
||||
const lifecycle = yield* LocationLifecycle.Service
|
||||
yield* lifecycle.retain
|
||||
yield* lifecycle.awaitShutdown.pipe(
|
||||
Effect.andThen(Queue.offer(outbox, new Socket.CloseEvent(1012, "location shut down"))),
|
||||
Effect.forkChild,
|
||||
)
|
||||
})
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@ import {
|
|||
PTY_CONNECT_TOKEN_HEADER_VALUE,
|
||||
} from "@opencode/protocol/groups/pty"
|
||||
import { locationErrors, requestRef, response } from "../location"
|
||||
import { type Outbound, runPtySocket } from "./pty-socket"
|
||||
import { holdLocation, type Outbound, runPtySocket } from "./pty-socket"
|
||||
|
||||
const notFound = (ptyID: Pty.ID) => new PtyNotFoundError({ ptyID, message: `PTY session not found: ${ptyID}` })
|
||||
|
||||
|
|
@ -137,6 +137,7 @@ export const PtyHandler = HttpApiBuilder.group(Api, "server.pty", (handlers) =>
|
|||
const socket = yield* Effect.orDie(ctx.request.upgrade)
|
||||
// TODO: Integrate graceful-shutdown socket tracking before clients migrate to this route.
|
||||
const outbox = yield* Queue.unbounded<Outbound>()
|
||||
yield* holdLocation(outbox)
|
||||
const attachment = yield* pty
|
||||
.attach(ctx.params.ptyID, {
|
||||
cursor,
|
||||
|
|
@ -173,7 +174,7 @@ export const PtyHandler = HttpApiBuilder.group(Api, "server.pty", (handlers) =>
|
|||
detach: () => attachment?.detach(),
|
||||
})
|
||||
return HttpServerResponse.empty()
|
||||
}).pipe(Effect.provide(locations.get(ref)), locationErrors)
|
||||
}).pipe(Effect.scoped, Effect.provide(locations.get(ref)), locationErrors)
|
||||
}),
|
||||
)
|
||||
}),
|
||||
|
|
|
|||
|
|
@ -1,15 +1,18 @@
|
|||
import { expect } from "bun:test"
|
||||
import { NodeHttpServer } from "@effect/platform-node"
|
||||
import { Location } from "@opencode/core/location"
|
||||
import { AbsolutePath } from "@opencode/core/schema"
|
||||
import { LocationLifecycle } from "@opencode/core/location-lifecycle"
|
||||
import { LocationServiceMap } from "@opencode/core/location-service-map"
|
||||
import { Mcp } from "@opencode/core/mcp/index"
|
||||
import { Session } from "@opencode/core/session"
|
||||
import { Tool } from "@opencode/core/tool"
|
||||
import { McpTool } from "@opencode/core/tool/mcp"
|
||||
import { Context, Effect, Exit, Layer, Schedule, Scope } from "effect"
|
||||
import { Context, Effect, Exit, Layer, MutableHashMap, Option, Schedule, Scope } from "effect"
|
||||
import { HttpRouter, HttpServer } from "effect/http"
|
||||
import { createServer } from "node:http"
|
||||
import path from "node:path"
|
||||
import { connectClientMcp } from "../../client/src/mcp"
|
||||
import { tmpdirScoped } from "../../core/test/fixture/tmpdir"
|
||||
import { it } from "../../core/test/lib/effect"
|
||||
import { ServerProcess } from "../src/process"
|
||||
|
|
@ -47,6 +50,7 @@ const serve = Effect.fnUntraced(function* () {
|
|||
return {
|
||||
base,
|
||||
directory,
|
||||
context,
|
||||
url: (query: Record<string, string>) => {
|
||||
const url = new URL("/api/experimental/client-mcp", base)
|
||||
url.protocol = "ws:"
|
||||
|
|
@ -168,6 +172,66 @@ it.live(
|
|||
20_000,
|
||||
)
|
||||
|
||||
it.live(
|
||||
"closes a client MCP socket with 1012 on reload and reattaches to the fresh location",
|
||||
() =>
|
||||
Effect.gen(function* () {
|
||||
const server = yield* serve()
|
||||
const events: string[] = []
|
||||
class RecordingWebSocket extends globalThis.WebSocket {
|
||||
constructor(...args: ConstructorParameters<typeof globalThis.WebSocket>) {
|
||||
super(...args)
|
||||
this.addEventListener("close", (event) => events.push(`close ${event.code}`))
|
||||
this.addEventListener("message", (event) => {
|
||||
if (String(event.data).includes('"notifications/initialized"')) events.push("attached")
|
||||
})
|
||||
}
|
||||
}
|
||||
const connection = yield* Effect.acquireRelease(
|
||||
Effect.sync(() =>
|
||||
connectClientMcp({
|
||||
url: server.base,
|
||||
name: "tui",
|
||||
directory: server.directory,
|
||||
headers: { authorization },
|
||||
WebSocket: RecordingWebSocket,
|
||||
tools: [{ name: "open_terminal", handler: () => ({ content: [{ type: "text", text: "opened" }] }) }],
|
||||
}),
|
||||
),
|
||||
(connection) => Effect.promise(() => connection.close()),
|
||||
)
|
||||
yield* until(() => events.includes("attached"))
|
||||
const locations = Context.get(server.context, LocationServiceMap.Service)
|
||||
const ref = LocationServiceMap.canonical(Location.Ref.make({ directory: AbsolutePath.make(server.directory) }))
|
||||
const entry = () =>
|
||||
locations.rcMap.state._tag === "Open"
|
||||
? Option.getOrUndefined(MutableHashMap.get(locations.rcMap.state.map, ref))
|
||||
: undefined
|
||||
const old = entry()
|
||||
expect(old).toBeDefined()
|
||||
// An attached client keeps the idle sweep away from its location.
|
||||
const lifecycle = Context.get(yield* locations.contextEffect(ref).pipe(Effect.scoped), LocationLifecycle.Service)
|
||||
expect(lifecycle.isRetained()).toBe(true)
|
||||
|
||||
const response = yield* Effect.promise(() =>
|
||||
fetch(new URL("/api/location/reload", server.base), { method: "POST", headers: { authorization } }),
|
||||
)
|
||||
expect(response.status).toBe(204)
|
||||
// The socket was the old location's last borrower, so closing it let the old location close.
|
||||
yield* until(() => old?.scope.state._tag === "Closed")
|
||||
// The client reconnected on its own and the replacement location attached it.
|
||||
yield* until(() => events.length === 3)
|
||||
expect(events).toEqual(["attached", "close 1012", "attached"])
|
||||
expect(entry()).toBeDefined()
|
||||
expect(entry()).not.toBe(old)
|
||||
|
||||
// Disconnect before the server stops, like the other tests here.
|
||||
yield* Effect.promise(() => connection.close())
|
||||
yield* until(() => events.length === 4)
|
||||
}),
|
||||
20_000,
|
||||
)
|
||||
|
||||
it.live(
|
||||
"accepts a single-use ticket instead of credentials",
|
||||
() =>
|
||||
|
|
|
|||
|
|
@ -128,6 +128,69 @@ live(
|
|||
20_000,
|
||||
)
|
||||
|
||||
live(
|
||||
"PTY websocket closes with 1012 when its location reloads",
|
||||
() =>
|
||||
Effect.gen(function* () {
|
||||
const directory = yield* Effect.acquireRelease(
|
||||
Effect.promise(() => fs.mkdtemp(path.join(os.tmpdir(), "opencode-pty-reload-"))),
|
||||
(directory) => Effect.promise(() => fs.rm(directory, { recursive: true, force: true })),
|
||||
)
|
||||
const server = yield* ServerProcess.start<never, never>({
|
||||
hostname: "127.0.0.1",
|
||||
port: 0,
|
||||
password: "secret",
|
||||
app: { version: "test-version" },
|
||||
database: { path: ":memory:" },
|
||||
models: { fetch: false },
|
||||
config: { directory, project: false, content: "{}" },
|
||||
fs: { filewatcher: false },
|
||||
})
|
||||
const base = HttpServer.formatAddress(server.address)
|
||||
const client = OpenCode.make({ baseUrl: base, headers: { authorization: `Basic ${btoa("opencode:secret")}` } })
|
||||
const location = { directory }
|
||||
yield* Effect.promise(async () => {
|
||||
const terminal = await client.pty.create({
|
||||
location,
|
||||
command: "/bin/sh",
|
||||
args: ["-c", "sleep 30"],
|
||||
cwd: directory,
|
||||
})
|
||||
const token = await client.pty.connect.token({ location, ptyID: terminal.data.id, "x-opencode-ticket": "1" })
|
||||
const url = new URL(`/api/pty/${terminal.data.id}/connect`, base)
|
||||
url.protocol = "ws:"
|
||||
url.searchParams.set("location[directory]", directory)
|
||||
url.searchParams.set("ticket", token.data.ticket)
|
||||
const socket = new WebSocket(url)
|
||||
const state = { open: false, code: 0 }
|
||||
socket.addEventListener("open", () => (state.open = true))
|
||||
socket.addEventListener("close", (event) => (state.code = event.code))
|
||||
try {
|
||||
await waitFor(() => state.open)
|
||||
// The socket holds the location; reload must still close it so the old location can shut down.
|
||||
await client.location.reload()
|
||||
await waitFor(() => state.code !== 0)
|
||||
expect(state.code).toBe(1012)
|
||||
// Releasing the socket let the old location close, and its shell with it.
|
||||
await waitFor(() => !alive(terminal.data.pid))
|
||||
expect(await client.pty.get({ location, ptyID: terminal.data.id }).catch(() => undefined)).toBeUndefined()
|
||||
} finally {
|
||||
socket.close()
|
||||
}
|
||||
})
|
||||
}),
|
||||
20_000,
|
||||
)
|
||||
|
||||
function alive(pid: number) {
|
||||
try {
|
||||
process.kill(pid, 0)
|
||||
return true
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
async function waitFor(predicate: () => boolean) {
|
||||
for (let attempt = 0; attempt < 200; attempt++) {
|
||||
if (predicate()) return
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue