modelcontextprotocol/typescript-sdk
Watch
1
0
Fork
You've already forked typescript-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor(client): response cache stores serialized documents; drop structuredClone (#2468)

Browse source
This commit is contained in:
Felix Weinberger 2026-07-13 17:39:41 +01:00 • committed by GitHub
commit 5db6e38f2b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 457 additions and 112 deletions
Download patch file Download diff file Expand all files Collapse all files

36
packages/client/src/client/client.ts
View file

@ -1475,7 +1475,7 @@ export class Client extends Protocol<ClientContext> {
* Lists available prompts.
*
* Called without a `cursor` (the common case), this walks every page and
* returns the complete aggregated list with `nextCursor: undefined`; the
* returns the complete aggregated list with no `nextCursor`; the
* aggregate is also written to the {@linkcode ResponseCacheStore}. Pass an
* explicit `{ cursor }` to fetch a single page and walk pagination
* yourself — the per-page path returns the server's raw page (with
@ -1515,7 +1515,7 @@ export class Client extends Protocol<ClientContext> {
* Lists available resources.
*
* Called without a `cursor` (the common case), this walks every page and
* returns the complete aggregated list with `nextCursor: undefined`; the
* returns the complete aggregated list with no `nextCursor`; the
* aggregate is also written to the {@linkcode ResponseCacheStore}. Pass an
* explicit `{ cursor }` to fetch a single page and walk pagination
* yourself — the per-page path returns the server's raw page (with
@ -1557,7 +1557,7 @@ export class Client extends Protocol<ClientContext> {
* Lists available resource URI templates for dynamic resources.
*
* Called without a `cursor`, this walks every page and returns the
* complete aggregated list with `nextCursor: undefined`; the aggregate is
* complete aggregated list with no `nextCursor`; the aggregate is
* also written to the {@linkcode ResponseCacheStore}. Pass an explicit
* `{ cursor }` to fetch a single page — see
* {@linkcode listResources | listResources()} for the per-page contract.
@ -1642,7 +1642,10 @@ export class Client extends Protocol<ClientContext> {
cursor = page.nextCursor;
pages++;
}
acc.nextCursor = undefined;
// delete, not `= undefined`: the cache's JSON codec drops
// explicit-undefined properties, so only absence keeps
// `'nextCursor' in result` identical between wire and cache hit.
delete acc.nextCursor;
finalize?.(acc);
if (bypass) return acc;
// The aggregate is ALWAYS written: even when the resolved TTL is ≤0
@ -1681,12 +1684,13 @@ export class Client extends Protocol<ClientContext> {
/**
* The cache-serving front of every cacheable verb (mcp.d's `cachedFetch`
* read half): under `cacheMode: 'use'` (the default), a held entry whose
* `expiresAt` is in the future is returned as a `structuredClone` and the
* round trip is skipped. `'refresh'` and `'bypass'` always fetch (the
* caller decides whether to write). A custom store whose `get()` rejects
* is routed to `onerror` and treated as a miss — cache bookkeeping never
* blocks a request from reaching the wire.
* read half): under `cacheMode: 'use'` (the default), a fresh held entry
* is served and the round trip is skipped. `'refresh'` and `'bypass'`
* always fetch (the caller decides whether to write). Freshness and
* decoding live in {@linkcode ClientResponseCache.read}; every hit is
* freshly parsed, so the caller owns it outright. A custom store
* whose `get()` rejects is routed to `onerror` and treated as a miss —
* cache bookkeeping never blocks a request from reaching the wire.
*/
private async _serveFromCache<R>(
method: string,
@ -1694,8 +1698,8 @@ export class Client extends Protocol<ClientContext> {
options: CacheableRequestOptions | undefined
): Promise<R | undefined> {
if (options?.cacheMode === 'bypass' || options?.cacheMode === 'refresh') return undefined;
const entry = await this._cache.read(method, params).catch(error => void this._reportStoreError(error));
if (entry?.expiresAt !== undefined && entry.expiresAt > this._cache.now()) {
const hit = await this._cache.read(method, params).catch(error => void this._reportStoreError(error));
if (hit !== undefined) {
// A pre-aborted caller signal must reject the same way it would on
// the wire path (`Protocol.request()` wraps an already-aborted
// signal as `SdkError(RequestTimeout, reason)`); without this guard
@ -1705,11 +1709,7 @@ export class Client extends Protocol<ClientContext> {
const reason = options.signal.reason;
throw reason instanceof SdkError ? reason : new SdkError(SdkErrorCode.RequestTimeout, String(reason));
}
// Clone on the way out so a caller mutating the returned aggregate
// (e.g. `result.tools.sort(...)`) cannot reach the cache or the
// stamp-memoized indices derived from it — the same invariant
// `_cache.write` upholds on the way in.
return structuredClone(entry.value) as R;
return hit.value as R;
}
return undefined;
}
@ -2369,7 +2369,7 @@ export class Client extends Protocol<ClientContext> {
* Lists available tools.
*
* Called without a `cursor` (the common case), this walks every page and
* returns the complete aggregated list with `nextCursor: undefined`; the
* returns the complete aggregated list with no `nextCursor`; the
* aggregate is also written to the {@linkcode ResponseCacheStore} (the
* source for {@linkcode callTool | callTool()}'s output-schema validation
* and SEP-2243 `Mcp-Param-*` header mirroring). Pass an explicit

144
packages/client/src/client/responseCache.ts
View file

@ -52,16 +52,18 @@ export interface CacheKey {
}
/**
* One cached response body. `value` is the verbatim decoded result; `stamp` is
* the store-generated monotonically increasing write counter — opaque to
* callers. Derived views (e.g. a `name → Tool` index) memoize against it and
* re-derive only when it changes. `expiresAt` (absolute ms epoch, `now +
* ttlMs`) and `scope` are the client-computed freshness metadata; the store
* MUST persist them and hand them back on `get` so the read path can decide
* freshness and gate the shared-partition fallback on `scope === 'public'`.
* One cached response body. `value` is the JSON-serialized result document —
* a store is a dumb string carrier and persists it verbatim; the cache owns
* both codec halves. `stamp` is the store-generated monotonically increasing write counter —
* opaque to callers. Derived views (e.g. a `name → Tool` index) memoize
* against it and re-derive only when it changes. `expiresAt` (absolute ms
* epoch, `now + ttlMs`) and `scope` are the client-computed freshness
* metadata; the store MUST persist them and hand them back on `get` so the
* read path can decide freshness and gate the shared-partition fallback on
* `scope === 'public'`.
*/
export interface CacheEntry {
readonly value: unknown;
readonly value: string;
readonly stamp: number;
readonly expiresAt?: number;
readonly scope?: CacheScope;
@ -96,7 +98,7 @@ export interface ResponseCacheStore {
* `scope` (the client-computed freshness metadata); the store MUST persist
* them and hand them back on `get`.
*/
set(key: CacheKey, entry: { value: unknown; expiresAt?: number; scope?: CacheScope }): MaybePromise<number>;
set(key: CacheKey, entry: { value: string; expiresAt?: number; scope?: CacheScope }): MaybePromise<number>;
/**
* Drop the single entry under `key` (no-op if absent). Called for both
* `notifications/resources/updated` (per-URI) and the `list_changed`
@ -175,7 +177,7 @@ export class InMemoryResponseCacheStore implements ResponseCacheStore {
return this._entries.get(keyOf(key));
}
set(key: CacheKey, entry: { value: unknown; expiresAt?: number; scope?: CacheScope }): number {
set(key: CacheKey, entry: { value: string; expiresAt?: number; scope?: CacheScope }): number {
const k = keyOf(key);
const exempt = CAP_EXEMPT_METHODS.has(key.method);
const isNew = !this._entries.has(k);
@ -409,16 +411,25 @@ export class ClientResponseCache {
*/
async evict(method: string): Promise<void> {
this._evictionGeneration.set(method, (this._evictionGeneration.get(method) ?? 0) + 1);
await this._deleteBoth(method, '');
}
/**
* Guarded two-partition delete of `{method, params}`: each partition's
* `delete` is independently wrapped so a custom store's failure on one is
* reported and does not skip the other, and the call always resolves.
*/
private async _deleteBoth(method: string, params: string): Promise<void> {
const ownPartition = this._partitionFor('private');
const sharedPartition = this._partitionFor('public');
try {
await this._store.delete({ method, params: '', partition: ownPartition });
await this._store.delete({ method, params, partition: ownPartition });
} catch (error) {
this._reportError(error);
}
if (sharedPartition !== ownPartition) {
try {
await this._store.delete({ method, params: '', partition: sharedPartition });
await this._store.delete({ method, params, partition: sharedPartition });
} catch (error) {
this._reportError(error);
}
@ -449,20 +460,7 @@ export class ClientResponseCache {
// `resetForReconnect`).
const current = this._evictionGeneration.get(gk);
if (current !== undefined) this._evictionGeneration.set(gk, current + 1);
const ownPartition = this._partitionFor('private');
const sharedPartition = this._partitionFor('public');
try {
await this._store.delete({ method, params, partition: ownPartition });
} catch (error) {
this._reportError(error);
}
if (sharedPartition !== ownPartition) {
try {
await this._store.delete({ method, params, partition: sharedPartition });
} catch (error) {
this._reportError(error);
}
}
await this._deleteBoth(method, params);
}
/**
@ -487,13 +485,14 @@ export class ClientResponseCache {
* overwriting the eviction with the stale aggregate would lose the
* invalidation.
*
* The stored value is a `structuredClone` of `value`, so a caller
* mutating the aggregate it was returned (e.g. `result.tools.sort(...)`)
* cannot reach the cache or the stamp-memoized indices derived from it. A
* custom store whose `set()` throws or rejects is routed to the
* `reportError` sink and the write resolves — cache bookkeeping never
* costs the caller a result it already fetched (consistent with the
* eviction path).
* The value is stored as its JSON-serialized document; serialization
* doubles as the mutation barrier, so a caller mutating the returned
* aggregate cannot reach the cache or its derived indices. A value that
* is not JSON-serializable (reachable only via in-process transports)
* fails the write loudly into the `reportError` sink. A custom store
* whose `set()` throws or rejects is routed to the same sink and the
* write resolves — cache bookkeeping never costs the caller a result it
* already fetched.
*
* `freshness` carries the client-computed `expiresAt` (absolute ms epoch,
* `now + ttlMs`) and the server-reported `cacheScope`. The storage
@ -529,7 +528,7 @@ export class ClientResponseCache {
try {
await this._store.set(
{ method, params, partition },
{ value: structuredClone(value), expiresAt: freshness?.expiresAt, scope: freshness?.scope }
{ value: encodeCacheValue(value), expiresAt: freshness?.expiresAt, scope: freshness?.scope }
);
} catch (error) {
this._reportError(error);
@ -548,13 +547,32 @@ export class ClientResponseCache {
}
/**
* Read the cached entry for `{method, params}` via the two-probe lookup
* (own-partition then this server's shared partition, gated on
* `scope === 'public'`). The caller owns the freshness check
* (`entry.expiresAt > now()`); a missing `expiresAt` is never fresh.
* Serve the fresh cached result for `{method, params}`, or `undefined`.
* Lookup is the two-probe order (own-partition then this server's shared
* partition, gated on `scope === 'public'`); freshness is
* `entry.expiresAt > now()` (a missing `expiresAt` is never fresh),
* checked BEFORE decoding so stale entries cost no parse. Every hit is
* freshly parsed, so the caller owns the value outright. An entry whose
* document does not parse or is not an object (corrupted external
* store) is reported,
* deleted, and treated as a miss — deleted because a fresh-but-corrupt
* entry would otherwise re-parse and re-report on every read until its
* `expiresAt` passes.
*/
async read(method: string, params?: string): Promise<CacheEntry | undefined> {
return this._probe(method, params);
async read(method: string, params?: string): Promise<{ value: unknown } | undefined> {
const entry = await this._probe(method, params);
if (entry?.expiresAt === undefined || !(entry.expiresAt > this.now())) return undefined;
try {
const parsed: unknown = JSON.parse(entry.value);
if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) {
throw new TypeError('cached document is not an object');
}
return { value: parsed };
} catch (error) {
this._reportError(error);
await this._deleteBoth(method, params ?? '');
return undefined;
}
}
/**
@ -592,8 +610,12 @@ export class ClientResponseCache {
return undefined;
}
if (this._toolIndex?.stamp !== entry.stamp) {
const list = this._decodeListTools(entry);
const byName = new Map<string, Tool>();
for (const tool of (entry.value as ListToolsResult).tools) byName.set(tool.name, tool);
if (list !== undefined) for (const tool of list.tools) byName.set(tool.name, tool);
// A failed decode memoizes an EMPTY index under the same stamp, so
// the corrupt document is parsed (and reported) once per stamp —
// not once per callTool — and a later rewrite re-derives.
this._toolIndex = { stamp: entry.stamp, byName };
}
return this._toolIndex.byName.get(name);
@ -628,8 +650,9 @@ export class ClientResponseCache {
return undefined;
}
if (this._toolOutputValidatorIndex?.stamp !== entry.stamp) {
const list = this._decodeListTools(entry) ?? { tools: [] };
const byName = new Map<string, unknown>();
for (const tool of (entry.value as ListToolsResult).tools) {
for (const tool of list.tools) {
const compiled = compile(tool);
if (compiled !== undefined) byName.set(tool.name, compiled);
}
@ -637,4 +660,41 @@ export class ClientResponseCache {
}
return this._toolOutputValidatorIndex.byName.get(name) as V | undefined;
}
/** Parse a held `tools/list` document for the index builders; a document
* that does not parse OR whose `tools` is not an array of objects
* (both mean a corrupted external store) is reported and treated as if
* nothing were held. Callers memoize the outcome against the entry's
* stamp, so a corrupt document costs one parse + report per stamp, not
* per lookup. */
private _decodeListTools(entry: CacheEntry): ListToolsResult | undefined {
try {
const parsed = JSON.parse(entry.value) as ListToolsResult | null;
if (!Array.isArray(parsed?.tools) || !parsed.tools.every(t => t !== null && typeof t === 'object')) {
throw new TypeError('cached tools/list document has a malformed tools array');
}
return parsed;
} catch (error) {
this._reportError(error);
return undefined;
}
}
}
/**
* Serialize a result for storage; throws TypeError for anything without a
* JSON representation (legal wire results always have one). Two failure
* modes need folding into that one throw: `JSON.stringify` throws on cycles
* and BigInt, but silently returns `undefined` (not a string) for a
* top-level function, symbol, or `undefined`.
*/
function encodeCacheValue(value: unknown): string {
let json: string | undefined;
try {
json = JSON.stringify(value);
} catch (error) {
throw new TypeError(`cache value is not JSON-serializable: ${error instanceof Error ? error.message : String(error)}`);
}
if (typeof json !== 'string') throw new TypeError('cache value is not JSON-serializable: it has no JSON representation');
return json;
}