fix(stats): harden vocabulary rollups

src/core/services/immersion-tracker-service.test.ts

@@ -559,6 +559,56 @@ test('fresh tracker DB creates lifetime summary tables', async () => {
   }
 });
 
+test('fresh tracker DB skips lexical rollup backfill work', async () => {
+  const dbPath = makeDbPath();
+  let tracker: ImmersionTrackerService | null = null;
+  let backfillRuns = 0;
+
+  try {
+    const Ctor = await loadTrackerCtor();
+    tracker = new Ctor({ dbPath }, {
+      runLexicalRollupBackfillTask: async () => {
+        backfillRuns += 1;
+      },
+    } as never);
+
+    assert.equal(backfillRuns, 0);
+  } finally {
+    tracker?.destroy();
+    cleanupDbPath(dbPath);
+  }
+});
+
+test('tracker starts the injected lexical rollup backfill when it is pending', async () => {
+  const dbPath = makeDbPath();
+  let tracker: ImmersionTrackerService | null = null;
+  let backfillRuns = 0;
+
+  try {
+    const setupDb = new Database(dbPath);
+    const { ensureSchema } = await import('./immersion-tracker/storage');
+    ensureSchema(setupDb);
+    setupDb
+      .prepare(
+        `UPDATE imm_rollup_state SET state_value = '0' WHERE state_key = 'lexical_daily_rollups_ready'`,
+      )
+      .run();
+    setupDb.close();
+
+    const Ctor = await loadTrackerCtor();
+    tracker = new Ctor({ dbPath }, {
+      runLexicalRollupBackfillTask: async () => {
+        backfillRuns += 1;
+      },
+    } as never);
+
+    assert.equal(backfillRuns, 1);
+  } finally {
+    tracker?.destroy();
+    cleanupDbPath(dbPath);
+  }
+});
+
 test('startup backfills lifetime summaries when retained sessions exist but summary tables are empty', async () => {
   const dbPath = makeDbPath();
   let tracker: ImmersionTrackerService | null = null;

src/core/services/immersion-tracker-service.ts

@@ -102,6 +102,7 @@ import {
   type RunVocabularySummaryTask,
 } from './immersion-tracker/vocabulary-summary-worker-runtime';
 import { LexicalRollupWorkerRuntime } from './immersion-tracker/lexical-rollup-worker-runtime';
+import { areLexicalDailyRollupsReady } from './immersion-tracker/lexical-rollups';
 import { DeleteMaintenanceScheduler } from './immersion-tracker/delete-maintenance-scheduler';
 import {
   cleanupDuplicateSubtitleLines,
@@ -418,6 +419,7 @@ export class ImmersionTrackerService {
     knownWords: ReadonlySet<string> | null,
   ) => Promise<VocabularyStatsSummary>;
   private readonly destroyVocabularySummaryRunner: () => void;
+  private readonly runLexicalRollupBackfillTask: () => Promise<void>;
   private readonly destroyLexicalRollupBackfillRunner: () => void;
   private readonly deleteMaintenanceScheduler: DeleteMaintenanceScheduler;
   private flushTimer: ReturnType<typeof setTimeout> | null = null;
@@ -448,6 +450,8 @@ export class ImmersionTrackerService {
       destroyDeleteMaintenanceRunner?: () => void;
       runVocabularySummaryTask?: RunVocabularySummaryTask;
       destroyVocabularySummaryRunner?: () => void;
+      runLexicalRollupBackfillTask?: (dbPath: string) => Promise<void>;
+      destroyLexicalRollupBackfillRunner?: () => void;
     } = {},
   ) {
     this.dbPath = options.dbPath;
@@ -485,8 +489,16 @@ export class ImmersionTrackerService {
         vocabularySummaryRuntime.run(this.dbPath, knownWords);
       this.destroyVocabularySummaryRunner = () => vocabularySummaryRuntime.destroy();
     }
-    const lexicalRollupRuntime = new LexicalRollupWorkerRuntime();
-    this.destroyLexicalRollupBackfillRunner = () => lexicalRollupRuntime.destroy();
+    if (dependencies.runLexicalRollupBackfillTask) {
+      this.runLexicalRollupBackfillTask = () =>
+        dependencies.runLexicalRollupBackfillTask!(this.dbPath);
+      this.destroyLexicalRollupBackfillRunner =
+        dependencies.destroyLexicalRollupBackfillRunner ?? (() => {});
+    } else {
+      const lexicalRollupRuntime = new LexicalRollupWorkerRuntime();
+      this.runLexicalRollupBackfillTask = () => lexicalRollupRuntime.run(this.dbPath);
+      this.destroyLexicalRollupBackfillRunner = () => lexicalRollupRuntime.destroy();
+    }
     const parentDir = path.dirname(this.dbPath);
     if (!fs.existsSync(parentDir)) {
       fs.mkdirSync(parentDir, { recursive: true });
@@ -546,12 +558,14 @@ export class ImmersionTrackerService {
     this.db = new Database(this.dbPath);
     applyPragmas(this.db);
     ensureSchema(this.db);
-    void lexicalRollupRuntime.run(this.dbPath).catch((error: unknown) => {
-      this.logger.warn(
-        'Lexical daily rollup backfill failed; it will retry on next startup',
-        error,
-      );
-    });
+    if (!areLexicalDailyRollupsReady(this.db)) {
+      void this.runLexicalRollupBackfillTask().catch((error: unknown) => {
+        this.logger.warn(
+          'Lexical daily rollup backfill failed; it will retry on next startup',
+          error,
+        );
+      });
+    }
     const reconciledSessions = reconcileStaleActiveSessions(this.db);
     if (reconciledSessions > 0) {
       this.logger.info(

src/core/services/immersion-tracker/lexical-rollup-worker-runtime.test.ts

@@ -67,3 +67,37 @@ test('lexical rollup worker leaves a backfill pending when no worker can start',
     runtime.destroy();
   }
 });
+
+test('lexical rollup worker absorbs termination failures after settling', async () => {
+  let sendMessage: ((message: { ok: boolean }) => void) | null = null;
+  const runtime = new LexicalRollupWorkerRuntime({
+    resolveWorkerPath: () => '/tmp/fake-worker.js',
+    createWorker: async () => ({
+      once(event: string, listener: (value: never) => void) {
+        if (event === 'message') sendMessage = listener as (message: { ok: boolean }) => void;
+        return this;
+      },
+      terminate: async () => {
+        throw new Error('termination failed');
+      },
+    }),
+    warn: () => {},
+  } as never);
+
+  const unhandled: unknown[] = [];
+  const captureUnhandled = (reason: unknown) => unhandled.push(reason);
+  process.on('unhandledRejection', captureUnhandled);
+  try {
+    const task = runtime.run('/tmp/not-used.sqlite');
+    await new Promise((resolve) => setImmediate(resolve));
+    const notify = sendMessage as ((message: { ok: boolean }) => void) | null;
+    assert.ok(notify);
+    notify({ ok: true });
+    await task;
+    await new Promise((resolve) => setImmediate(resolve));
+    assert.deepEqual(unhandled, []);
+  } finally {
+    process.off('unhandledRejection', captureUnhandled);
+    runtime.destroy();
+  }
+});

src/core/services/immersion-tracker/lexical-rollup-worker-runtime.ts

@@ -70,7 +70,7 @@ export class LexicalRollupWorkerRuntime {
         if (settled) return;
         settled = true;
         this.activeWorkers.delete(worker);
-        void worker.terminate();
+        void worker.terminate().catch(() => undefined);
         if (error) reject(error);
         else resolve();
       };
@@ -101,7 +101,9 @@ export class LexicalRollupWorkerRuntime {
   destroy(): void {
     if (this.destroyed) return;
     this.destroyed = true;
-    for (const worker of this.activeWorkers) void worker.terminate();
+    for (const worker of this.activeWorkers) {
+      void worker.terminate().catch(() => undefined);
+    }
     this.activeWorkers.clear();
   }
 }

src/core/services/immersion-tracker/lexical-rollups.test.ts

@@ -3,10 +3,11 @@ import fs from 'node:fs';
 import os from 'node:os';
 import path from 'node:path';
 import test from 'node:test';
-import { getLexicalDailyRollups } from './lexical-rollups';
+import { getLexicalDailyRollups, rebuildLexicalDailyRollups } from './lexical-rollups';
 import { getTrendsDashboard } from './query-trends';
-import { getVocabularyChartData } from './query-lexical';
+import { getVocabularyChartData, replaceStatsExcludedWords } from './query-lexical';
 import { Database } from './sqlite';
+import type { DatabaseSync } from './sqlite';
 import { ensureSchema } from './storage';
 
 function makeDbPath(): string {
@@ -75,6 +76,81 @@ test('vocabulary charts use complete top-word and lexical rollup data', () => {
   }
 });
 
+test('vocabulary charts find full top-word sets beyond excluded and name rows', () => {
+  const dbPath = makeDbPath();
+  const db = new Database(dbPath);
+
+  try {
+    ensureSchema(db);
+    const insertWord = db.prepare(
+      `INSERT INTO imm_words(headword, word, reading, pos2, first_seen, last_seen, frequency)
+       VALUES (?, ?, '', ?, 1700000000, 1700000000, ?)`,
+    );
+    const exclusions = [];
+    for (let index = 0; index < 100; index += 1) {
+      const headword = `語${index}`;
+      insertWord.run(
+        headword,
+        headword,
+        index < 80 && index >= 60 ? '固有名詞' : '一般',
+        100 - index,
+      );
+      if (index < 60) exclusions.push({ headword, word: headword, reading: '' });
+    }
+    replaceStatsExcludedWords(db, exclusions);
+
+    const charts = getVocabularyChartData(db);
+
+    assert.equal(charts.topWords.length, 12);
+    assert.equal(charts.topWords[0]?.headword, '語60');
+    assert.equal(charts.topWordsWithoutNames.length, 12);
+    assert.equal(charts.topWordsWithoutNames[0]?.headword, '語80');
+  } finally {
+    db.close();
+    fs.rmSync(path.dirname(dbPath), { recursive: true, force: true });
+  }
+});
+
+test('vocabulary charts handle exclusion lists above one SQLite variable batch', () => {
+  const dbPath = makeDbPath();
+  const db = new Database(dbPath);
+
+  try {
+    ensureSchema(db);
+    db.prepare(
+      `INSERT INTO imm_words(headword, word, reading, first_seen, last_seen, frequency)
+       VALUES ('語0', '語0', '', 1700000000, 1700000000, 1)`,
+    ).run();
+    const exclusions = Array.from({ length: 10_923 }, (_, index) => ({
+      headword: `語${index}`,
+      word: `語${index}`,
+      reading: '',
+    }));
+    replaceStatsExcludedWords(db, exclusions);
+
+    const charts = getVocabularyChartData(db);
+
+    assert.deepEqual(charts.topWords, []);
+    assert.deepEqual(charts.newWordsTimeline, []);
+  } finally {
+    db.close();
+    fs.rmSync(path.dirname(dbPath), { recursive: true, force: true });
+  }
+});
+
+test('lexical rollup rebuild preserves the original error when rollback also fails', () => {
+  const originalError = new Error('rebuild failed');
+  const db = {
+    exec(sql: string) {
+      if (sql === 'BEGIN IMMEDIATE') return;
+      if (sql === 'ROLLBACK') throw new Error('rollback failed');
+      throw originalError;
+    },
+  } as unknown as DatabaseSync;
+
+  assert.throws(() => rebuildLexicalDailyRollups(db), originalError);
+});
+
 test('trends read historical new-word buckets from lexical rollups', () => {
   const dbPath = makeDbPath();
   const db = new Database(dbPath);

src/core/services/immersion-tracker/lexical-rollups.ts

@@ -153,7 +153,13 @@ export function rebuildLexicalDailyRollups(db: DatabaseSync): void {
     markLexicalDailyRollupsReady(db);
     db.exec('COMMIT');
   } catch (error) {
-    if (transactionStarted) db.exec('ROLLBACK');
+    if (transactionStarted) {
+      try {
+        db.exec('ROLLBACK');
+      } catch {
+        // Preserve the rebuild failure; it is the actionable cause.
+      }
+    }
     throw error;
   }
 }

src/core/services/immersion-tracker/query-lexical.ts

@@ -28,6 +28,9 @@ import {
 
 const VOCABULARY_STATS_FILTER_OVERSAMPLE_FACTOR = 4;
 const VOCABULARY_STATS_FILTER_OVERSAMPLE_MIN = 100;
+const VOCABULARY_CHART_LIMIT = 12;
+const VOCABULARY_CHART_PAGE_SIZE = 100;
+const EXCLUSION_ALIAS_BATCH_SIZE = 300;
 const SENTENCE_SEARCH_DEFAULT_LIMIT = 50;
 const SENTENCE_SEARCH_MAX_LIMIT = 100;
 const KANJI_PATTERN = /\p{Script=Han}/gu;
@@ -179,25 +182,39 @@ export function getVocabularyChartData(db: DatabaseSync): VocabularyChartData {
   );
   const isExcluded = (word: Pick<VocabularyStatsRow, 'headword' | 'word' | 'reading'>): boolean =>
     excludedVocabularyAliases(word).some((alias) => excludedAliases.has(alias));
-  const topWords = getVocabularyStats(db, 48).filter((word) => !isExcluded(word));
+  const topWords = getTopVocabularyChartWords(db, isExcluded);
   const rollups = ready ? getLexicalDailyRollups(db) : [];
   const timeline = new Map(rollups.map((row) => [row.epochDay, { ...row }]));
   if (excludedAliases.size > 0 && ready) {
     const aliases = [...excludedAliases];
-    const placeholders = aliases.map(() => '?').join(', ');
-    const excludedRows = db
-      .prepare(
-        `
-          SELECT headword, word, reading, pos2,
-            ${localEpochDaySql('first_seen')} AS epochDay
-          FROM imm_words
-          WHERE headword IN (${placeholders}) OR word IN (${placeholders}) OR reading IN (${placeholders})
-        `,
-      )
-      .all(...aliases, ...aliases, ...aliases) as Array<
-      Pick<VocabularyStatsRow, 'headword' | 'word' | 'reading' | 'pos2'> & { epochDay: number }
-    >;
-    for (const word of excludedRows) {
+    const excludedRows = new Map<
+      number,
+      Pick<VocabularyStatsRow, 'headword' | 'word' | 'reading' | 'pos2'> & {
+        wordId: number;
+        epochDay: number;
+      }
+    >();
+    for (let offset = 0; offset < aliases.length; offset += EXCLUSION_ALIAS_BATCH_SIZE) {
+      const batch = aliases.slice(offset, offset + EXCLUSION_ALIAS_BATCH_SIZE);
+      const placeholders = batch.map(() => '?').join(', ');
+      const rows = db
+        .prepare(
+          `
+            SELECT id AS wordId, headword, word, reading, pos2,
+              ${localEpochDaySql('first_seen')} AS epochDay
+            FROM imm_words
+            WHERE headword IN (${placeholders}) OR word IN (${placeholders}) OR reading IN (${placeholders})
+          `,
+        )
+        .all(...batch, ...batch, ...batch) as Array<
+        Pick<VocabularyStatsRow, 'headword' | 'word' | 'reading' | 'pos2'> & {
+          wordId: number;
+          epochDay: number;
+        }
+      >;
+      for (const row of rows) excludedRows.set(row.wordId, row);
+    }
+    for (const word of excludedRows.values()) {
       if (!isExcluded(word)) continue;
       const rollup = timeline.get(word.epochDay);
       if (!rollup) continue;
@@ -207,15 +224,16 @@ export function getVocabularyChartData(db: DatabaseSync): VocabularyChartData {
   }
   return {
     ready,
-    topWords: topWords.slice(0, 12).map((word) => ({
+    topWords: topWords.all.map((word) => ({
+      wordId: word.wordId,
+      headword: word.headword,
+      frequency: word.frequency,
+    })),
+    topWordsWithoutNames: topWords.withoutNames.map((word) => ({
       wordId: word.wordId,
       headword: word.headword,
       frequency: word.frequency,
     })),
-    topWordsWithoutNames: topWords
-      .filter((word) => word.pos2 !== '固有名詞')
-      .slice(0, 12)
-      .map((word) => ({ wordId: word.wordId, headword: word.headword, frequency: word.frequency })),
     newWordsTimeline: [...timeline.values()]
       .filter((row) => row.wordCount > 0)
       .map((row) => ({ epochDay: row.epochDay, wordCount: row.wordCount })),
@@ -225,6 +243,40 @@ export function getVocabularyChartData(db: DatabaseSync): VocabularyChartData {
   };
 }
 
+function getTopVocabularyChartWords(
+  db: DatabaseSync,
+  isExcluded: (word: Pick<VocabularyStatsRow, 'headword' | 'word' | 'reading'>) => boolean,
+): { all: VocabularyStatsRow[]; withoutNames: VocabularyStatsRow[] } {
+  const stmt = db.prepare(`
+    SELECT id AS wordId, headword, word, reading,
+      part_of_speech AS partOfSpeech, pos1, pos2, pos3,
+      frequency, frequency_rank AS frequencyRank,
+      first_seen AS firstSeen, last_seen AS lastSeen,
+      0 AS animeCount
+    FROM imm_words
+    ORDER BY frequency DESC, id
+    LIMIT ? OFFSET ?
+  `);
+  const all: VocabularyStatsRow[] = [];
+  const withoutNames: VocabularyStatsRow[] = [];
+  let offset = 0;
+
+  while (all.length < VOCABULARY_CHART_LIMIT || withoutNames.length < VOCABULARY_CHART_LIMIT) {
+    const page = stmt.all(VOCABULARY_CHART_PAGE_SIZE, offset) as VocabularyStatsRow[];
+    if (page.length === 0) break;
+    for (const word of page) {
+      if (!isVocabularyStatsRowVisible(word) || isExcluded(word)) continue;
+      if (all.length < VOCABULARY_CHART_LIMIT) all.push(word);
+      if (word.pos2 !== '固有名詞' && withoutNames.length < VOCABULARY_CHART_LIMIT) {
+        withoutNames.push(word);
+      }
+    }
+    offset += page.length;
+  }
+
+  return { all, withoutNames };
+}
+
 function excludedVocabularyAliases(
   word: Pick<VocabularyStatsRow, 'headword' | 'word' | 'reading'>,
 ): string[] {

src/core/services/immersion-tracker/vocabulary-summary-worker-runtime.test.ts

@@ -49,3 +49,19 @@ test('vocabulary summary worker module resolves in the current layout', () => {
   assert.ok(workerPath, 'expected the vocabulary summary worker module to resolve');
   assert.ok(workerPath.endsWith(__filename.endsWith('.ts') ? '.ts' : '.js'));
 });
+
+test('vocabulary summary worker never falls back to the caller thread', async () => {
+  const runtime = new VocabularySummaryWorkerRuntime({
+    resolveWorkerPath: () => null,
+    warn: () => {},
+  });
+
+  try {
+    await assert.rejects(
+      runtime.run('/tmp/subminer-summary-worker-not-used.sqlite', null),
+      /worker unavailable/i,
+    );
+  } finally {
+    runtime.destroy();
+  }
+});

src/core/services/immersion-tracker/vocabulary-summary-worker-runtime.ts

@@ -2,7 +2,6 @@ import fs from 'node:fs';
 import path from 'node:path';
 import { createLogger } from '../../../logger';
 import type { VocabularyStatsSummary } from './types';
-import { executeVocabularySummaryTask } from './vocabulary-summary-worker';
 
 interface VocabularySummaryWorkerResponse {
   summary?: VocabularyStatsSummary;
@@ -22,7 +21,6 @@ interface VocabularySummaryWorkerRuntimeOptions {
     workerPath: string,
     workerData: { dbPath: string; knownWords: string[] | null },
   ) => Promise<VocabularySummaryWorkerHandle>;
-  executeFallback?: typeof executeVocabularySummaryTask;
   warn?: (message: string, ...meta: unknown[]) => void;
 }
 
@@ -67,13 +65,10 @@ export class VocabularySummaryWorkerRuntime {
     } catch (error) {
       if (this.destroyed) throw new Error('Vocabulary summary worker is shut down');
       (this.options.warn ?? logger.warn)(
-        'Vocabulary summary worker unavailable; running summary on the current thread',
+        'Vocabulary summary worker unavailable; refusing to scan vocabulary on the current thread',
         error,
       );
-      return (this.options.executeFallback ?? executeVocabularySummaryTask)(
-        dbPath,
-        workerData.knownWords,
-      );
+      throw new Error('Vocabulary summary worker unavailable');
     }
 
     if (this.destroyed) {
@@ -88,7 +83,7 @@ export class VocabularySummaryWorkerRuntime {
         if (settled) return;
         settled = true;
         this.activeWorkers.delete(worker);
-        void worker.terminate();
+        void worker.terminate().catch(() => undefined);
         if (result instanceof Error) reject(result);
         else resolve(result);
       };
@@ -122,7 +117,9 @@ export class VocabularySummaryWorkerRuntime {
   destroy(): void {
     if (this.destroyed) return;
     this.destroyed = true;
-    for (const worker of this.activeWorkers) void worker.terminate();
+    for (const worker of this.activeWorkers) {
+      void worker.terminate().catch(() => undefined);
+    }
     this.activeWorkers.clear();
   }
 }