diff --git a/.changeset/serialize-response-cache-mutations.md b/.changeset/serialize-response-cache-mutations.md new file mode 100644 index 0000000000..9fd83ff745 --- /dev/null +++ b/.changeset/serialize-response-cache-mutations.md @@ -0,0 +1,13 @@ +--- +'@modelcontextprotocol/client': patch +--- + +Serialize response-cache mutations for each logical key. A custom +`ResponseCacheStore` may apply `set()` asynchronously; previously, a +`list_changed` or `resources/updated` invalidation could finish its delete +while an earlier write was still pending, allowing that stale write to restore +the entry afterward. + +Writes and invalidations now retain their invocation order per key. An +invalidation removes any earlier delayed write, while a fresh write started +after the invalidation remains cached. diff --git a/packages/client/src/client/responseCache.ts b/packages/client/src/client/responseCache.ts index e6103d34ce..45c0e431eb 100644 --- a/packages/client/src/client/responseCache.ts +++ b/packages/client/src/client/responseCache.ts @@ -284,6 +284,14 @@ export class ClientResponseCache { * has never read therefore cannot grow this map. */ private readonly _evictionGeneration = new Map(); + /** + * Per-logical-key store-mutation tails. Custom stores may apply `set()` + * asynchronously, so an invalidation must run after any earlier write; + * otherwise `delete()` can finish first and the delayed write can restore + * the stale entry. Writes that start after the invalidation queue behind + * its delete, preserving call order without deleting a newer value. + */ + private readonly _mutationTails = new Map>(); /** * `name → Tool` index derived from the cached `tools/list` entry, memoized * against the entry's `stamp` so it re-derives only when the backing entry @@ -398,6 +406,20 @@ export class ClientResponseCache { return shared?.scope === 'public' ? shared : undefined; } + /** Run mutations for one logical cache key in invocation order. */ + private async _mutate(key: string, operation: () => Promise): Promise { + const previous = this._mutationTails.get(key) ?? Promise.resolve(); + const current = previous.then(operation, operation); + this._mutationTails.set(key, current); + try { + await current; + } finally { + if (this._mutationTails.get(key) === current) { + this._mutationTails.delete(key); + } + } + } + /** * Bump the per-method generation (so an in-flight {@linkcode write} for the * same method becomes a no-op) and drop the connected server's two list @@ -416,7 +438,9 @@ export class ClientResponseCache { */ async evict(method: string): Promise { this._evictionGeneration.set(method, (this._evictionGeneration.get(method) ?? 0) + 1); - await this._deleteBoth(method, ''); + const ownPartition = this._partitionFor('private'); + const sharedPartition = this._partitionFor('public'); + await this._mutate(method, () => this._deleteBoth(method, '', ownPartition, sharedPartition)); } /** @@ -424,9 +448,7 @@ export class ClientResponseCache { * `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 { - const ownPartition = this._partitionFor('private'); - const sharedPartition = this._partitionFor('public'); + private async _deleteBoth(method: string, params: string, ownPartition: string, sharedPartition: string): Promise { try { await this._store.delete({ method, params, partition: ownPartition }); } catch (error) { @@ -465,7 +487,9 @@ export class ClientResponseCache { // `resetForReconnect`). const current = this._evictionGeneration.get(gk); if (current !== undefined) this._evictionGeneration.set(gk, current + 1); - await this._deleteBoth(method, params); + const ownPartition = this._partitionFor('private'); + const sharedPartition = this._partitionFor('public'); + await this._mutate(gk, () => this._deleteBoth(method, params, ownPartition, sharedPartition)); } /** @@ -525,30 +549,33 @@ export class ClientResponseCache { capturedGen: number, freshness?: { expiresAt: number; scope: CacheScope; params?: string } ): Promise { - if ((this._evictionGeneration.get(genKey(method, freshness?.params)) ?? 0) !== capturedGen) return; + const gk = genKey(method, freshness?.params); const params = freshness?.params ?? ''; const ownPartition = this._partitionFor('private'); const sharedPartition = this._partitionFor('public'); const partition = (freshness?.scope ?? 'private') === 'public' ? sharedPartition : ownPartition; - try { - await this._store.set( - { method, params, partition }, - { value: encodeCacheValue(value), expiresAt: freshness?.expiresAt, scope: freshness?.scope } - ); - } catch (error) { - this._reportError(error); - } - if (sharedPartition !== ownPartition) { + await this._mutate(gk, async () => { + if ((this._evictionGeneration.get(gk) ?? 0) !== capturedGen) return; try { - await this._store.delete({ - method, - params, - partition: partition === ownPartition ? sharedPartition : ownPartition - }); + await this._store.set( + { method, params, partition }, + { value: encodeCacheValue(value), expiresAt: freshness?.expiresAt, scope: freshness?.scope } + ); } catch (error) { this._reportError(error); } - } + if (sharedPartition !== ownPartition) { + try { + await this._store.delete({ + method, + params, + partition: partition === ownPartition ? sharedPartition : ownPartition + }); + } catch (error) { + this._reportError(error); + } + } + }); } /** @@ -575,7 +602,9 @@ export class ClientResponseCache { return { value: parsed }; } catch (error) { this._reportError(error); - await this._deleteBoth(method, params ?? ''); + const ownPartition = this._partitionFor('private'); + const sharedPartition = this._partitionFor('public'); + await this._deleteBoth(method, params ?? '', ownPartition, sharedPartition); return undefined; } } diff --git a/packages/client/test/client/responseCache.test.ts b/packages/client/test/client/responseCache.test.ts index cec28ac6d8..4d916b9169 100644 --- a/packages/client/test/client/responseCache.test.ts +++ b/packages/client/test/client/responseCache.test.ts @@ -233,6 +233,79 @@ describe('ClientResponseCache', () => { expect(store.get({ method: 'resources/read', params: 'res://a', partition: PRE })).toBeDefined(); }); + it('evict removes a write whose asynchronous store.set was already in flight', async () => { + const backing = new InMemoryResponseCacheStore(); + let releaseSet!: () => void; + const setCanFinish = new Promise(resolve => { + releaseSet = resolve; + }); + let markSetStarted!: () => void; + const setStarted = new Promise(resolve => { + markSetStarted = resolve; + }); + const store: ResponseCacheStore = { + get: key => backing.get(key), + set: async (key, entry) => { + markSetStarted(); + await setCanFinish; + return backing.set(key, entry); + }, + delete: key => backing.delete(key), + evict: method => backing.evict(method), + clear: () => backing.clear() + }; + const cache = new ClientResponseCache(store, true); + const generation = cache.captureGeneration('tools/list'); + + const write = cache.write('tools/list', { tools: [TOOL_A] }, generation); + await setStarted; + // The invalidation starts after set() has been called but before the + // asynchronous backend applies the write. + const eviction = cache.evict('tools/list'); + releaseSet(); + await Promise.all([write, eviction]); + + expect(backing.get({ method: 'tools/list', params: '', partition: PRE })).toBeUndefined(); + }); + + it('evict preserves a fresh write that starts after an asynchronous invalidation', async () => { + const backing = new InMemoryResponseCacheStore(); + let releaseFirstSet!: () => void; + const firstSetCanFinish = new Promise(resolve => { + releaseFirstSet = resolve; + }); + let markFirstSetStarted!: () => void; + const firstSetStarted = new Promise(resolve => { + markFirstSetStarted = resolve; + }); + let setCount = 0; + const store: ResponseCacheStore = { + get: key => backing.get(key), + set: async (key, entry) => { + setCount += 1; + if (setCount === 1) { + markFirstSetStarted(); + await firstSetCanFinish; + } + return backing.set(key, entry); + }, + delete: key => backing.delete(key), + evict: method => backing.evict(method), + clear: () => backing.clear() + }; + const cache = new ClientResponseCache(store, true); + + const staleWrite = cache.write('tools/list', { tools: [TOOL_A] }, cache.captureGeneration('tools/list')); + await firstSetStarted; + const eviction = cache.evict('tools/list'); + const freshWrite = cache.write('tools/list', { tools: [TOOL_B] }, cache.captureGeneration('tools/list')); + releaseFirstSet(); + await Promise.all([staleWrite, eviction, freshWrite]); + + const entry = backing.get({ method: 'tools/list', params: '', partition: PRE }); + expect(JSON.parse(entry!.value)).toEqual({ tools: [TOOL_B] }); + }); + it('evictKey: own-partition store.delete rejecting does not skip the shared-partition delete', async () => { const deleted: string[] = []; const store: ResponseCacheStore = {