diff --git a/fluxer_api/pkgs/cache/src/ICacheService.ts b/fluxer_api/pkgs/cache/src/ICacheService.ts index 878a65e2a..17f6d040f 100644 --- a/fluxer_api/pkgs/cache/src/ICacheService.ts +++ b/fluxer_api/pkgs/cache/src/ICacheService.ts @@ -16,6 +16,10 @@ interface CacheProduceTracking { produces: number; } +interface CacheProduceAbandonment { + abandoned: boolean; +} + export type CacheLookupResult = {hit: true; value: T} | {hit: false}; type CacheTtlSeconds = number | ((value: T) => number); @@ -122,14 +126,6 @@ export abstract class ICacheService { return this.produceInvalidations.get(key)?.generation ?? 0; } - private abandonProduce(key: string, generation: number): void { - this.trackProduce(key); - const tracked = this.produceInvalidations.get(key); - if (tracked?.generation === generation) { - tracked.generation += 1; - } - } - private releaseProduce(key: string): void { const tracked = this.produceInvalidations.get(key); if (!tracked) { @@ -156,10 +152,10 @@ export abstract class ICacheService { generation: number, produceTimeoutMs: number, ): Promise { + const abandonment: CacheProduceAbandonment = {abandoned: false}; const produced = this.boundProduce( - this.produceAndStore(key, valueFactory, ttlSeconds, generation), - key, - generation, + this.produceAndStore(key, valueFactory, ttlSeconds, generation, abandonment), + abandonment, produceTimeoutMs, ); if (this.inflightValues.size >= CACHE_INFLIGHT_MAX_ENTRIES) { @@ -172,27 +168,24 @@ export abstract class ICacheService { return await pending; } - private boundProduce(produced: Promise, key: string, generation: number, produceTimeoutMs: number): Promise { + private boundProduce( + produced: Promise, + abandonment: CacheProduceAbandonment, + produceTimeoutMs: number, + ): Promise { return new Promise((resolve, reject) => { - let abandoned = false; const timer = setTimeout(() => { - abandoned = true; - this.abandonProduce(key, generation); + abandonment.abandoned = true; reject(new Error(CACHE_PRODUCE_TIMEOUT_MESSAGE)); }, produceTimeoutMs); - const settle = () => { - clearTimeout(timer); - if (abandoned) { - this.releaseProduce(key); - } - }; + timer.unref?.(); produced.then( (value) => { - settle(); + clearTimeout(timer); resolve(value); }, (error: unknown) => { - settle(); + clearTimeout(timer); reject(error); }, ); @@ -204,9 +197,10 @@ export abstract class ICacheService { valueFactory: () => Promise, ttlSeconds: CacheTtlSeconds | undefined, generation: number, + abandonment: CacheProduceAbandonment, ): Promise { const value = await valueFactory(); - if (this.currentGeneration(key) === generation) { + if (!abandonment.abandoned && this.currentGeneration(key) === 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 index fdc4c6450..6bc8d5299 100644 --- a/fluxer_api/pkgs/cache/src/__tests__/CacheInvalidationRace.test.ts +++ b/fluxer_api/pkgs/cache/src/__tests__/CacheInvalidationRace.test.ts @@ -1,7 +1,7 @@ // SPDX-License-Identifier: AGPL-3.0-or-later import {InMemoryProvider} from '@pkgs/cache/src/providers/InMemoryProvider'; -import {describe, expect, it} from 'vitest'; +import {describe, expect, it, vi} from 'vitest'; const INFLIGHT_OVERFLOW_ENTRIES = 10000; const PRODUCE_TIMEOUT_MS = 50; @@ -95,6 +95,27 @@ describe('cache invalidation during an in-flight produce', () => { expect(await cache.get('session')).toBe('retried-session'); }); + it('stores the value a retried produce built after the first produce was invalidated', 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(); + await cache.delete('session'); + gates[0].reject(new Error('produce failed')); + await expect(producer).rejects.toThrow('produce failed'); + await flush(); + expect(gates).toHaveLength(2); + gates[1].resolve('retried-session'); + await expect(joiner).resolves.toBe('retried-session'); + expect(await cache.get('session')).toBe('retried-session'); + }); + it('does not resurrect a value deleted after a concurrent caller released its produce', async () => { const cache = new InMemoryProvider(); const gate = deferred(); @@ -207,6 +228,59 @@ describe('cache invalidation during an in-flight produce', () => { expect(await cache.get('session')).toBe('retried-session'); }); + it('releases produce tracking when the factory never settles', async () => { + const cache = new InMemoryProvider(); + const stuck = deferred(); + const pending = cache.getOrSet('session', async () => await stuck.promise, 30, PRODUCE_TIMEOUT_MS); + await expect(pending).rejects.toThrow(PRODUCE_TIMEOUT_MESSAGE); + await flush(); + expect(trackedProduceKeys(cache)).toEqual([]); + }); + + it('stores a sibling produce that succeeded after an overflow produce timed out', async () => { + const cache = new InMemoryProvider(); + const fillers: Array>> = []; + const filling: Array> = []; + for (let index = 0; index < INFLIGHT_OVERFLOW_ENTRIES; index++) { + const gate = deferred(); + fillers.push(gate); + filling.push(cache.getOrSet(`filler:${index}`, async () => await gate.promise, 30)); + } + await flush(); + const stuck = deferred(); + const sibling = deferred(); + const abandoned = cache.getOrSet('session', async () => await stuck.promise, 30, PRODUCE_TIMEOUT_MS); + const succeeding = cache.getOrSet('session', async () => await sibling.promise, 30, PRODUCE_TIMEOUT_MS * 100); + await expect(abandoned).rejects.toThrow(PRODUCE_TIMEOUT_MESSAGE); + sibling.resolve('sibling-produce'); + await expect(succeeding).resolves.toBe('sibling-produce'); + expect(await cache.get('session')).toBe('sibling-produce'); + for (const gate of fillers) { + gate.resolve('filler'); + } + await Promise.all(filling); + }); + + it('does not hold the event loop open while a produce is in flight', async () => { + const cache = new InMemoryProvider(); + const stuck = deferred(); + const timers: Array = []; + const scheduled = globalThis.setTimeout; + const spy = vi.spyOn(globalThis, 'setTimeout').mockImplementation(((handler: () => void, ms?: number) => { + const timer = scheduled(handler, ms); + if (ms === PRODUCE_TIMEOUT_MS) { + timers.push(timer); + } + return timer; + }) as typeof globalThis.setTimeout); + const pending = cache.getOrSet('session', async () => await stuck.promise, 30, PRODUCE_TIMEOUT_MS); + await flush(); + spy.mockRestore(); + expect(timers).toHaveLength(1); + expect(timers[0].hasRef()).toBe(false); + await expect(pending).rejects.toThrow(PRODUCE_TIMEOUT_MESSAGE); + }); + it('keeps a later produce cacheable after an earlier one was invalidated', async () => { const cache = new InMemoryProvider(); const first = deferred();