diff --git a/docs/reference/environment-variables.md b/docs/reference/environment-variables.md index 858f3489..7b36f869 100644 --- a/docs/reference/environment-variables.md +++ b/docs/reference/environment-variables.md @@ -70,6 +70,7 @@ The stdio MCP package reads these environment variables directly. A CLI flag tak | `MEMWAL_CLIENT_LABEL` | `--label ` | `MCP Client` / `Walrus Memory MCP` | Friendly delegate-key label shown in the dashboard | | `MEMWAL_MCP_DEBUG` | none | `0` | Set to `1` for verbose stderr logging | | `MEMWAL_MCP_SSE_IDLE_MS` | none | `30000` | Maximum milliseconds of silence on the SSE stream before the bridge treats the session as dead and reconnects. Values below `500` are ignored and fall back to the default. Mainly for tests | +| `MEMWAL_MCP_CALL_TIMEOUT_MS` | none | `240000` | Maximum milliseconds a single request might wait for its response before the bridge answers with a retryable error. Covers a reply lost while the stream itself stays healthy, which `MEMWAL_MCP_SSE_IDLE_MS` cannot detect. The default is derived in code from the slowest server-side tool deadline plus headroom, so it moves with that tool rather than being pinned here. Values below `1000` are ignored and fall back to the default | ## Self-hosted relayer diff --git a/packages/mcp/src/bridge.ts b/packages/mcp/src/bridge.ts index 280a6f73..bdb80754 100644 --- a/packages/mcp/src/bridge.ts +++ b/packages/mcp/src/bridge.ts @@ -185,6 +185,33 @@ function resolveSseIdleMs(): number { return n; } +/** Longest deadline a server-side tool gives its own work (`analyze`). It + * lives in a package this one cannot import from, so raise this whenever that + * grows — otherwise the bridge declares healthy requests orphaned while the + * relayer is still working. */ +const SLOWEST_SERVER_TOOL_MS = 180_000; + +/** 240s as the constants stand. The headroom absorbs the relayer's own + * overhead, so expiry means the reply is lost rather than merely late. */ +const DEFAULT_CALL_TIMEOUT_MS = SLOWEST_SERVER_TOOL_MS + 60_000; + +/** An override below this is a mistake, not an intent. */ +const MIN_CALL_TIMEOUT_MS = 1_000; + +/** Without a cap, a long deadline drifts by a third of itself. */ +const MAX_ORPHAN_SWEEP_MS = 5_000; + +/** How long one request might sit unanswered. The idle watchdog above only sees + * a silent *stream*, which the keepalive prevents, so a lost reply needs its + * own deadline. Override via `MEMWAL_MCP_CALL_TIMEOUT_MS`, mostly for tests. */ +function resolveCallTimeoutMs(): number { + const raw = process.env.MEMWAL_MCP_CALL_TIMEOUT_MS; + if (!raw) return DEFAULT_CALL_TIMEOUT_MS; + const n = Number(raw); + if (!Number.isFinite(n) || n < MIN_CALL_TIMEOUT_MS) return DEFAULT_CALL_TIMEOUT_MS; + return n; +} + interface RpcMessage { jsonrpc: "2.0"; id?: number | string | null; @@ -194,6 +221,12 @@ interface RpcMessage { error?: unknown; } +/** A request forwarded upstream and still awaiting its response. */ +interface InFlightEntry { + msg: RpcMessage; + startedAt: number; +} + interface SseHandshakeResult { /** Absolute URL the client must POST to for outbound JSON-RPC messages. */ postUrl: string; @@ -747,7 +780,10 @@ export async function runBridge( // reconnect so a server-side session swap doesn't strand a tool call // forever waiting for a reply that will never come. Notifications // (no id) and responses (no method) are not tracked. - const inFlight = new Map(); + // `startedAt` is never refreshed, not even by a replay: a reconnect loop + // would otherwise keep pushing the deadline out. + const inFlight = new Map(); + const callTimeoutMs = resolveCallTimeoutMs(); /** IDs of `tools/list` requests we've forwarded to the relayer. When * the response comes back through the SSE pump, we splice in the @@ -827,12 +863,20 @@ export async function runBridge( log.info("bridge.reconnected", { relayer: openingCreds.relayerUrl, replayCount: inFlight.size, + // The count alone cannot tell "nothing was pending" + // from "the entry was dropped early" — the ambiguity + // behind WALM-328's unexplained `replayCount: 0`. + inFlight: Array.from(inFlight.entries()).map(([id, entry]) => ({ + id, + method: entry.msg.method ?? null, + })), }); // Replay any requests that haven't been answered yet against the // fresh session. Iterate over a snapshot — postMessage is async // and the SSE pump may delete entries concurrently as replies // start arriving on the new session. - for (const [id, msg] of Array.from(inFlight.entries())) { + for (const [id, entry] of Array.from(inFlight.entries())) { + const msg = entry.msg; try { // A replayed `initialize` produces a fresh upstream // reply on the NEW session that must also be dropped. @@ -867,9 +911,9 @@ export async function runBridge( // doesn't leak if the loop exits before another replay // re-arms (a leaked arm would swallow a later reused-id // reply). A surviving candidate re-arms fresh next pass. - for (const [, msg] of inFlight) { - if (msg.method === "initialize" && msg.id != null) { - suppressUpstreamReplies.delete(msg.id); + for (const [, entry] of inFlight) { + if (entry.msg.method === "initialize" && entry.msg.id != null) { + suppressUpstreamReplies.delete(entry.msg.id); } } continue; @@ -938,7 +982,7 @@ export async function runBridge( }, }); }; - for (const [, msg] of Array.from(inFlight.entries())) purge(msg); + for (const [, entry] of Array.from(inFlight.entries())) purge(entry.msg); inFlight.clear(); for (const msg of pendingForward.splice(0, pendingForward.length)) purge(msg); } else { @@ -1161,7 +1205,7 @@ export async function runBridge( msg.id !== undefined && msg.id !== null ) { - inFlight.set(msg.id, msg); + inFlight.set(msg.id, { msg, startedAt: Date.now() }); } // Relayer session not up yet, OR the post-connect flush is still // draining — buffer so this request stays behind everything that @@ -1270,17 +1314,23 @@ export async function runBridge( }); } - /** Write the "relayer unavailable" reply for one open request, stop tracking - * it, and never double-answer a locally-answered request. Shared by the - * buffered (`failPendingForward`) and in-flight (`failInFlightRequests`) - * close-outs. Skips: + /** Write a failure reply for one open request, stop tracking it, and never + * double-answer a locally-answered request. Shared by the buffered + * (`failPendingForward`) and in-flight (`failInFlightRequests`) close-outs, + * and by the orphan sweeper — which passes `opts` because "relayer + * unavailable" would be a lie there: the relayer is fine, one reply just + * never arrived. Skips: * - notifications (no id → nothing to reply to; also unforwardable now). * - `initialize` (we already answered it locally; a second response for * that id would corrupt the client's JSON-RPC state — just untrack). * Only `tools/call` shaped requests get the tool-result error envelope; any * other id-bearing request gets a JSON-RPC error object (the correct shape * for a non-tool request). */ - function failRequest(msg: RpcMessage, reason: string): void { + function failRequest( + msg: RpcMessage, + reason: string, + opts: { toolText?: string; errorMessage?: string } = {}, + ): void { if (msg.id == null) return; // notification — nothing to answer if (msg.method === "initialize") { // Locally answered already. Never write a second reply for this id. @@ -1302,7 +1352,9 @@ export async function runBridge( content: [ { type: "text", - text: `❌ Walrus Memory relayer unavailable: ${reason}. The memory tool could not run. Please retry shortly.`, + text: + opts.toolText ?? + `❌ Walrus Memory relayer unavailable: ${reason}. The memory tool could not run. Please retry shortly.`, }, ], isError: true, @@ -1314,7 +1366,9 @@ export async function runBridge( id: msg.id, error: { code: -32000, - message: `Walrus Memory relayer unavailable: ${reason}`, + message: + opts.errorMessage ?? + `Walrus Memory relayer unavailable: ${reason}`, }, }); } @@ -1331,9 +1385,39 @@ export async function runBridge( * to a torn-down session would otherwise hang, since no upstream reply is * coming. Idempotent w.r.t. ids already closed out (delete-then-skip). */ function failInFlightRequests(reason: string): void { - for (const msg of Array.from(inFlight.values())) failRequest(msg, reason); + for (const entry of Array.from(inFlight.values())) failRequest(entry.msg, reason); } + /** Close out requests whose deadline has passed. Without this a reply lost + * on a still-healthy stream leaves its request tracked forever. */ + // Same shape as the SSE watchdog's check interval, but capped. + const sweepIntervalMs = Math.min( + MAX_ORPHAN_SWEEP_MS, + Math.max(500, Math.floor(callTimeoutMs / 3)), + ); + const orphanSweeper = setInterval(() => { + const now = Date.now(); + for (const [id, entry] of Array.from(inFlight.entries())) { + const elapsedMs = now - entry.startedAt; + if (elapsedMs <= callTimeoutMs) continue; + log.warn("bridge.call_orphaned", { + id, + method: entry.msg.method ?? null, + elapsedMs, + }); + failRequest(entry.msg, "no response", { + toolText: + "❌ Walrus Memory did not answer this call. The connection to " + + "the relayer dropped before the result came back. Please retry.", + errorMessage: + "Walrus Memory call was orphaned by a reconnect and never " + + "received a response. Please retry.", + }); + } + }, sweepIntervalMs); + // unref so the sweeper never holds the event loop open during shutdown. + orphanSweeper.unref?.(); + // Kick off the relayer connect in the BACKGROUND — do NOT await it before // wiring stdin below. This is the whole fix: `initialize` / `tools/list` are // answered locally the moment they arrive, while the (possibly slow / cold) @@ -1421,7 +1505,11 @@ export async function runBridge( sse?.abort(); }); - await Promise.race([serverPump, clientPump]); + try { + await Promise.race([serverPump, clientPump]); + } finally { + clearInterval(orphanSweeper); + } markStdinClosed(); const finalStream = sse as SseHandshakeResult | null; finalStream?.abort(); diff --git a/packages/mcp/test/orphaned-call.test.mjs b/packages/mcp/test/orphaned-call.test.mjs new file mode 100644 index 00000000..8df52246 --- /dev/null +++ b/packages/mcp/test/orphaned-call.test.mjs @@ -0,0 +1,306 @@ +/** + * Regression test for the per-call deadline (bridge.ts). + * + * Bug being guarded against (WALM-328): the idle watchdog only notices a silent + * SSE *stream*. The relayer sends a keepalive every 3s, so a stream whose + * heartbeats keep flowing looks perfectly healthy even when one response frame + * has gone missing. That request then sits in `inFlight` forever — no watchdog, + * therefore no reconnect, therefore no replay — and the caller can only report a + * bare "timeout" with nothing to act on. + * + * Repro: + * - Mock relayer keeps the SSE session alive and heartbeating throughout. + * - It answers `initialize` normally, so the bridge is fully connected. + * - It accepts the `memwal_remember` POST with 202 and then never emits the + * matching response frame. + * - With MEMWAL_MCP_CALL_TIMEOUT_MS=2000 the sweeper closes the call out with + * an explicit, retryable error instead of leaving it hanging. + * + * The "only one SSE handshake" assertion is what proves this is the new code + * path: if the watchdog had fired we would see a reconnect, and the test would + * be passing for the wrong reason. + */ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import http from "node:http"; +import { spawn } from "node:child_process"; +import { mkdtempSync, mkdirSync, writeFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +const __dirname = dirname(fileURLToPath(import.meta.url)); +const BIN = resolve(__dirname, "../dist/bin/memwal-mcp.js"); +const EXPECTED_BEARER = "a".repeat(64); +const EXPECTED_ACCOUNT_ID = "0x" + "3".repeat(64); + +function hasBridgeAuth(req) { + return ( + req.headers.authorization === `Bearer ${EXPECTED_BEARER}` && + req.headers["x-memwal-account-id"] === EXPECTED_ACCOUNT_ID + ); +} + +/** Mock relayer whose SSE session stays healthy for the whole test. It answers + * `initialize`, but swallows `memwal_remember` — the POST is accepted and no + * response frame is ever written. `releaseSwallowed()` lets the test emit that + * withheld reply afterwards, to prove a late arrival is not written a second + * time for an id already closed out. */ +function startMockRelayer() { + const sessions = new Map(); + let sseGetCount = 0; + let swallowed = null; + const server = http.createServer((req, res) => { + const url = new URL(req.url, "http://127.0.0.1"); + if (req.method === "GET" && url.pathname === "/version") { + res.writeHead(200, { "content-type": "application/json" }); + res.end( + JSON.stringify({ + apiVersion: "1.0.0", + relayerVersion: "1.0.0", + minSupportedSdk: { mcp: "0.0.1" }, + }), + ); + return; + } + if (req.method === "GET" && url.pathname === "/api/mcp/sse") { + if (!hasBridgeAuth(req)) { + res.writeHead(401); + res.end(); + return; + } + sseGetCount += 1; + const sessionId = `session-${sseGetCount}`; + res.writeHead(200, { + "content-type": "text/event-stream", + "cache-control": "no-cache", + connection: "keep-alive", + }); + res.write( + `event: endpoint\ndata: /api/mcp/messages?sessionId=${sessionId}\n\n`, + ); + sessions.set(sessionId, { res }); + // Heartbeat far faster than the idle timeout so the stream is never + // idle. This is the condition the watchdog cannot help with. + const hb = setInterval(() => { + if (res.writableEnded) { + clearInterval(hb); + return; + } + res.write(":keepalive\n\n"); + }, 200); + hb.unref?.(); + res.on("close", () => clearInterval(hb)); + return; + } + if (req.method === "POST" && url.pathname === "/api/mcp/messages") { + if (!hasBridgeAuth(req)) { + res.writeHead(401); + res.end(); + return; + } + const sessionId = url.searchParams.get("sessionId"); + const session = sessions.get(sessionId); + let body = ""; + req.on("data", (c) => (body += c)); + req.on("end", () => { + if (!session) { + res.writeHead(404); + res.end(); + return; + } + res.writeHead(202); + res.end(); + let msg; + try { + msg = JSON.parse(body); + } catch { + return; + } + if (msg.method === "initialize") { + session.res.write( + `event: message\ndata: ${JSON.stringify({ + jsonrpc: "2.0", + id: msg.id, + result: { + protocolVersion: "2024-11-05", + capabilities: { tools: { listChanged: true } }, + serverInfo: { name: "memwal", version: "0.0.1" }, + }, + })}\n\n`, + ); + return; + } + if (msg.method === "tools/call" && msg.params?.name === "memwal_remember") { + // Accepted, executed server-side as far as the client knows, + // and the reply never comes back. + swallowed = { session, id: msg.id }; + return; + } + }); + return; + } + res.writeHead(404); + res.end(); + }); + return new Promise((res) => { + server.listen(0, "127.0.0.1", () => { + const { port } = server.address(); + res({ + server, + base: `http://127.0.0.1:${port}`, + getSseGetCount: () => sseGetCount, + releaseSwallowed: () => { + if (!swallowed) return false; + swallowed.session.res.write( + `event: message\ndata: ${JSON.stringify({ + jsonrpc: "2.0", + id: swallowed.id, + result: { + content: [{ type: "text", text: "LATE_REPLY" }], + isError: false, + }, + })}\n\n`, + ); + return true; + }, + }); + }); + }); +} + +function makeCreds(relayerUrl) { + return { + delegatePrivateKey: EXPECTED_BEARER, + delegatePublicKeyHex: "b".repeat(64), + delegateAddress: "0x" + "1".repeat(64), + walletAddress: "0x" + "2".repeat(64), + accountId: EXPECTED_ACCOUNT_ID, + packageId: "0x" + "4".repeat(64), + relayerUrl, + label: "Orphan Test", + createdAt: new Date(0).toISOString(), + version: 1, + }; +} + +test("a call whose reply never arrives is closed out with a retryable error", async (t) => { + const mock = await startMockRelayer(); + const home = mkdtempSync(join(tmpdir(), "memwal-orphan-test-")); + const credsPath = join(home, ".memwal", "credentials.json"); + mkdirSync(dirname(credsPath), { recursive: true }); + writeFileSync(credsPath, JSON.stringify(makeCreds(mock.base)), { mode: 0o600 }); + + const child = spawn(process.execPath, [BIN, "--relayer", mock.base, "--web-url", mock.base], { + env: { + ...process.env, + HOME: home, + USERPROFILE: home, + // Well above the call deadline: the stream must never be judged idle, + // so the watchdog cannot be what rescues this call. + MEMWAL_MCP_SSE_IDLE_MS: "30000", + MEMWAL_MCP_CALL_TIMEOUT_MS: "2000", + }, + stdio: ["pipe", "pipe", "pipe"], + }); + + const received = []; + const listeners = new Set(); + let buf = ""; + child.stdout.on("data", (d) => { + buf += d.toString(); + let nl; + while ((nl = buf.indexOf("\n")) >= 0) { + const line = buf.slice(0, nl); + buf = buf.slice(nl + 1); + if (!line.trim()) continue; + let msg; + try { + msg = JSON.parse(line); + } catch { + continue; + } + received.push(msg); + for (const l of [...listeners]) l(msg); + } + }); + let stderrBuf = ""; + child.stderr.on("data", (d) => (stderrBuf += d.toString())); + + const send = (obj) => child.stdin.write(JSON.stringify(obj) + "\n"); + const waitFor = (pred, ms = 15000) => { + const hit = received.find(pred); + if (hit) return Promise.resolve(hit); + return new Promise((res, rej) => { + const timer = setTimeout(() => { + listeners.delete(l); + rej( + new Error( + `timed out waiting for message\n--- stderr ---\n${stderrBuf}\n--- received ---\n${received.map((m) => JSON.stringify(m)).join("\n")}`, + ), + ); + }, ms); + const l = (m) => { + if (pred(m)) { + clearTimeout(timer); + listeners.delete(l); + res(m); + } + }; + listeners.add(l); + }); + }; + + t.after(() => { + child.kill("SIGKILL"); + mock.server.close(); + rmSync(home, { recursive: true, force: true }); + }); + + send({ jsonrpc: "2.0", id: 1, method: "initialize", params: {} }); + const init = await waitFor((m) => m.id === 1 && m.result, 10_000); + assert.equal(init.result.serverInfo.name, "memwal"); + + // The relayer accepts this and never answers it. + send({ + jsonrpc: "2.0", + id: 2, + method: "tools/call", + params: { name: "memwal_remember", arguments: { text: "orphan me" } }, + }); + + // Before the fix this never resolved. + const orphaned = await waitFor((m) => m.id === 2, 10_000); + assert.equal(orphaned.result?.isError, true, "expected a tool-result error envelope"); + assert.match( + JSON.stringify(orphaned.result), + /retry/i, + "the message should tell the caller it is safe to retry", + ); + assert.doesNotMatch( + JSON.stringify(orphaned.result), + /relayer unavailable/i, + "the relayer was healthy — saying otherwise sends debugging the wrong way", + ); + + // The stream stayed healthy throughout, so no reconnect should have happened. + // If this fails, the watchdog rescued the call and the deadline was never + // exercised. + assert.equal( + mock.getSseGetCount(), + 1, + `expected exactly 1 SSE handshake, saw ${mock.getSseGetCount()}`, + ); + + // A late genuine reply for an id already closed out must be dropped, or the + // client would see two responses for the same id. + assert.ok(mock.releaseSwallowed(), "mock should have had a withheld reply"); + await new Promise((r) => setTimeout(r, 1000)); + const repliesForId2 = received.filter((m) => m.id === 2); + assert.equal( + repliesForId2.length, + 1, + `expected exactly one reply for id 2, got ${repliesForId2.length}`, + ); + assert.doesNotMatch(JSON.stringify(repliesForId2), /LATE_REPLY/); +});