From e81f3f7eae84a78ad29ed1e16d0d67cb41c5cec6 Mon Sep 17 00:00:00 2001 From: Hampus Date: Mon, 31 Aug 2026 15:41:05 +0200 Subject: [PATCH] fix(cache): do not resurrect a key deleted during getOrSet (#2246) --- fluxer_api/pkgs/cache/src/ICacheService.ts | 48 +++++++++++++++++-- .../__tests__/CacheInvalidationRace.test.ts | 45 +++++++++++++++++ .../cache/src/providers/InMemoryProvider.ts | 2 +- .../cache/src/providers/KVCacheProvider.ts | 2 +- 4 files changed, 90 insertions(+), 7 deletions(-) create mode 100644 fluxer_api/pkgs/cache/src/__tests__/CacheInvalidationRace.test.ts diff --git a/fluxer_api/pkgs/cache/src/ICacheService.ts b/fluxer_api/pkgs/cache/src/ICacheService.ts index ba43c2c90..7c59cef91 100644 --- a/fluxer_api/pkgs/cache/src/ICacheService.ts +++ b/fluxer_api/pkgs/cache/src/ICacheService.ts @@ -14,12 +14,21 @@ type CacheTtlSeconds = number | ((value: T) => number); export abstract class ICacheService { private readonly inflightValues = new Map>(); + private readonly produceInvalidations = new Map(); abstract getEntry(key: string): Promise>; abstract set(key: string, value: T, ttlSeconds?: number): Promise; - abstract delete(key: string): Promise; + protected abstract deleteEntry(key: string): Promise; + + async delete(key: string): Promise { + const pending = this.produceInvalidations.get(key); + if (pending !== undefined) { + this.produceInvalidations.set(key, pending + 1); + } + await this.deleteEntry(key); + } abstract getAndDelete(key: string): Promise; @@ -59,6 +68,32 @@ export abstract class ICacheService { } async getOrSet(key: string, valueFactory: () => Promise, ttlSeconds?: CacheTtlSeconds): Promise { + const generation = this.trackProduce(key); + try { + return await this.getOrSetTracked(key, valueFactory, ttlSeconds, generation); + } finally { + this.releaseProduce(key, generation); + } + } + + private trackProduce(key: string): number { + const generation = this.produceInvalidations.get(key) ?? 0; + this.produceInvalidations.set(key, generation); + return generation; + } + + private releaseProduce(key: string, generation: number): void { + if ((this.produceInvalidations.get(key) ?? 0) === generation) { + this.produceInvalidations.delete(key); + } + } + + private async getOrSetTracked( + key: string, + valueFactory: () => Promise, + ttlSeconds: CacheTtlSeconds | undefined, + generation: number, + ): Promise { const existing = await this.getEntry(key); if (existing.hit) { return existing.value; @@ -68,9 +103,9 @@ export abstract class ICacheService { return (await inflight) as T; } if (this.inflightValues.size >= CACHE_INFLIGHT_MAX_ENTRIES) { - return await this.produceAndStore(key, valueFactory, ttlSeconds); + return await this.produceAndStore(key, valueFactory, ttlSeconds, generation); } - const pending = this.produceAndStore(key, valueFactory, ttlSeconds).finally(() => { + const pending = this.produceAndStore(key, valueFactory, ttlSeconds, generation).finally(() => { this.inflightValues.delete(key); }); this.inflightValues.set(key, pending); @@ -80,10 +115,13 @@ export abstract class ICacheService { private async produceAndStore( key: string, valueFactory: () => Promise, - ttlSeconds?: CacheTtlSeconds, + ttlSeconds: CacheTtlSeconds | undefined, + generation: number, ): Promise { const value = await valueFactory(); - await this.set(key, value, typeof ttlSeconds === 'function' ? ttlSeconds(value) : ttlSeconds); + if ((this.produceInvalidations.get(key) ?? 0) === generation) { + await this.set(key, value, typeof ttlSeconds === 'function' ? ttlSeconds(value) : ttlSeconds); + } return value; } } diff --git a/fluxer_api/pkgs/cache/src/__tests__/CacheInvalidationRace.test.ts b/fluxer_api/pkgs/cache/src/__tests__/CacheInvalidationRace.test.ts new file mode 100644 index 000000000..7273c7056 --- /dev/null +++ b/fluxer_api/pkgs/cache/src/__tests__/CacheInvalidationRace.test.ts @@ -0,0 +1,45 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +import {InMemoryProvider} from '@pkgs/cache/src/providers/InMemoryProvider'; +import {describe, expect, it} from 'vitest'; + +function deferred(): {promise: Promise; resolve: (value: T) => void} { + let resolve!: (value: T) => void; + const promise = new Promise((r) => { + resolve = r; + }); + return {promise, resolve}; +} + +describe('cache invalidation during an in-flight produce', () => { + it('does not resurrect a value deleted while the factory was running', async () => { + const cache = new InMemoryProvider(); + const gate = deferred(); + const pending = cache.getOrSet('session', async () => await gate.promise, 30); + await cache.delete('session'); + gate.resolve('revoked-session'); + await expect(pending).resolves.toBe('revoked-session'); + expect(await cache.get('session')).toBeNull(); + }); + + it('still stores the value when no invalidation happens', async () => { + const cache = new InMemoryProvider(); + const gate = deferred(); + const pending = cache.getOrSet('session', async () => await gate.promise, 30); + gate.resolve('live-session'); + await pending; + expect(await cache.get('session')).toBe('live-session'); + }); + + it('keeps a later produce cacheable after an earlier one was invalidated', async () => { + const cache = new InMemoryProvider(); + const first = deferred(); + const pending = cache.getOrSet('session', async () => await first.promise, 30); + await cache.delete('session'); + first.resolve('stale'); + await pending; + expect(await cache.get('session')).toBeNull(); + await cache.getOrSet('session', async () => 'fresh', 30); + expect(await cache.get('session')).toBe('fresh'); + }); +}); diff --git a/fluxer_api/pkgs/cache/src/providers/InMemoryProvider.ts b/fluxer_api/pkgs/cache/src/providers/InMemoryProvider.ts index 0e7775820..6514dd594 100644 --- a/fluxer_api/pkgs/cache/src/providers/InMemoryProvider.ts +++ b/fluxer_api/pkgs/cache/src/providers/InMemoryProvider.ts @@ -86,7 +86,7 @@ export class InMemoryProvider extends ICacheService { this.cache.set(key, entry); } - async delete(key: string): Promise { + protected async deleteEntry(key: string): Promise { this.cache.delete(key); } diff --git a/fluxer_api/pkgs/cache/src/providers/KVCacheProvider.ts b/fluxer_api/pkgs/cache/src/providers/KVCacheProvider.ts index a388f849f..8b37a9918 100644 --- a/fluxer_api/pkgs/cache/src/providers/KVCacheProvider.ts +++ b/fluxer_api/pkgs/cache/src/providers/KVCacheProvider.ts @@ -93,7 +93,7 @@ export class KVCacheProvider extends ICacheService { }); } - async delete(key: string): Promise { + protected async deleteEntry(key: string): Promise { return this.instrumented('delete', key, async () => { await this.client.del(key); });