mirror of
https://github.com/aaronsb/obsidian-mcp-plugin.git
synced 2026-07-22 06:45:14 +00:00
fix(bridge): self-heal on session expiry — reinitialize + replay (#238)
The .mcpb stdio bridge cleared its session on a 404 and emitted -32000 "please reinitialize", relying on the client to send a fresh initialize. Clients that ignore that signal (Claude Desktop) then sent the next tool call with no session → HTTP 400, and the connection dead-ended until a manual restart. This is the #221/#238 "worked yesterday, broke this morning, tools time out" symptom: a server restart (Obsidian reload / plugin update) or the 1-hour idle session GC orphans the bridge's in-memory session. The bridge now caches the initialize handshake and, on a 404 for a non-initialize call, transparently re-initializes (fresh synthetic id, result discarded) + completes the handshake with notifications/initialized + replays the original message once, then emits that real result. Recovery is invisible to the client and no client-side fix is required. An isReplay flag caps it at a single attempt so a session that dies again can't spin a reinit loop; the existing `initializing` gate serializes concurrent 404s onto one re-init. Made the bridge require()-able (bootstrap guarded by require.main === module, dispatch/state exported) so the self-heal path is unit-testable without spawning a process. Adds tests/bridge-self-heal.test.ts covering reinit+replay success, the give-up path when re-init fails (no loop), and initialize caching. Closes #238. Addresses #221 for users on the bundled .mcpb bridge (third-party bridges like supergateway are unaffected; the server's 404 signal remains spec-compliant per ADR-106). Claude-Session: https://claude.ai/code/session_011yowKCMFCBp61DRBu5xQm4
This commit is contained in:
parent
4f319b300d
commit
add60a43ef
2 changed files with 297 additions and 47 deletions
184
mcpb/server.js
184
mcpb/server.js
|
|
@ -7,20 +7,18 @@ const MCP_URL = process.env.MCP_URL;
|
|||
const API_KEY = process.env.MCP_API_KEY || '';
|
||||
const FETCH_TIMEOUT_MS = 30_000;
|
||||
|
||||
if (!MCP_URL) {
|
||||
process.stderr.write('[obsidian-mcp-bridge] MCP_URL not set\n');
|
||||
process.exit(1);
|
||||
}
|
||||
try { new URL(MCP_URL); } catch {
|
||||
process.stderr.write(`[obsidian-mcp-bridge] invalid MCP_URL: ${MCP_URL}\n`);
|
||||
process.exit(1);
|
||||
}
|
||||
|
||||
let sessionId = null;
|
||||
let notifyAbort = null;
|
||||
// Serializes dispatch until initialize lands a session id. Subsequent
|
||||
// messages chain off this promise so they don't race past the handshake.
|
||||
// Also gates an in-flight self-heal re-initialize so concurrent 404s don't
|
||||
// each kick off their own handshake.
|
||||
let initializing = null;
|
||||
// The last `initialize` we forwarded, kept so the bridge can transparently
|
||||
// re-establish a session if the server evicts/loses it (issue #238). Replayed
|
||||
// with a synthetic id and its result suppressed — the client never sees it.
|
||||
let cachedInitialize = null;
|
||||
let reinitCounter = 0;
|
||||
|
||||
function headers(extra = {}) {
|
||||
const h = {
|
||||
|
|
@ -96,9 +94,48 @@ async function postOnce(message) {
|
|||
});
|
||||
}
|
||||
|
||||
async function dispatch(message) {
|
||||
// Transparently re-establish a session the server has evicted (#238). Replays
|
||||
// the cached `initialize` (with a fresh synthetic id, result discarded), then
|
||||
// completes the handshake with `notifications/initialized` so the new session
|
||||
// is ready for tool calls. Returns true once a fresh session id is in hand.
|
||||
// Never emits to the client — the caller replays the original message and emits
|
||||
// that result, so recovery is invisible.
|
||||
async function reinitialize() {
|
||||
if (!cachedInitialize) return false;
|
||||
// Drop the dead session so postOnce sends the initialize with no session id.
|
||||
sessionId = null;
|
||||
if (notifyAbort) { notifyAbort.abort(); notifyAbort = null; }
|
||||
|
||||
let response;
|
||||
try {
|
||||
response = await postOnce({ ...cachedInitialize, id: `reinit-${reinitCounter++}` });
|
||||
} catch (e) {
|
||||
process.stderr.write(`[obsidian-mcp-bridge] reinit fetch failed: ${e.message}\n`);
|
||||
return false;
|
||||
}
|
||||
const sid = response.headers.get('mcp-session-id');
|
||||
// We only need the session id from the headers; discard the body unread.
|
||||
await response.body?.cancel().catch(() => {});
|
||||
if (!response.ok || !sid) {
|
||||
process.stderr.write(`[obsidian-mcp-bridge] reinit failed: HTTP ${response.status}${sid ? '' : ', no session id'}\n`);
|
||||
return false;
|
||||
}
|
||||
sessionId = sid;
|
||||
|
||||
// Best-effort: a stateful server expects `notifications/initialized` before
|
||||
// it will service tool calls. It's a notification (202, no body to emit).
|
||||
try {
|
||||
const note = await postOnce({ jsonrpc: '2.0', method: 'notifications/initialized' });
|
||||
await note.body?.cancel().catch(() => {});
|
||||
} catch { /* server may not require it; the replay will reveal the truth */ }
|
||||
|
||||
startNotifyStream();
|
||||
return true;
|
||||
}
|
||||
|
||||
async function dispatch(message, isReplay = false) {
|
||||
// If we haven't established a session yet and this isn't the initialize,
|
||||
// wait for the in-flight initialize to land first.
|
||||
// wait for the in-flight initialize (or self-heal re-init) to land first.
|
||||
if (!sessionId && initializing && message.method !== 'initialize') {
|
||||
try { await initializing; } catch { /* let dispatch surface its own error below */ }
|
||||
}
|
||||
|
|
@ -115,6 +152,8 @@ async function dispatch(message) {
|
|||
}
|
||||
|
||||
if (message.method === 'initialize') {
|
||||
// Cache the handshake so we can transparently replay it on session loss.
|
||||
cachedInitialize = message;
|
||||
const sid = response.headers.get('mcp-session-id');
|
||||
if (sid) {
|
||||
sessionId = sid;
|
||||
|
|
@ -123,10 +162,26 @@ async function dispatch(message) {
|
|||
}
|
||||
|
||||
if (response.status === 404 && sessionId) {
|
||||
// The server evicted/lost our session (idle GC, or a server restart). Per
|
||||
// MCP spec the 404 means "re-initialize," but some clients (e.g. Claude
|
||||
// Desktop) don't act on it and the connection dead-ends. Self-heal once:
|
||||
// re-establish a session and replay this exact message so recovery is
|
||||
// invisible to the client (#238). `isReplay` caps it at a single attempt
|
||||
// so a session that dies again instantly can't spin a reinit loop.
|
||||
if (!isReplay && message.method !== 'initialize' && cachedInitialize) {
|
||||
if (!initializing) {
|
||||
initializing = reinitialize().finally(() => { initializing = null; });
|
||||
}
|
||||
let healed = false;
|
||||
try { healed = await initializing; } catch { healed = false; }
|
||||
if (healed) {
|
||||
return await dispatch(message, true);
|
||||
}
|
||||
}
|
||||
sessionId = null;
|
||||
if (notifyAbort) { notifyAbort.abort(); notifyAbort = null; }
|
||||
if (message.id != null) {
|
||||
emit({ jsonrpc: '2.0', id: message.id, error: { code: -32000, message: 'Session expired; please reinitialize' } });
|
||||
emit({ jsonrpc: '2.0', id: message.id, error: { code: -32000, message: 'Session expired and automatic reinitialize failed; please retry.' } });
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
|
@ -156,42 +211,77 @@ async function dispatch(message) {
|
|||
}
|
||||
}
|
||||
|
||||
const rl = createInterface({ input: process.stdin });
|
||||
rl.on('line', (line) => {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) return;
|
||||
let msg;
|
||||
try { msg = JSON.parse(trimmed); }
|
||||
catch (e) {
|
||||
process.stderr.write(`[obsidian-mcp-bridge] stdin parse: ${e.message}\n`);
|
||||
return;
|
||||
// Stdio bootstrap. Guarded so the module can be `require`d in tests (which
|
||||
// drive `dispatch` directly) without validating env, claiming stdin, or
|
||||
// exiting the process. When launched as the bridge, `require.main === module`.
|
||||
function main() {
|
||||
if (!MCP_URL) {
|
||||
process.stderr.write('[obsidian-mcp-bridge] MCP_URL not set\n');
|
||||
process.exit(1);
|
||||
}
|
||||
// Capture the initialize promise so concurrent messages can wait for the
|
||||
// session id before posting.
|
||||
if (msg.method === 'initialize') {
|
||||
initializing = dispatch(msg).finally(() => { initializing = null; });
|
||||
initializing.catch(err => {
|
||||
process.stderr.write(`[obsidian-mcp-bridge] dispatch: ${err.message}\n`);
|
||||
});
|
||||
} else {
|
||||
dispatch(msg).catch(err => {
|
||||
process.stderr.write(`[obsidian-mcp-bridge] dispatch: ${err.message}\n`);
|
||||
});
|
||||
try { new URL(MCP_URL); } catch {
|
||||
process.stderr.write(`[obsidian-mcp-bridge] invalid MCP_URL: ${MCP_URL}\n`);
|
||||
process.exit(1);
|
||||
}
|
||||
});
|
||||
|
||||
process.stdin.on('end', async () => {
|
||||
if (notifyAbort) notifyAbort.abort();
|
||||
// Best-effort: tell the server to retire the session so it doesn't linger
|
||||
// in the pool until idle GC.
|
||||
if (sessionId) {
|
||||
try {
|
||||
await fetch(MCP_URL, {
|
||||
method: 'DELETE',
|
||||
headers: headers(),
|
||||
signal: AbortSignal.timeout(5_000),
|
||||
const rl = createInterface({ input: process.stdin });
|
||||
rl.on('line', (line) => {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) return;
|
||||
let msg;
|
||||
try { msg = JSON.parse(trimmed); }
|
||||
catch (e) {
|
||||
process.stderr.write(`[obsidian-mcp-bridge] stdin parse: ${e.message}\n`);
|
||||
return;
|
||||
}
|
||||
// Capture the initialize promise so concurrent messages can wait for the
|
||||
// session id before posting.
|
||||
if (msg.method === 'initialize') {
|
||||
initializing = dispatch(msg).finally(() => { initializing = null; });
|
||||
initializing.catch(err => {
|
||||
process.stderr.write(`[obsidian-mcp-bridge] dispatch: ${err.message}\n`);
|
||||
});
|
||||
} catch { /* server may have already gone; ignore */ }
|
||||
}
|
||||
process.exit(0);
|
||||
});
|
||||
} else {
|
||||
dispatch(msg).catch(err => {
|
||||
process.stderr.write(`[obsidian-mcp-bridge] dispatch: ${err.message}\n`);
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
process.stdin.on('end', async () => {
|
||||
if (notifyAbort) notifyAbort.abort();
|
||||
// Best-effort: tell the server to retire the session so it doesn't linger
|
||||
// in the pool until idle GC.
|
||||
if (sessionId) {
|
||||
try {
|
||||
await fetch(MCP_URL, {
|
||||
method: 'DELETE',
|
||||
headers: headers(),
|
||||
signal: AbortSignal.timeout(5_000),
|
||||
});
|
||||
} catch { /* server may have already gone; ignore */ }
|
||||
}
|
||||
process.exit(0);
|
||||
});
|
||||
}
|
||||
|
||||
if (require.main === module) {
|
||||
main();
|
||||
}
|
||||
|
||||
// Test seam (inert when run as the bridge): expose dispatch and a state reset
|
||||
// so the self-heal path can be exercised without spawning a process.
|
||||
module.exports = {
|
||||
dispatch,
|
||||
reinitialize,
|
||||
__reset() {
|
||||
sessionId = null;
|
||||
notifyAbort = null;
|
||||
initializing = null;
|
||||
cachedInitialize = null;
|
||||
reinitCounter = 0;
|
||||
},
|
||||
__state() {
|
||||
return { sessionId, cachedInitialize, initializing };
|
||||
},
|
||||
};
|
||||
|
|
|
|||
160
tests/bridge-self-heal.test.ts
Normal file
160
tests/bridge-self-heal.test.ts
Normal file
|
|
@ -0,0 +1,160 @@
|
|||
/**
|
||||
* Regression harness for #238: the .mcpb stdio bridge (mcpb/server.js) must
|
||||
* self-heal when the server evicts/loses a session.
|
||||
*
|
||||
* Before the fix, a 404 on a non-initialize call made the bridge clear its
|
||||
* session and emit `-32000 "please reinitialize"` to the client; clients that
|
||||
* ignore the signal (e.g. Claude Desktop) then sent the next tool call with no
|
||||
* session → HTTP 400 → the connection dead-ended until a manual restart.
|
||||
*
|
||||
* The bridge now caches the `initialize` handshake and, on a 404, transparently
|
||||
* re-initializes and replays the original message once, so recovery is
|
||||
* invisible to the client. These tests drive `dispatch` directly with a mocked
|
||||
* `fetch`, capturing what the bridge writes to stdout.
|
||||
*/
|
||||
|
||||
// Must be set before the bridge module is required (it reads env at load).
|
||||
process.env.MCP_URL = 'http://localhost:3001/mcp';
|
||||
process.env.MCP_API_KEY = 'test-key';
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-require-imports -- bridge is plain CommonJS, loaded as a module to test dispatch
|
||||
const bridge = require('../mcpb/server.js') as {
|
||||
dispatch: (msg: unknown, isReplay?: boolean) => Promise<void>;
|
||||
__reset: () => void;
|
||||
__state: () => { sessionId: string | null; cachedInitialize: unknown };
|
||||
};
|
||||
|
||||
interface FakeResponse {
|
||||
ok: boolean;
|
||||
status: number;
|
||||
headers: { get: (k: string) => string | null };
|
||||
text: () => Promise<string>;
|
||||
body: { cancel: () => Promise<void>; getReader: () => { read: () => Promise<{ done: boolean; value: undefined }> } };
|
||||
}
|
||||
|
||||
function res(status: number, opts: { sid?: string; json?: unknown } = {}): FakeResponse {
|
||||
const h = new Map<string, string>();
|
||||
if (opts.sid) h.set('mcp-session-id', opts.sid);
|
||||
h.set('content-type', opts.json !== undefined ? 'application/json' : 'text/plain');
|
||||
return {
|
||||
ok: status >= 200 && status < 300,
|
||||
status,
|
||||
headers: { get: (k: string) => h.get(k.toLowerCase()) ?? null },
|
||||
text: async () => (opts.json !== undefined ? JSON.stringify(opts.json) : ''),
|
||||
body: {
|
||||
cancel: async () => {},
|
||||
getReader: () => ({ read: async () => ({ done: true, value: undefined }) }),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
interface FetchRecord { method: string; session?: string; rpc?: string; id?: unknown }
|
||||
|
||||
const INIT = { jsonrpc: '2.0', id: 0, method: 'initialize', params: { protocolVersion: '2025-06-18', capabilities: {}, clientInfo: { name: 't', version: '0' } } };
|
||||
const TOOL_CALL = { jsonrpc: '2.0', id: 1, method: 'tools/call', params: { name: 'vault', arguments: { action: 'list' } } };
|
||||
|
||||
describe('mcpb bridge self-heal on session expiry (#238)', () => {
|
||||
let emitted: Array<Record<string, unknown>>;
|
||||
let calls: FetchRecord[];
|
||||
let writeSpy: jest.SpyInstance;
|
||||
|
||||
function installFetch(impl: (rec: FetchRecord, opts: { method: string; headers?: Record<string, string>; body?: string }) => FakeResponse) {
|
||||
(global as unknown as { fetch: unknown }).fetch = jest.fn(async (_url: string, opts: { method: string; headers?: Record<string, string>; body?: string }) => {
|
||||
const session = opts.headers?.['Mcp-Session-Id'];
|
||||
if (opts.method === 'GET' || opts.method === 'DELETE') {
|
||||
calls.push({ method: opts.method, session });
|
||||
return res(200);
|
||||
}
|
||||
const msg = JSON.parse(opts.body as string) as { method: string; id?: unknown };
|
||||
const rec: FetchRecord = { method: opts.method, session, rpc: msg.method, id: msg.id };
|
||||
calls.push(rec);
|
||||
return impl(rec, opts);
|
||||
});
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
bridge.__reset();
|
||||
emitted = [];
|
||||
calls = [];
|
||||
// emit() writes JSON + '\n' to stdout; capture and parse each line.
|
||||
writeSpy = jest.spyOn(process.stdout, 'write').mockImplementation((chunk: string | Uint8Array) => {
|
||||
const text = typeof chunk === 'string' ? chunk : Buffer.from(chunk).toString('utf8');
|
||||
for (const line of text.split('\n')) {
|
||||
if (line.trim()) emitted.push(JSON.parse(line));
|
||||
}
|
||||
return true;
|
||||
});
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
writeSpy.mockRestore();
|
||||
});
|
||||
|
||||
test('404 on a tool call → bridge re-initializes and replays, emitting the real result (not an error)', async () => {
|
||||
let sidSeq = 0;
|
||||
installFetch((rec) => {
|
||||
if (rec.rpc === 'initialize') {
|
||||
sidSeq += 1;
|
||||
return res(200, { sid: `sess-${sidSeq}`, json: { jsonrpc: '2.0', id: rec.id, result: { serverInfo: { name: 'x' } } } });
|
||||
}
|
||||
if (rec.rpc === 'notifications/initialized') return res(202);
|
||||
// First session (sess-1) is the evicted one; the replay uses sess-2.
|
||||
if (rec.session === 'sess-1') return res(404, { sid: 'sess-1', json: { jsonrpc: '2.0', id: rec.id, error: { code: -32001, message: 'expired' } } });
|
||||
return res(200, { json: { jsonrpc: '2.0', id: rec.id, result: { ok: true } } });
|
||||
});
|
||||
|
||||
await bridge.dispatch(INIT);
|
||||
await bridge.dispatch(TOOL_CALL);
|
||||
|
||||
// The client receives the genuine tool result, never the expiry error.
|
||||
const toolResult = emitted.find(m => m.id === 1);
|
||||
expect(toolResult).toBeDefined();
|
||||
expect((toolResult as { result?: { ok?: boolean } }).result?.ok).toBe(true);
|
||||
expect(emitted.some(m => (m as { error?: { code?: number } }).error?.code === -32000)).toBe(false);
|
||||
|
||||
// It re-initialized once and replayed the tool call under the fresh session.
|
||||
const reinit = calls.filter(c => c.rpc === 'initialize');
|
||||
expect(reinit).toHaveLength(2);
|
||||
const toolCalls = calls.filter(c => c.rpc === 'tools/call');
|
||||
expect(toolCalls.map(c => c.session)).toEqual(['sess-1', 'sess-2']);
|
||||
});
|
||||
|
||||
test('reinitialize itself fails → bridge surfaces a single -32000 and does not replay or loop', async () => {
|
||||
let initCount = 0;
|
||||
installFetch((rec) => {
|
||||
if (rec.rpc === 'initialize') {
|
||||
initCount += 1;
|
||||
// First handshake establishes sess-1; the self-heal handshake fails
|
||||
// (200 but no session id), so recovery must give up — not loop.
|
||||
if (initCount === 1) return res(200, { sid: 'sess-1', json: { jsonrpc: '2.0', id: rec.id, result: {} } });
|
||||
return res(200, { json: { jsonrpc: '2.0', id: rec.id, result: {} } });
|
||||
}
|
||||
if (rec.rpc === 'notifications/initialized') return res(202);
|
||||
return res(404, { sid: rec.session, json: { jsonrpc: '2.0', id: rec.id, error: { code: -32001, message: 'expired' } } });
|
||||
});
|
||||
|
||||
await bridge.dispatch(INIT);
|
||||
await bridge.dispatch(TOOL_CALL);
|
||||
|
||||
const err = emitted.find(m => m.id === 1) as { error?: { code?: number; message?: string } } | undefined;
|
||||
expect(err?.error?.code).toBe(-32000);
|
||||
expect(err?.error?.message).toMatch(/reinitialize failed/i);
|
||||
// Exactly one self-heal attempt: 2 initialize calls total, never a third.
|
||||
expect(calls.filter(c => c.rpc === 'initialize')).toHaveLength(2);
|
||||
// The original tool call was attempted once; the failed heal means no
|
||||
// successful replay, and the cap prevents a reinit loop.
|
||||
expect(calls.filter(c => c.rpc === 'tools/call')).toHaveLength(1);
|
||||
});
|
||||
|
||||
test('initialize handshake is cached so a later session loss can be replayed', async () => {
|
||||
installFetch((rec) => {
|
||||
if (rec.rpc === 'initialize') return res(200, { sid: 'sess-1', json: { jsonrpc: '2.0', id: rec.id, result: {} } });
|
||||
return res(200, { json: { jsonrpc: '2.0', id: rec.id, result: {} } });
|
||||
});
|
||||
|
||||
expect(bridge.__state().cachedInitialize).toBeNull();
|
||||
await bridge.dispatch(INIT);
|
||||
expect(bridge.__state().cachedInitialize).toMatchObject({ method: 'initialize' });
|
||||
expect(bridge.__state().sessionId).toBe('sess-1');
|
||||
});
|
||||
});
|
||||
Loading…
Reference in a new issue