fluxerapp/fluxer
1
0
Fork 0
mirror of https://github.com/fluxerapp/fluxer.git synced 2026-09-02 21:04:06 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(cache): stop a timed out produce pinning its tracking entry (#2304)

Browse Source
This commit is contained in:
Hampus
2026-09-01 01:28:12 +02:00
committed by GitHub
parent 0cd12b2f32
commit f1e6e94041
2 changed files with 93 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fluxer_api/pkgs/cache/src/ICacheService.ts
Vendored
+18 -24
View File
@@ -16,6 +16,10 @@ interface CacheProduceTracking {
produces: number;
}
interface CacheProduceAbandonment {
abandoned: boolean;
}
export type CacheLookupResult<T> = {hit: true; value: T} | {hit: false};
type CacheTtlSeconds<T> = 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<T> {
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<T>(produced: Promise<T>, key: string, generation: number, produceTimeoutMs: number): Promise<T> {
private boundProduce<T>(
produced: Promise<T>,
abandonment: CacheProduceAbandonment,
produceTimeoutMs: number,
): Promise<T> {
return new Promise<T>((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<T>,
ttlSeconds: CacheTtlSeconds<T> | undefined,
generation: number,
abandonment: CacheProduceAbandonment,
): Promise<T> {
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;
fluxer_api/pkgs/cache/src/__tests__/CacheInvalidationRace.test.ts
Vendored
+75 -1
View File
@@ -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<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();
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<string>();
@@ -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<string>();
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<ReturnType<typeof deferred<string>>> = [];
const filling: Array<Promise<string>> = [];
for (let index = 0; index < INFLIGHT_OVERFLOW_ENTRIES; index++) {
const gate = deferred<string>();
fillers.push(gate);
filling.push(cache.getOrSet(`filler:${index}`, async () => await gate.promise, 30));
}
await flush();
const stuck = deferred<string>();
const sibling = deferred<string>();
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<string>();
const timers: Array<NodeJS.Timeout> = [];
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<string>();