fix(cache): do not fan a failed getOrSet out to its joiners (#2260)

fluxer_api/pkgs/cache/src/ICacheService.ts

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
 const CACHE_INFLIGHT_MAX_ENTRIES = 10000;
+const CACHE_INFLIGHT_JOIN_RETRIES = 1;
 
 interface CacheMSetEntry<T> {
 	key: string;
@@ -12,6 +13,8 @@ export type CacheLookupResult<T> = {hit: true; value: T} | {hit: false};
 
 type CacheTtlSeconds<T> = number | ((value: T) => number);
 
+type CacheJoinResult<T> = {joined: true; value: T} | {joined: false; error: unknown};
+
 export abstract class ICacheService {
 	private readonly inflightValues = new Map<string, Promise<unknown>>();
 	private readonly produceInvalidations = new Map<string, number>();
@@ -68,9 +71,26 @@ export abstract class ICacheService {
 	}
 
 	async getOrSet<T>(key: string, valueFactory: () => Promise<T>, ttlSeconds?: CacheTtlSeconds<T>): Promise<T> {
-		const generation = this.trackProduce(key);
+		let generation = this.trackProduce(key);
 		try {
-			return await this.getOrSetTracked(key, valueFactory, ttlSeconds, generation);
+			for (let attempt = 0; ; attempt++) {
+				const existing = await this.getEntry<T>(key);
+				if (existing.hit) {
+					return existing.value;
+				}
+				const inflight = this.inflightValues.get(key);
+				if (!inflight) {
+					return await this.produceSingleFlight(key, valueFactory, ttlSeconds, generation);
+				}
+				const joined = await this.joinInflight<T>(inflight);
+				if (joined.joined) {
+					return joined.value;
+				}
+				if (attempt >= CACHE_INFLIGHT_JOIN_RETRIES) {
+					throw joined.error;
+				}
+				generation = this.trackProduce(key);
+			}
 		} finally {
 			this.releaseProduce(key, generation);
 		}
@@ -88,20 +108,20 @@ export abstract class ICacheService {
 		}
 	}
 
-	private async getOrSetTracked<T>(
+	private async joinInflight<T>(inflight: Promise<unknown>): Promise<CacheJoinResult<T>> {
+		try {
+			return {joined: true, value: (await inflight) as T};
+		} catch (error) {
+			return {joined: false, error};
+		}
+	}
+
+	private async produceSingleFlight<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;
-		}
-		const inflight = this.inflightValues.get(key);
-		if (inflight) {
-			return (await inflight) as T;
-		}
 		if (this.inflightValues.size >= CACHE_INFLIGHT_MAX_ENTRIES) {
 			return await this.produceAndStore(key, valueFactory, ttlSeconds, generation);
 		}

fluxer_api/pkgs/cache/src/__tests__/CacheGetOrSet.test.ts

@@ -94,18 +94,49 @@ describe('ICacheService.getOrSet', () => {
 		expect(store.get('absent')).toBe('null');
 	});
 
-	it('rejects every waiter and retries on the next call when the factory fails', async () => {
+	it('rejects every waiter after a single coalesced retry when the factory keeps failing', async () => {
 		const cache = new InMemoryProvider();
 		const failing = vi.fn(async () => {
 			await delay(10);
 			throw new Error('factory failed');
 		});
-		const settled = await Promise.allSettled([cache.getOrSet('key', failing), cache.getOrSet('key', failing)]);
-		expect(settled.map((result) => result.status)).toEqual(['rejected', 'rejected']);
-		expect(failing).toHaveBeenCalledTimes(1);
+		const settled = await Promise.allSettled([
+			cache.getOrSet('key', failing),
+			cache.getOrSet('key', failing),
+			cache.getOrSet('key', failing),
+			cache.getOrSet('key', failing),
+		]);
+		expect(settled.map((result) => result.status)).toEqual(['rejected', 'rejected', 'rejected', 'rejected']);
+		expect(failing).toHaveBeenCalledTimes(2);
 		await expect(cache.exists('key')).resolves.toBe(false);
 		const succeeding = vi.fn(async () => 11);
 		await expect(cache.getOrSet('key', succeeding)).resolves.toBe(11);
 		expect(succeeding).toHaveBeenCalledTimes(1);
 	});
+
+	it('does not fan a transient producer failure out to the callers that joined it', async () => {
+		const cache = new InMemoryProvider();
+		let calls = 0;
+		const factory = vi.fn(async () => {
+			calls += 1;
+			const attempt = calls;
+			await delay(10);
+			if (attempt === 1) {
+				throw new Error('transient failure');
+			}
+			return 11;
+		});
+		const settled = await Promise.allSettled([
+			cache.getOrSet('key', factory),
+			cache.getOrSet('key', factory),
+			cache.getOrSet('key', factory),
+			cache.getOrSet('key', factory),
+		]);
+		expect(settled.map((result) => result.status)).toEqual(['rejected', 'fulfilled', 'fulfilled', 'fulfilled']);
+		expect(settled.filter((result) => result.status === 'fulfilled').map((result) => result.value)).toEqual([
+			11, 11, 11,
+		]);
+		expect(factory).toHaveBeenCalledTimes(2);
+		await expect(cache.get('key')).resolves.toBe(11);
+	});
 });

fluxer_api/pkgs/cache/src/__tests__/CacheInvalidationRace.test.ts

@@ -3,12 +3,18 @@
 import {InMemoryProvider} from '@pkgs/cache/src/providers/InMemoryProvider';
 import {describe, expect, it} from 'vitest';
 
-function deferred<T>(): {promise: Promise<T>; resolve: (value: T) => void} {
+function deferred<T>(): {promise: Promise<T>; resolve: (value: T) => void; reject: (error: Error) => void} {
 	let resolve!: (value: T) => void;
-	const promise = new Promise<T>((r) => {
-		resolve = r;
+	let reject!: (error: Error) => void;
+	const promise = new Promise<T>((res, rej) => {
+		resolve = res;
+		reject = rej;
 	});
-	return {promise, resolve};
+	return {promise, resolve, reject};
+}
+
+function flush(): Promise<void> {
+	return new Promise((resolve) => setTimeout(resolve, 0));
 }
 
 describe('cache invalidation during an in-flight produce', () => {
@@ -31,6 +37,46 @@ describe('cache invalidation during an in-flight produce', () => {
 		expect(await cache.get('session')).toBe('live-session');
 	});
 
+	it('does not resurrect a value deleted while a retried produce was running', async () => {
+		const cache = new InMemoryProvider();
+		const gates: Array<ReturnType<typeof deferred<string>>> = [];
+		const factory = async () => {
+			const gate = deferred<string>();
+			gates.push(gate);
+			return await gate.promise;
+		};
+		const producer = cache.getOrSet('session', factory, 30);
+		const joiner = cache.getOrSet('session', factory, 30);
+		await flush();
+		gates[0].reject(new Error('produce failed'));
+		await expect(producer).rejects.toThrow('produce failed');
+		await flush();
+		expect(gates).toHaveLength(2);
+		await cache.delete('session');
+		gates[1].resolve('fresh-after-delete');
+		await expect(joiner).resolves.toBe('fresh-after-delete');
+		expect(await cache.get('session')).toBeNull();
+	});
+
+	it('stores the value a retried produce built when no invalidation happens', async () => {
+		const cache = new InMemoryProvider();
+		const gates: Array<ReturnType<typeof deferred<string>>> = [];
+		const factory = async () => {
+			const gate = deferred<string>();
+			gates.push(gate);
+			return await gate.promise;
+		};
+		const producer = cache.getOrSet('session', factory, 30);
+		const joiner = cache.getOrSet('session', factory, 30);
+		await flush();
+		gates[0].reject(new Error('produce failed'));
+		await expect(producer).rejects.toThrow('produce failed');
+		await flush();
+		gates[1].resolve('retried-session');
+		await expect(joiner).resolves.toBe('retried-session');
+		expect(await cache.get('session')).toBe('retried-session');
+	});
+
 	it('keeps a later produce cacheable after an earlier one was invalidated', async () => {
 		const cache = new InMemoryProvider();
 		const first = deferred<string>();