From afeaddea22543d1471ba674e0ce390117ebc7a76 Mon Sep 17 00:00:00 2001 From: Hampus Date: Mon, 31 Aug 2026 18:44:31 +0200 Subject: [PATCH] fix(cache): do not fan a failed getOrSet out to its joiners (#2260) --- fluxer_api/pkgs/cache/src/ICacheService.ts | 42 +++++++++++---- .../cache/src/__tests__/CacheGetOrSet.test.ts | 39 ++++++++++++-- .../__tests__/CacheInvalidationRace.test.ts | 54 +++++++++++++++++-- 3 files changed, 116 insertions(+), 19 deletions(-) diff --git a/fluxer_api/pkgs/cache/src/ICacheService.ts b/fluxer_api/pkgs/cache/src/ICacheService.ts index 7c59cef91..e1846b1b3 100644 --- a/fluxer_api/pkgs/cache/src/ICacheService.ts +++ b/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 { key: string; @@ -12,6 +13,8 @@ export type CacheLookupResult = {hit: true; value: T} | {hit: false}; type CacheTtlSeconds = number | ((value: T) => number); +type CacheJoinResult = {joined: true; value: T} | {joined: false; error: unknown}; + export abstract class ICacheService { private readonly inflightValues = new Map>(); private readonly produceInvalidations = new Map(); @@ -68,9 +71,26 @@ export abstract class ICacheService { } async getOrSet(key: string, valueFactory: () => Promise, ttlSeconds?: CacheTtlSeconds): Promise { - 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(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(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( + private async joinInflight(inflight: Promise): Promise> { + try { + return {joined: true, value: (await inflight) as T}; + } catch (error) { + return {joined: false, error}; + } + } + + private async produceSingleFlight( key: string, valueFactory: () => Promise, ttlSeconds: CacheTtlSeconds | undefined, generation: number, ): Promise { - const existing = await this.getEntry(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); } diff --git a/fluxer_api/pkgs/cache/src/__tests__/CacheGetOrSet.test.ts b/fluxer_api/pkgs/cache/src/__tests__/CacheGetOrSet.test.ts index 6440b3229..66d478878 100644 --- a/fluxer_api/pkgs/cache/src/__tests__/CacheGetOrSet.test.ts +++ b/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); + }); }); diff --git a/fluxer_api/pkgs/cache/src/__tests__/CacheInvalidationRace.test.ts b/fluxer_api/pkgs/cache/src/__tests__/CacheInvalidationRace.test.ts index 7273c7056..52cfcc4e1 100644 --- a/fluxer_api/pkgs/cache/src/__tests__/CacheInvalidationRace.test.ts +++ b/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(): {promise: Promise; resolve: (value: T) => void} { +function deferred(): {promise: Promise; resolve: (value: T) => void; reject: (error: Error) => void} { let resolve!: (value: T) => void; - const promise = new Promise((r) => { - resolve = r; + let reject!: (error: Error) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; }); - return {promise, resolve}; + return {promise, resolve, reject}; +} + +function flush(): Promise { + 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>> = []; + const factory = async () => { + const gate = deferred(); + 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>> = []; + const factory = async () => { + const gate = deferred(); + 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();