fix(stats): serialize excluded word writes to avoid races

- Queue setExcludedWords writes so a slower in-flight save can't land after and overwrite a newer edit
- Run server-sync listeners independently so one throwing listener doesn't roll back a successful write or block the rest
- Add regression tests for write ordering, listener isolation, coalesced known-word snapshots, and vocabulary hook aggregate refresh/retry
This commit is contained in:
2026-08-17 01:29:11 -07:00
parent 6018f393d2
commit 3c111d2e3b
6 changed files with 470 additions and 25 deletions
@@ -5,3 +5,4 @@ area: stats
- New-word history now uses permanent daily lexical rollups, backfilled in the background and repaired when tracked material is removed or reprocessed; playback writes queue safely during the one-time rebuild and resume afterward.
- Calendar-day chart labels now preserve the recorded local date in time zones west of UTC.
- Vocabulary summary cards and charts refresh automatically after the word exclusion list changes, and failed loads retry with backoff before showing an inline error with a Retry control.
- Rapid exclusion edits no longer race each other; writes are sent in order so a slower earlier save cannot overwrite a newer list.
@@ -5058,6 +5058,7 @@ test('getVocabularySummary coalesces concurrent requests into one worker task',
let tracker: ImmersionTrackerService | null = null;
let taskRuns = 0;
let releaseTask: (() => void) | null = null;
const seenKnownWords: Array<ReadonlySet<string> | null> = [];
const summary = {
uniqueWords: 1,
uniqueWordsWithoutNames: 1,
@@ -5073,8 +5074,9 @@ test('getVocabularySummary coalesces concurrent requests into one worker task',
tracker = new Ctor(
{ dbPath },
{
runVocabularySummaryTask: async () => {
runVocabularySummaryTask: async (_dbPath, knownWords) => {
taskRuns += 1;
seenKnownWords.push(knownWords);
await new Promise<void>((resolve) => {
releaseTask = resolve;
});
@@ -5093,6 +5095,9 @@ test('getVocabularySummary coalesces concurrent requests into one worker task',
assert.deepEqual(await first, summary);
assert.equal(await second, await first);
assert.equal(taskRuns, 1);
// The coalesced caller's known-words set must not replace the snapshot the
// in-flight scan already started with.
assert.deepEqual(seenKnownWords, [null]);
releaseTask = null;
const third = tracker.getVocabularySummary(null);
+88
View File
@@ -6,6 +6,7 @@ import {
initializeExcludedWordsStore,
resetExcludedWordsStoreForTests,
setExcludedWords,
subscribeExcludedWordsServerSync,
} from './useExcludedWords';
import { BASE_URL } from '../lib/api-client';
@@ -199,3 +200,90 @@ test('initializeExcludedWordsStore retries after transient database load failure
resetExcludedWordsStoreForTests();
}
});
test('a failing server-sync listener neither rolls back the write nor blocks other listeners', async () => {
resetExcludedWordsStoreForTests();
const { values: storage, restore } = installLocalStorage();
const originalFetch = globalThis.fetch;
const originalConsoleError = console.error;
console.error = () => {};
globalThis.fetch = (async () =>
new Response(JSON.stringify({ ok: true }), { status: 200 })) as typeof globalThis.fetch;
const notified: string[] = [];
const unsubscribeFirst = subscribeExcludedWordsServerSync(() => {
notified.push('first');
throw new Error('listener exploded');
});
const unsubscribeSecond = subscribeExcludedWordsServerSync(() => {
notified.push('second');
});
try {
const rows = [{ headword: 'する', word: 'する', reading: 'する' }];
await assert.doesNotReject(() => setExcludedWords(rows));
assert.deepEqual(notified, ['first', 'second']);
assert.deepEqual(getExcludedWordsSnapshot(), rows);
assert.equal(storage.get(STORAGE_KEY), JSON.stringify(rows));
} finally {
unsubscribeFirst();
unsubscribeSecond();
globalThis.fetch = originalFetch;
console.error = originalConsoleError;
restore();
resetExcludedWordsStoreForTests();
}
});
test('overlapping writes serialize so an older list cannot overwrite a newer edit', async () => {
resetExcludedWordsStoreForTests();
const { restore } = installLocalStorage();
const originalFetch = globalThis.fetch;
const sentBodies: string[] = [];
let releaseFirst: (() => void) | null = null;
globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => {
sentBodies.push(String(init?.body ?? ''));
if (sentBodies.length === 1) {
await new Promise<void>((resolve) => {
releaseFirst = resolve;
});
}
return new Response(JSON.stringify({ ok: true }), { status: 200 });
}) as typeof globalThis.fetch;
const syncs: string[] = [];
const unsubscribe = subscribeExcludedWordsServerSync(() => {
syncs.push(JSON.stringify(getExcludedWordsSnapshot()));
});
try {
const first = [{ headword: '猫', word: '猫', reading: 'ねこ' }];
const second = [...first, { headword: '犬', word: '犬', reading: 'いぬ' }];
const third = [...second, { headword: '鳥', word: '鳥', reading: 'とり' }];
// The first write reaches the network before the later edits are made.
const firstWrite = setExcludedWords(first);
await new Promise((resolve) => setTimeout(resolve, 0));
const secondWrite = setExcludedWords(second);
const thirdWrite = setExcludedWords(third);
const release = releaseFirst as (() => void) | null;
assert.ok(release, 'expected the first write to be in flight');
release();
await Promise.all([firstWrite, secondWrite, thirdWrite]);
// The in-flight write finishes first, the superseded middle write is
// dropped, and the newest list is the last thing the server is told.
assert.deepEqual(sentBodies, [
JSON.stringify({ words: first }),
JSON.stringify({ words: third }),
]);
assert.deepEqual(getExcludedWordsSnapshot(), third);
assert.equal(syncs.length, 2);
assert.equal(syncs.at(-1), JSON.stringify(third));
} finally {
unsubscribe();
globalThis.fetch = originalFetch;
restore();
resetExcludedWordsStoreForTests();
}
});
+34 -10
View File
@@ -55,6 +55,22 @@ export function subscribeExcludedWordsServerSync(fn: () => void): () => void {
};
}
function notifyServerSync(): void {
// Listener failures are their own concern: one must not roll back a write
// that already succeeded, nor stop the remaining listeners from running.
for (const fn of serverSyncListeners) {
try {
fn();
} catch (error) {
console.error('Excluded words server-sync listener failed', error);
}
}
}
// Full-list writes are serialized so a slow earlier request cannot land after a
// newer one and overwrite it with a stale list.
let writeChain: Promise<void> = Promise.resolve();
function readLocalStorage(): ExcludedWord[] {
if (typeof localStorage === 'undefined') return [];
try {
@@ -112,17 +128,24 @@ export async function setExcludedWords(words: ExcludedWord[]): Promise<void> {
const normalized = dedupeExcludedWords(words);
revision = writeRevision;
applyWords(normalized);
try {
await apiClient.setExcludedWords(normalized);
for (const fn of serverSyncListeners) fn();
} catch (error) {
if (revision === writeRevision) {
revision = previousRevision;
applyWords(previousWords);
const write = writeChain.then(async () => {
// A newer edit already superseded this list and carries the newest state,
// so sending this one would push a stale list to the server.
if (revision !== writeRevision) return;
try {
await apiClient.setExcludedWords(normalized);
} catch (error) {
if (revision === writeRevision) {
revision = previousRevision;
applyWords(previousWords);
}
console.error('Failed to persist excluded words to stats database', error);
throw error;
}
console.error('Failed to persist excluded words to stats database', error);
throw error;
}
notifyServerSync();
});
writeChain = write.catch(() => {});
return write;
}
export function initializeExcludedWordsStore(): Promise<void> {
@@ -167,6 +190,7 @@ export function resetExcludedWordsStoreForTests(): void {
revision = 0;
listeners.clear();
serverSyncListeners.clear();
writeChain = Promise.resolve();
}
function subscribe(fn: () => void): () => void {
+341
View File
@@ -0,0 +1,341 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { Window } from 'happy-dom';
import { act } from 'react';
import { createRoot, type Root } from 'react-dom/client';
import { apiClient } from '../lib/api-client';
import { resetExcludedWordsStoreForTests, setExcludedWords } from './useExcludedWords';
import { useVocabulary } from './useVocabulary';
import type { StatsVocabularyCharts, StatsVocabularySummary } from '../types/stats';
type VocabularyState = ReturnType<typeof useVocabulary>;
function installDom(): () => void {
const previousWindow = globalThis.window;
const previousDocument = globalThis.document;
const previousHTMLElement = globalThis.HTMLElement;
const globals = globalThis as typeof globalThis & { IS_REACT_ACT_ENVIRONMENT?: boolean };
const previousIsReactActEnvironment = globals.IS_REACT_ACT_ENVIRONMENT;
const window = new Window();
Object.defineProperty(globalThis, 'window', { value: window, configurable: true });
Object.defineProperty(globalThis, 'document', { value: window.document, configurable: true });
Object.defineProperty(globalThis, 'HTMLElement', {
value: window.HTMLElement,
configurable: true,
});
globals.IS_REACT_ACT_ENVIRONMENT = true;
return () => {
Object.defineProperty(globalThis, 'window', { value: previousWindow, configurable: true });
Object.defineProperty(globalThis, 'document', { value: previousDocument, configurable: true });
Object.defineProperty(globalThis, 'HTMLElement', {
value: previousHTMLElement,
configurable: true,
});
globals.IS_REACT_ACT_ENVIRONMENT = previousIsReactActEnvironment;
};
}
function installLocalStorage(): () => void {
const previous = Object.getOwnPropertyDescriptor(globalThis, 'localStorage');
const values = new Map<string, string>();
Object.defineProperty(globalThis, 'localStorage', {
configurable: true,
value: {
getItem: (key: string) => values.get(key) ?? null,
setItem: (key: string, value: string) => values.set(key, value),
removeItem: (key: string) => values.delete(key),
},
});
return () => {
if (previous) Object.defineProperty(globalThis, 'localStorage', previous);
else delete (globalThis as { localStorage?: unknown }).localStorage;
};
}
interface FakeClock {
tick: (ms: number) => void;
restore: () => void;
}
/** Bun's `node:test` shim has no `mock.timers`, so the retry clock is faked here. */
function installFakeTimers(): FakeClock {
const originalSetTimeout = globalThis.setTimeout;
const originalClearTimeout = globalThis.clearTimeout;
const timers = new Map<number, { at: number; fn: () => void }>();
let now = 0;
let nextId = 1;
globalThis.setTimeout = ((fn: () => void, delay = 0) => {
const id = nextId;
nextId += 1;
timers.set(id, { at: now + delay, fn });
return id;
}) as unknown as typeof globalThis.setTimeout;
globalThis.clearTimeout = ((id: number) => {
timers.delete(id);
}) as unknown as typeof globalThis.clearTimeout;
return {
tick: (ms: number) => {
now += ms;
const due = [...timers.entries()]
.filter(([, timer]) => timer.at <= now)
.sort(([, a], [, b]) => a.at - b.at);
for (const [id, timer] of due) {
timers.delete(id);
timer.fn();
}
},
restore: () => {
globalThis.setTimeout = originalSetTimeout;
globalThis.clearTimeout = originalClearTimeout;
},
};
}
function summaryFixture(): StatsVocabularySummary {
return {
uniqueWords: 42,
uniqueWordsWithoutNames: 40,
uniqueKanji: 7,
newThisWeek: 3,
newThisWeekWithoutNames: 2,
knownWordCount: 10,
knownWordCountWithoutNames: 9,
};
}
function chartsFixture(overrides: Partial<StatsVocabularyCharts> = {}): StatsVocabularyCharts {
return {
ready: true,
topWords: [{ wordId: 1, headword: '猫', frequency: 5 }],
topWordsWithoutNames: [{ wordId: 1, headword: '猫', frequency: 5 }],
newWordsTimeline: [{ epochDay: 20_000, wordCount: 4 }],
newWordsTimelineWithoutNames: [{ epochDay: 20_000, wordCount: 4 }],
...overrides,
};
}
interface Harness {
state: () => VocabularyState;
flush: () => Promise<void>;
tick: (ms: number) => Promise<void>;
unmount: () => Promise<void>;
teardown: () => void;
}
async function mountHook(): Promise<Harness> {
const uninstallDom = installDom();
const uninstallLocalStorage = installLocalStorage();
const clock = installFakeTimers();
let latest: VocabularyState | null = null;
function Probe() {
latest = useVocabulary();
return null;
}
const container = document.createElement('div');
document.body.append(container);
let root: Root | null = createRoot(container);
await act(async () => {
root!.render(<Probe />);
});
const flush = async (): Promise<void> => {
// Drain promise callbacks without advancing the mocked clock.
await act(async () => {
await Promise.resolve();
await Promise.resolve();
});
};
return {
state: () => {
assert.ok(latest, 'expected the hook to have rendered');
return latest;
},
flush,
tick: async (ms: number) => {
await act(async () => {
clock.tick(ms);
});
await flush();
},
unmount: async () => {
await act(async () => {
root?.unmount();
root = null;
});
},
teardown: () => {
if (root) root.unmount();
clock.restore();
uninstallLocalStorage();
uninstallDom();
resetExcludedWordsStoreForTests();
},
};
}
function stubVocabularyClient(overrides: {
getVocabularySummary: () => Promise<StatsVocabularySummary>;
getVocabularyCharts: () => Promise<StatsVocabularyCharts>;
}): () => void {
const original = {
getVocabulary: apiClient.getVocabulary,
getKanji: apiClient.getKanji,
getKnownWords: apiClient.getKnownWords,
getVocabularySummary: apiClient.getVocabularySummary,
getVocabularyCharts: apiClient.getVocabularyCharts,
setExcludedWords: apiClient.setExcludedWords,
};
apiClient.getVocabulary = async () => [];
apiClient.getKanji = async () => [];
apiClient.getKnownWords = async () => [];
apiClient.setExcludedWords = async () => {};
apiClient.getVocabularySummary = overrides.getVocabularySummary;
apiClient.getVocabularyCharts = overrides.getVocabularyCharts;
return () => Object.assign(apiClient, original);
}
test('aggregate failures retry with backoff, then surface an error that Retry clears', async () => {
const originalConsoleError = console.error;
console.error = () => {};
let summaryCalls = 0;
let failSummary = true;
const restoreClient = stubVocabularyClient({
getVocabularySummary: async () => {
summaryCalls += 1;
if (failSummary) throw new Error('summary unavailable');
return summaryFixture();
},
getVocabularyCharts: async () => chartsFixture(),
});
const harness = await mountHook();
try {
await harness.flush();
assert.equal(summaryCalls, 1);
assert.equal(harness.state().aggregatesError, null, 'no error until retries are exhausted');
// Backoff is 1s, 2s, 4s, 8s across the remaining four attempts.
for (const delayMs of [1_000, 2_000, 4_000, 8_000]) {
await harness.tick(delayMs);
}
assert.equal(summaryCalls, 5, 'retries are bounded at the attempt limit');
assert.match(harness.state().aggregatesError ?? '', /totals failed to load/i);
// Nothing further is scheduled once the limit is reached.
await harness.tick(60_000);
assert.equal(summaryCalls, 5);
failSummary = false;
await act(async () => {
harness.state().refreshAggregates();
});
await harness.flush();
assert.equal(summaryCalls, 6);
assert.equal(harness.state().aggregatesError, null);
assert.deepEqual(harness.state().summary, summaryFixture());
} finally {
harness.teardown();
restoreClient();
console.error = originalConsoleError;
}
});
test('charts poll while the backfill is pending and stop once it is ready', async () => {
let chartCalls = 0;
const restoreClient = stubVocabularyClient({
getVocabularySummary: async () => summaryFixture(),
getVocabularyCharts: async () => {
chartCalls += 1;
return chartsFixture({ ready: chartCalls >= 3 });
},
});
const harness = await mountHook();
try {
await harness.flush();
assert.equal(chartCalls, 1);
assert.equal(harness.state().charts?.ready, false);
await harness.tick(1_000);
assert.equal(chartCalls, 2);
await harness.tick(1_000);
assert.equal(chartCalls, 3);
assert.equal(harness.state().charts?.ready, true);
// A ready result ends the poll.
await harness.tick(60_000);
assert.equal(chartCalls, 3);
} finally {
harness.teardown();
restoreClient();
}
});
test('aggregates refetch after an exclusion edit is acknowledged by the server', async () => {
let summaryCalls = 0;
let chartCalls = 0;
const restoreClient = stubVocabularyClient({
getVocabularySummary: async () => {
summaryCalls += 1;
return summaryFixture();
},
getVocabularyCharts: async () => {
chartCalls += 1;
return chartsFixture();
},
});
const harness = await mountHook();
try {
await harness.flush();
assert.equal(summaryCalls, 1);
assert.equal(chartCalls, 1);
await act(async () => {
await setExcludedWords([{ headword: '猫', word: '猫', reading: 'ねこ' }]);
});
await harness.flush();
assert.equal(summaryCalls, 2, 'totals must not keep counting the excluded word');
assert.equal(chartCalls, 2);
} finally {
harness.teardown();
restoreClient();
}
});
test('pending retries are cancelled when the tab unmounts', async () => {
const originalConsoleError = console.error;
console.error = () => {};
let summaryCalls = 0;
const restoreClient = stubVocabularyClient({
getVocabularySummary: async () => {
summaryCalls += 1;
throw new Error('summary unavailable');
},
getVocabularyCharts: async () => chartsFixture(),
});
const harness = await mountHook();
try {
await harness.flush();
assert.equal(summaryCalls, 1);
await harness.unmount();
await harness.tick(60_000);
assert.equal(summaryCalls, 1, 'no retry may run after unmount');
} finally {
harness.teardown();
restoreClient();
console.error = originalConsoleError;
}
});
-14
View File
@@ -50,17 +50,3 @@ test('useVocabulary loads exact card totals without holding up the vocabulary ta
assert.match(source, /client\s*\.getVocabularySummary\(\)\s*\.then\(/);
assert.match(source, /client\s*\.getVocabularyCharts\(\)/);
});
test('useVocabulary refetches aggregates after server-acknowledged exclusion edits', () => {
const source = fs.readFileSync(VOCABULARY_HOOK_PATH, 'utf8');
assert.match(source, /subscribeExcludedWordsServerSync\(refreshAggregates\)/);
});
test('useVocabulary bounds aggregate retries instead of polling failures forever', () => {
const source = fs.readFileSync(VOCABULARY_HOOK_PATH, 'utf8');
assert.match(source, /AGGREGATE_RETRY_LIMIT/);
assert.match(source, /aggregateRetryDelayMs\(attempt\)/);
assert.match(source, /CHART_BACKFILL_SLOW_POLL_MS/);
});