fix(cache): do not resurrect a key deleted during getOrSet (#2246)

fluxer_api/pkgs/cache/src/ICacheService.ts

@@ -14,12 +14,21 @@ type CacheTtlSeconds<T> = number | ((value: T) => number);
 
 export abstract class ICacheService {
 	private readonly inflightValues = new Map<string, Promise<unknown>>();
+	private readonly produceInvalidations = new Map<string, number>();
 
 	abstract getEntry<T>(key: string): Promise<CacheLookupResult<T>>;
 
 	abstract set<T>(key: string, value: T, ttlSeconds?: number): Promise<void>;
 
-	abstract delete(key: string): Promise<void>;
+	protected abstract deleteEntry(key: string): Promise<void>;
+
+	async delete(key: string): Promise<void> {
+		const pending = this.produceInvalidations.get(key);
+		if (pending !== undefined) {
+			this.produceInvalidations.set(key, pending + 1);
+		}
+		await this.deleteEntry(key);
+	}
 
 	abstract getAndDelete<T>(key: string): Promise<T | null>;
 
@@ -59,6 +68,32 @@ export abstract class ICacheService {
 	}
 
 	async getOrSet<T>(key: string, valueFactory: () => Promise<T>, ttlSeconds?: CacheTtlSeconds<T>): Promise<T> {
+		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<T>(
+		key: string,
+		valueFactory: () => Promise<T>,
+		ttlSeconds: CacheTtlSeconds<T> | undefined,
+		generation: number,
+	): Promise<T> {
 		const existing = await this.getEntry<T>(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<T>(
 		key: string,
 		valueFactory: () => Promise<T>,
-		ttlSeconds?: CacheTtlSeconds<T>,
+		ttlSeconds: CacheTtlSeconds<T> | undefined,
+		generation: number,
 	): Promise<T> {
 		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;
 	}
 }

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<T>(): {promise: Promise<T>; resolve: (value: T) => void} {
+	let resolve!: (value: T) => void;
+	const promise = new Promise<T>((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<string>();
+		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<string>();
+		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<string>();
+		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');
+	});
+});

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<void> {
+	protected async deleteEntry(key: string): Promise<void> {
 		this.cache.delete(key);
 	}
 

fluxer_api/pkgs/cache/src/providers/KVCacheProvider.ts

@@ -93,7 +93,7 @@ export class KVCacheProvider extends ICacheService {
 		});
 	}
 
-	async delete(key: string): Promise<void> {
+	protected async deleteEntry(key: string): Promise<void> {
 		return this.instrumented('delete', key, async () => {
 			await this.client.del(key);
 		});