diff --git a/packages/opencode/src/lsp/index.ts b/packages/opencode/src/lsp/index.ts index 0c83890e55..8fe4de65be 100644 --- a/packages/opencode/src/lsp/index.ts +++ b/packages/opencode/src/lsp/index.ts @@ -2,6 +2,7 @@ import { BusEvent } from "@/bus/bus-event" import { Bus } from "@/bus" import { Log } from "../util/log" import { LSPClient } from "./client" +import { watch as fswatch, type FSWatcher } from "fs" import path from "path" import { pathToFileURL, fileURLToPath } from "url" import { LSPServer } from "./server" @@ -11,8 +12,9 @@ import { Instance } from "../project/instance" import { Flag } from "@/flag/flag" import { Process } from "../util/process" import { spawn as lspspawn } from "./launch" -import { Effect, Layer, Context } from "effect" +import { Effect, Layer, Context, PubSub, Stream } from "effect" import { InstanceState } from "@/effect/instance-state" +import { Filesystem } from "@/util/filesystem" export namespace LSP { const log = Log.create({ service: "lsp" }) @@ -116,6 +118,8 @@ export namespace LSP { SymbolKind.Enum, ] + const key = (id: string, root: string) => `${id}\0${root}` + const filterExperimentalServers = (servers: Record) => { if (Flag.OPENCODE_EXPERIMENTAL_LSP_TY) { if (servers["pyright"]) { @@ -135,7 +139,10 @@ export namespace LSP { clients: LSPClient.Info[] servers: Record broken: Set + pulse: PubSub.PubSub + pruning: Promise | undefined spawning: Map> + subs: Map }> } export interface Interface { @@ -210,12 +217,25 @@ export namespace LSP { clients: [], servers, broken: new Set(), + pulse: yield* PubSub.unbounded(), + pruning: undefined, spawning: new Map(), + subs: new Map(), } + yield* Stream.fromPubSub(s.pulse).pipe( + Stream.debounce("50 millis"), + Stream.runForEach(() => Effect.promise(() => scan(s))), + Effect.forkScoped, + ) + yield* Effect.addFinalizer(() => - Effect.promise(async () => { - await Promise.all(s.clients.map((client) => client.shutdown())) + Effect.gen(function* () { + yield* PubSub.shutdown(s.pulse).pipe(Effect.ignore) + for (const item of s.subs.values()) { + item.sub.close() + } + yield* Effect.promise(() => Promise.all(s.clients.map((client) => client.shutdown()))) }), ) @@ -225,6 +245,7 @@ export namespace LSP { const getClients = Effect.fnUntraced(function* (file: string) { if (!Instance.containsPath(file)) return [] as LSPClient.Info[] + yield* trim() const s = yield* InstanceState.get(state) return yield* Effect.promise(async () => { const extension = path.parse(file).ext || file @@ -266,6 +287,7 @@ export namespace LSP { } s.clients.push(client) + sync(s) return client } @@ -274,7 +296,8 @@ export namespace LSP { const root = await server.root(file) if (!root) continue - if (s.broken.has(root + server.id)) continue + const id = key(server.id, root) + if (s.broken.has(id)) continue const match = s.clients.find((x) => x.root === root && x.serverID === server.id) if (match) { @@ -282,7 +305,7 @@ export namespace LSP { continue } - const inflight = s.spawning.get(root + server.id) + const inflight = s.spawning.get(id) if (inflight) { const client = await inflight if (!client) continue @@ -290,12 +313,12 @@ export namespace LSP { continue } - const task = schedule(server, root, root + server.id) - s.spawning.set(root + server.id, task) + const task = schedule(server, root, id) + s.spawning.set(id, task) task.finally(() => { - if (s.spawning.get(root + server.id) === task) { - s.spawning.delete(root + server.id) + if (s.spawning.get(id) === task) { + s.spawning.delete(id) } }) @@ -315,7 +338,89 @@ export namespace LSP { return yield* Effect.promise(() => Promise.all(clients.map((x) => fn(x)))) }) + function sync(s: State) { + const next = new Map>() + + for (const client of s.clients) { + const dir = path.dirname(client.root) + const names = next.get(dir) ?? new Set() + names.add(path.basename(client.root)) + next.set(dir, names) + } + + for (const [dir, item] of s.subs) { + if (next.has(dir)) continue + s.subs.delete(dir) + item.sub.close() + } + + for (const [dir, names] of next) { + const existing = s.subs.get(dir) + if (existing) { + existing.names = names + continue + } + try { + const sub = fswatch( + dir, + { persistent: false }, + Instance.bind((_, file) => { + if (file) { + const name = String(file) + if (!s.subs.get(dir)?.names.has(name)) return + } + fire(s) + }), + ) + sub.on( + "error", + Instance.bind(() => { + if (s.subs.get(dir)?.sub !== sub) return + s.subs.delete(dir) + sub.close() + fire(s) + }), + ) + s.subs.set(dir, { sub, names }) + } catch {} + } + } + + function fire(s: State) { + Effect.runFork(PubSub.publish(s.pulse, undefined).pipe(Effect.ignore)) + } + + async function scan(s: State) { + if (s.pruning) return s.pruning + + const task = (async () => { + const dead = ( + await Promise.all( + s.clients.map(async (client) => ((await Filesystem.exists(client.root)) ? undefined : client)), + ) + ).filter((client): client is LSPClient.Info => Boolean(client)) + if (!dead.length) return + + const ids = new Set(dead.map((client) => key(client.serverID, client.root))) + s.clients = s.clients.filter((client) => !ids.has(key(client.serverID, client.root))) + sync(s) + await Promise.all(dead.map((client) => client.shutdown().catch(() => undefined))) + await Bus.publish(Event.Updated, {}) + })().finally(() => { + if (s.pruning === task) s.pruning = undefined + }) + + s.pruning = task + return task + } + + const trim = Effect.fnUntraced(function* () { + const s = yield* InstanceState.get(state) + yield* Effect.promise(() => scan(s)) + }) + const runAll = Effect.fnUntraced(function* (fn: (client: LSPClient.Info) => Promise) { + yield* trim() const s = yield* InstanceState.get(state) return yield* Effect.promise(() => Promise.all(s.clients.map((x) => fn(x)))) }) @@ -325,6 +430,7 @@ export namespace LSP { }) const status = Effect.fn("LSP.status")(function* () { + yield* trim() const s = yield* InstanceState.get(state) const result: Status[] = [] for (const client of s.clients) { @@ -346,7 +452,7 @@ export namespace LSP { if (server.extensions.length && !server.extensions.includes(extension)) continue const root = await server.root(file) if (!root) continue - if (s.broken.has(root + server.id)) continue + if (s.broken.has(key(server.id, root))) continue return true } return false diff --git a/packages/opencode/test/fixture/lsp/fake-lsp-server.js b/packages/opencode/test/fixture/lsp/fake-lsp-server.js index 39e5788012..c26e560a66 100644 --- a/packages/opencode/test/fixture/lsp/fake-lsp-server.js +++ b/packages/opencode/test/fixture/lsp/fake-lsp-server.js @@ -1,7 +1,36 @@ // Simple JSON-RPC 2.0 LSP-like fake server over stdio // Implements a minimal LSP handshake and triggers a request upon notification +const fs = require("fs") const net = require("net") +const path = require("path") + +const mark = process.argv[2] + +if (mark) { + try { + process.chdir(path.dirname(mark)) + } catch {} +} + +function writeMark() { + if (!mark) return + try { + fs.writeFileSync(mark, "exit") + } catch {} +} + +process.on("exit", writeMark) +process.on("SIGTERM", () => { + writeMark() + process.exit(0) +}) +process.on("SIGINT", () => { + writeMark() + process.exit(0) +}) + +setInterval(() => {}, 1000) let nextId = 1 diff --git a/packages/opencode/test/lsp/cleanup-effect.test.ts b/packages/opencode/test/lsp/cleanup-effect.test.ts new file mode 100644 index 0000000000..819aa1f065 --- /dev/null +++ b/packages/opencode/test/lsp/cleanup-effect.test.ts @@ -0,0 +1,68 @@ +import { afterEach, describe, expect } from "bun:test" +import { Deferred, Effect, Layer } from "effect" +import { Bus } from "../../src/bus" +import path from "path" +import { setTimeout as sleep } from "node:timers/promises" +import * as CrossSpawnSpawner from "../../src/effect/cross-spawn-spawner" +import { AppFileSystem } from "../../src/filesystem" +import { LSP } from "../../src/lsp" +import { Instance } from "../../src/project/instance" +import { provideTmpdirInstance } from "../fixture/fixture" +import { testEffect } from "../lib/effect" + +afterEach(async () => { + await Instance.disposeAll() +}) + +const it = testEffect(Layer.mergeAll(LSP.defaultLayer, CrossSpawnSpawner.defaultLayer, AppFileSystem.defaultLayer)) +const server = path.join(import.meta.dir, "../fixture/lsp/fake-lsp-server.js") + +describe("LSP cleanup", () => { + it.live("shuts down clients when their root is deleted", () => + provideTmpdirInstance((dir) => + Effect.gen(function* () { + const fs = yield* AppFileSystem.Service + const mark = path.join(path.dirname(dir), `${path.basename(dir)}.exit`) + const file = path.join(dir, "test.ts") + + yield* Effect.addFinalizer(() => fs.remove(mark, { force: true }).pipe(Effect.ignore)) + yield* fs.writeWithDirs( + path.join(dir, "opencode.json"), + JSON.stringify({ + $schema: "https://opencode.ai/config.json", + lsp: { + typescript: { disabled: true }, + fake: { + command: [process.execPath, server, mark], + extensions: [".ts"], + }, + }, + }), + ) + yield* fs.writeWithDirs(file, "export {}\n") + yield* LSP.Service.use((svc) => svc.touchFile(file)) + expect(yield* LSP.Service.use((svc) => svc.status())).toHaveLength(1) + + const done = yield* Deferred.make() + const off = Bus.subscribe(LSP.Event.Updated, () => { + Deferred.doneUnsafe(done, Effect.void) + }) + yield* Effect.addFinalizer(() => Effect.sync(off)) + + yield* fs.remove(dir, { recursive: true, force: true }) + yield* Deferred.await(done).pipe(Effect.timeout("2 seconds")) + + const stopped = yield* Effect.promise(async () => { + for (const _ of Array.from({ length: 20 })) { + if (await fs.exists(mark)) return true + await sleep(50) + } + return false + }) + + expect(stopped).toBe(true) + expect(yield* LSP.Service.use((svc) => svc.status())).toHaveLength(0) + }), + ), + ) +})