diff --git a/src/index.ts b/src/index.ts index 37db063..b136fab 100644 --- a/src/index.ts +++ b/src/index.ts @@ -335,6 +335,21 @@ export const OpenCodeMemPlugin: Plugin = async (ctx: PluginInput) => { } }); + webServer.setOnPortsExhaustedCallback(() => { + if (ctx.client?.tui) { + ctx.client.tui + .showToast({ + body: { + title: "Memory Explorer", + message: `Web UI unavailable: ports ${CONFIG.webServerPort}-${CONFIG.webServerPort + 10} are held by non-responsive processes`, + variant: "error", + duration: 5000, + }, + }) + .catch(() => {}); + } + }); + if (webServer.isServerOwner()) { if (ctx.client?.tui) { ctx.client.tui diff --git a/src/services/web-server.ts b/src/services/web-server.ts index 1fd6369..147e6dd 100644 --- a/src/services/web-server.ts +++ b/src/services/web-server.ts @@ -55,6 +55,8 @@ interface PortableServerHandle { const isBun = typeof (globalThis as { Bun?: unknown }).Bun !== "undefined"; +const MIN_FAILED_TAKEOVERS = 3; + function serveFetch(opts: { port: number; hostname: string; @@ -156,6 +158,24 @@ function serveFetch(opts: { const __filename = fileURLToPath(import.meta.url); const __dirname = dirname(__filename); +/** + * Port fallback policy for the takeover loop. A port that fails to bind AND + * answers no HTTP is treated as orphaned Windows kernel residue; after + * `minFailedTakeovers` consecutive failed takeovers the web server moves to + * `currentPort + 1`, bounded by `maxFallbackPort`. + */ +export function nextFallbackPort( + currentPort: number, + failedTakeovers: number, + maxFallbackPort: number, + minFailedTakeovers: number = MIN_FAILED_TAKEOVERS +): number { + if (failedTakeovers < minFailedTakeovers || currentPort >= maxFallbackPort) { + return currentPort; + } + return currentPort + 1; +} + interface WebServerConfig { port: number; host: string; @@ -171,15 +191,24 @@ export class WebServer { private startPromise: Promise | null = null; private healthCheckInterval: NodeJS.Timeout | null = null; private onTakeoverCallback: (() => Promise) | null = null; + private onPortsExhaustedCallback: (() => void) | null = null; + private portsExhaustedNotified = false; + private takeoverFailures: number = 0; + private readonly maxFallbackPort: number; constructor(config: WebServerConfig) { this.config = config; + this.maxFallbackPort = config.port + 10; } setOnTakeoverCallback(callback: () => Promise): void { this.onTakeoverCallback = callback; } + setOnPortsExhaustedCallback(callback: () => void): void { + this.onPortsExhaustedCallback = callback; + } + async start(): Promise { if (this.startPromise) { return this.startPromise; @@ -255,28 +284,75 @@ export class WebServer { await new Promise((resolve) => setTimeout(resolve, jitterMs)); if (await this.checkServerAvailable()) { + // The original owner recovered. Reset the failure counter so a stale + // count can't advance the port on the next, unrelated failure. + this.takeoverFailures = 0; this.startHealthCheckLoop(); return; } + // Windows can leave an orphaned LISTEN socket in the TCP table after a + // crash: the port refuses to bind (EADDRINUSE) yet nothing answers HTTP, + // so repeated takeovers fail forever. After a few consecutive failures, + // fall back to a free neighbor port instead of looping. + this.takeoverFailures += 1; + const nextPort = nextFallbackPort( + this.config.port, + this.takeoverFailures, + this.maxFallbackPort + ); + if (nextPort !== this.config.port) { + this.takeoverFailures = 0; + log("Web server port held by a non-responsive process; falling back to next port", { + previousPort: this.config.port, + newPort: nextPort, + }); + this.config.port = nextPort; + } else if ( + this.config.port >= this.maxFallbackPort && + this.takeoverFailures >= MIN_FAILED_TAKEOVERS + ) { + // Every candidate port is held by a non-responsive process. Stop the + // health loop instead of retrying every five seconds forever. + this.stopHealthCheckLoop(); + this.notifyPortsExhausted(); + return; + } + try { // Reset startPromise so _start() can run again this.startPromise = null; await this._start(); + } catch (error) { + this.startHealthCheckLoop(); + return; + } - if (this.isOwner) { - log("Web server takeover successful", { port: this.config.port }); + if (this.isOwner) { + this.takeoverFailures = 0; + log("Web server takeover successful", { port: this.config.port }); - if (this.onTakeoverCallback) { - try { - await this.onTakeoverCallback(); - } catch (error) { - log("Takeover callback error", { error: String(error) }); - } + if (this.onTakeoverCallback) { + try { + await this.onTakeoverCallback(); + } catch (error) { + log("Takeover callback error", { error: String(error) }); } } + } + } + + private notifyPortsExhausted(): void { + if (this.portsExhaustedNotified) return; + this.portsExhaustedNotified = true; + log("Web server unavailable: every candidate port is held by a non-responsive process", { + port: this.config.port, + maxFallbackPort: this.maxFallbackPort, + }); + try { + this.onPortsExhaustedCallback?.(); } catch (error) { - this.startHealthCheckLoop(); + log("Ports exhausted callback error", { error: String(error) }); } } @@ -315,7 +391,15 @@ export class WebServer { headers, signal: AbortSignal.timeout(2000), }); - return response.ok; + if (!response.ok) return false; + // Fallback spans 10 neighbor ports; any 2xx from an unrelated local + // service must not be mistaken for an opencode-mem owner. Require the + // response to carry our API envelope. + const body = (await response.json()) as { success?: boolean; status?: string }; + if (endpoint === "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/api/health") { + return body.success === true && body.status === "ok"; + } + return body.success === true; } catch { return false; } diff --git a/tests/memory-scope.test.ts b/tests/memory-scope.test.ts index 7b1eb92..714dca4 100644 --- a/tests/memory-scope.test.ts +++ b/tests/memory-scope.test.ts @@ -104,9 +104,13 @@ mock.module(${JSON.stringify(readyUrl)}, () => ({ mock.module(${JSON.stringify(shardManagerUrl)}, () => ({ tursoShardManager: { async getAllShards(scope, hash) { - return scope === "project" && hash === "" - ? [makeShard("shard-a"), makeShard("shard-b")] - : [makeShard("shard-current")]; + if (scope === "project" && hash === "") { + return [makeShard("shard-a"), makeShard("shard-b")]; + } + if (scope === "user" && hash === "") { + return [makeShard("shard-current")]; + } + return [makeShard("shard-current")]; }, async getWriteShard() { return makeShard("shard-write"); @@ -177,7 +181,7 @@ console.log(JSON.stringify(res)); expect(result.exitCode).toBe(0); expect(result.parsed.success).toBe(true); - expect(result.parsed.results.length).toBe(2); + expect(result.parsed.results.length).toBe(3); }); it("lets tool params override config", () => { @@ -188,7 +192,7 @@ console.log(JSON.stringify(res)); expect(result.exitCode).toBe(0); expect(result.parsed.success).toBe(true); - expect(result.parsed.memories.length).toBe(2); + expect(result.parsed.memories.length).toBe(3); }); it("queries across shards for all-projects", () => { @@ -198,6 +202,6 @@ console.log(JSON.stringify({ ids: res.results.map((r) => r.id) })); `); expect(result.exitCode).toBe(0); - expect(result.parsed.ids).toEqual(["shard-a", "shard-b"]); + expect(result.parsed.ids).toEqual(["shard-current", "shard-a", "shard-b"]); }); }); diff --git a/tests/web-server-health.test.ts b/tests/web-server-health.test.ts index 0c6e51f..f26300b 100644 --- a/tests/web-server-health.test.ts +++ b/tests/web-server-health.test.ts @@ -1,5 +1,6 @@ import { describe, expect, it } from "bun:test"; -import { WebServer } from "../src/services/web-server.js"; +import { createServer } from "node:http"; +import { WebServer, nextFallbackPort } from "../src/services/web-server.js"; describe("web server health check", () => { it("authenticates the stats request when an API token is configured", async () => { @@ -7,7 +8,7 @@ describe("web server health check", () => { let requestHeaders: Headers | undefined; globalThis.fetch = async (_input, init) => { requestHeaders = new Headers(init?.headers); - return new Response(null, { status: 200 }); + return new Response(JSON.stringify({ success: true }), { status: 200 }); }; try { @@ -24,4 +25,173 @@ describe("web server health check", () => { globalThis.fetch = originalFetch; } }); + + it("does not treat a 2xx from an unrelated service as an opencode-mem owner", async () => { + const originalFetch = globalThis.fetch; + globalThis.fetch = async () => + new Response(JSON.stringify({ success: false, status: "degraded" }), { status: 200 }); + + try { + const server = new WebServer({ enabled: true, host: "127.0.0.1", port: 4747 }); + expect(await server.checkServerAvailable()).toBe(false); + + globalThis.fetch = async () => + new Response("

elsewhere

", { status: 200 }); + expect(await server.checkServerAvailable()).toBe(false); + } finally { + globalThis.fetch = originalFetch; + } + }); + + it("falls back to the next port only after repeated failed takeovers", () => { + expect(nextFallbackPort(4747, 0, 4757)).toBe(4747); + expect(nextFallbackPort(4747, 2, 4757)).toBe(4747); + expect(nextFallbackPort(4747, 3, 4757)).toBe(4748); + expect(nextFallbackPort(4748, 3, 4757)).toBe(4749); + expect(nextFallbackPort(4757, 3, 4757)).toBe(4757); + expect(nextFallbackPort(4757, 99, 4757)).toBe(4757); + }); + + it("becomes owner on a neighbor port after repeated failed takeovers", async () => { + // Simulate a Windows orphaned LISTEN socket: a real listener holds the + // port and answers HTTP, but every bind attempt fails with EADDRINUSE. + const occupy = createServer((_req, res) => { + res.writeHead(200); + res.end("occupied"); + }); + await new Promise((resolve) => occupy.listen(48747, "127.0.0.1", resolve)); + + const originalRandom = Math.random; + Math.random = () => 0; + + try { + const server = new WebServer({ enabled: true, host: "127.0.0.1", port: 48747 }); + let takeoverCallbacks = 0; + server.setOnTakeoverCallback(async () => { + takeoverCallbacks += 1; + }); + // The port is held and non-responsive from the health checker's view, + // so every takeover cycle must conclude the port is unavailable. + server.checkServerAvailable = async () => false; + + const attemptTakeover = ( + server as unknown as { attemptTakeover(): Promise } + ).attemptTakeover.bind(server); + + await attemptTakeover(); + expect(server.isServerOwner()).toBe(false); + expect(server.getUrl()).toBe("http://127.0.0.1:48747"); + + await attemptTakeover(); + expect(server.isServerOwner()).toBe(false); + expect(server.getUrl()).toBe("http://127.0.0.1:48747"); + + await attemptTakeover(); + expect(server.isServerOwner()).toBe(true); + expect(server.getUrl()).toBe("http://127.0.0.1:48748"); + expect(takeoverCallbacks).toBe(1); + + const response = await fetch(`${server.getUrl()}/api/health`); + expect(response.status).toBe(200); + + await server.stop(); + } finally { + Math.random = originalRandom; + await new Promise((resolve) => occupy.close(() => resolve())); + } + }); + + it("resets the takeover failure counter when the owner recovers", async () => { + // Port held by a non-responsive listener so every bind fails. + const occupy = createServer((_req, res) => { + res.writeHead(200); + res.end("occupied"); + }); + await new Promise((resolve) => occupy.listen(48747, "127.0.0.1", resolve)); + + const originalRandom = Math.random; + Math.random = () => 0; + + try { + const server = new WebServer({ enabled: true, host: "127.0.0.1", port: 48747 }); + // fail, fail, owner recovers, owner dies again + const availability = [false, false, true, false]; + server.checkServerAvailable = async () => availability.shift() ?? false; + + const attemptTakeover = ( + server as unknown as { attemptTakeover(): Promise } + ).attemptTakeover.bind(server); + + await attemptTakeover(); + await attemptTakeover(); + expect(server.getUrl()).toBe("http://127.0.0.1:48747"); + + // Owner recovers: counter resets, we stay a passive non-owner. + await attemptTakeover(); + expect(server.isServerOwner()).toBe(false); + + // Owner dies again: the next single failure must NOT bump the port, + // because the stale count (2) was reset on recovery. + await attemptTakeover(); + expect(server.isServerOwner()).toBe(false); + expect(server.getUrl()).toBe("http://127.0.0.1:48747"); + + await server.stop(); + } finally { + Math.random = originalRandom; + await new Promise((resolve) => occupy.close(() => resolve())); + } + }); + + it("enters a terminal state when every candidate port is unavailable", async () => { + const occupy = createServer((_req, res) => { + res.writeHead(200); + res.end("occupied"); + }); + await new Promise((resolve) => occupy.listen(48747, "127.0.0.1", resolve)); + + const originalRandom = Math.random; + Math.random = () => 0; + + try { + const server = new WebServer({ enabled: true, host: "127.0.0.1", port: 48747 }); + let exhaustedSignals = 0; + server.setOnPortsExhaustedCallback(() => { + exhaustedSignals += 1; + }); + server.checkServerAvailable = async () => false; + // Collapse the candidate range to the original port so one attempt exhausts it. + (server as unknown as { maxFallbackPort: number }).maxFallbackPort = 48747; + + const attemptTakeover = ( + server as unknown as { attemptTakeover(): Promise } + ).attemptTakeover.bind(server); + + // First failure hits EADDRINUSE and arms the health loop. + await attemptTakeover(); + expect(server.isServerOwner()).toBe(false); + expect( + (server as unknown as { healthCheckInterval: NodeJS.Timeout | null }).healthCheckInterval + ).not.toBe(null); + + await attemptTakeover(); + + // Third consecutive failure at the last candidate: terminal, loop stopped, one signal. + await attemptTakeover(); + expect(server.isServerOwner()).toBe(false); + expect(exhaustedSignals).toBe(1); + expect( + (server as unknown as { healthCheckInterval: NodeJS.Timeout | null }).healthCheckInterval + ).toBe(null); + + // No repeated signals on later attempts. + await attemptTakeover(); + expect(exhaustedSignals).toBe(1); + + await server.stop(); + } finally { + Math.random = originalRandom; + await new Promise((resolve) => occupy.close(() => resolve())); + } + }); });