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

fix(cache): refcount produce tracking so deletes are not lost (#2272)

Browse Source
This commit is contained in:
Hampus
2026-08-31 22:00:48 +02:00
committed by GitHub
parent c4594397e7
commit 32a64fb097
2 changed files with 103 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fluxer_api/pkgs/cache/src/ICacheService.ts
Vendored
+30 -12
View File
@@ -9,6 +9,11 @@ interface CacheMSetEntry<T> {
ttlSeconds?: number;
}
interface CacheProduceTracking {
generation: number;
produces: number;
}
export type CacheLookupResult<T> = {hit: true; value: T} | {hit: false};
type CacheTtlSeconds<T> = number | ((value: T) => number);
@@ -17,7 +22,7 @@ type CacheJoinResult<T> = {joined: true; value: T} | {joined: false; error: unkn
export abstract class ICacheService {
private readonly inflightValues = new Map<string, Promise<unknown>>();
private readonly produceInvalidations = new Map<string, number>();
private readonly produceInvalidations = new Map<string, CacheProduceTracking>();
abstract getEntry<T>(key: string): Promise<CacheLookupResult<T>>;
@@ -26,9 +31,9 @@ export abstract class ICacheService {
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);
const tracked = this.produceInvalidations.get(key);
if (tracked) {
tracked.generation += 1;
}
await this.deleteEntry(key);
}
@@ -89,21 +94,34 @@ export abstract class ICacheService {
if (attempt >= CACHE_INFLIGHT_JOIN_RETRIES) {
throw joined.error;
}
generation = this.trackProduce(key);
generation = this.currentGeneration(key);
}
} finally {
this.releaseProduce(key, generation);
this.releaseProduce(key);
}
}
private trackProduce(key: string): number {
const generation = this.produceInvalidations.get(key) ?? 0;
this.produceInvalidations.set(key, generation);
return generation;
const tracked = this.produceInvalidations.get(key);
if (tracked) {
tracked.produces += 1;
return tracked.generation;
}
this.produceInvalidations.set(key, {generation: 0, produces: 1});
return 0;
}
private releaseProduce(key: string, generation: number): void {
if ((this.produceInvalidations.get(key) ?? 0) === generation) {
private currentGeneration(key: string): number {
return this.produceInvalidations.get(key)?.generation ?? 0;
}
private releaseProduce(key: string): void {
const tracked = this.produceInvalidations.get(key);
if (!tracked) {
return;
}
tracked.produces -= 1;
if (tracked.produces <= 0) {
this.produceInvalidations.delete(key);
}
}
@@ -139,7 +157,7 @@ export abstract class ICacheService {
generation: number,
): Promise<T> {
const value = await valueFactory();
if ((this.produceInvalidations.get(key) ?? 0) === generation) {
if (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
+73
View File
@@ -3,6 +3,8 @@
import {InMemoryProvider} from '@pkgs/cache/src/providers/InMemoryProvider';
import {describe, expect, it} from 'vitest';
const INFLIGHT_OVERFLOW_ENTRIES = 10000;
function deferred<T>(): {promise: Promise<T>; resolve: (value: T) => void; reject: (error: Error) => void} {
let resolve!: (value: T) => void;
let reject!: (error: Error) => void;
@@ -17,6 +19,10 @@ function flush(): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, 0));
}
function trackedProduceKeys(cache: InMemoryProvider): Array<string> {
return [...(cache as unknown as {produceInvalidations: Map<string, unknown>}).produceInvalidations.keys()];
}
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();
@@ -77,6 +83,73 @@ describe('cache invalidation during an in-flight produce', () => {
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>();
const pending = cache.getOrSet('session', async () => await gate.promise, 30);
await flush();
await cache.set('session', 'served-from-cache', 30);
await expect(cache.getOrSet('session', async () => 'unused', 30)).resolves.toBe('served-from-cache');
await cache.delete('session');
gate.resolve('stale-produce');
await expect(pending).resolves.toBe('stale-produce');
expect(await cache.get('session')).toBeNull();
});
it('does not resurrect a value deleted while a second overflow produce was running', 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 first = deferred<string>();
const second = deferred<string>();
const firstProduce = cache.getOrSet('session', async () => await first.promise, 30);
const secondProduce = cache.getOrSet('session', async () => await second.promise, 30);
await flush();
first.resolve('first-produce');
await expect(firstProduce).resolves.toBe('first-produce');
await cache.delete('session');
second.resolve('second-produce');
await expect(secondProduce).resolves.toBe('second-produce');
expect(await cache.get('session')).toBeNull();
for (const gate of fillers) {
gate.resolve('filler');
}
await Promise.all(filling);
});
it('drops produce tracking once the last produce for a key settles', async () => {
const cache = new InMemoryProvider();
for (let index = 0; index < 50; index++) {
const gate = deferred<string>();
const pending = cache.getOrSet(`session:${index}`, async () => await gate.promise, 30);
await cache.delete(`session:${index}`);
gate.resolve('value');
await pending;
}
const shared = deferred<string>();
const producer = cache.getOrSet('shared', async () => await shared.promise, 30);
const joiner = cache.getOrSet('shared', async () => 'unused', 30);
await flush();
await cache.delete('shared');
shared.resolve('shared-value');
await Promise.all([producer, joiner]);
const failing = cache.getOrSet(
'failing',
async () => {
throw new Error('produce failed');
},
30,
);
await expect(failing).rejects.toThrow('produce failed');
expect(trackedProduceKeys(cache)).toEqual([]);
});
it('keeps a later produce cacheable after an earlier one was invalidated', async () => {
const cache = new InMemoryProvider();
const first = deferred<string>();