fix(stats): prevent delete maintenance from freezing the UI

- Serialize and coalesce delete requests through a dedicated scheduler
- Chunk large SQLite ID lists to stay below variable limits
This commit is contained in:
2026-08-11 23:11:45 -07:00
parent 320db591ae
commit 675cb33519
10 changed files with 525 additions and 155 deletions
@@ -1087,7 +1087,13 @@ test('delete maintenance batch preserves retained data across overlapping sessio
}
const rollupDay = getLocalEpochDay(db, startedAtMs);
const rollupMonth = 202311;
const rollupMonth = (
db
.prepare(
`SELECT CAST(strftime('%Y%m', CAST(? AS REAL) / 1000, 'unixepoch', 'localtime') AS INTEGER) AS rollupMonth`,
)
.get(startedAtMs) as { rollupMonth: number }
).rollupMonth;
for (const videoId of [retainedVideoId, deletedVideoId, animeVideoId]) {
db.prepare(
`INSERT INTO imm_daily_rollups (
@@ -1155,3 +1161,22 @@ test('delete maintenance batch preserves retained data across overlapping sessio
cleanupDbPath(dbPath);
}
});
test('delete maintenance batch chunks id lists below the SQLite variable limit', () => {
const { db, dbPath } = createDb();
try {
const ids = Array.from({ length: 32_767 }, (_, index) => index + 1);
assert.doesNotThrow(() => {
deleteMaintenanceBatch(db, [
{ kind: 'sessions', sessionIds: ids },
...ids.map((videoId) => ({ kind: 'video' as const, videoId })),
...ids.map((animeId) => ({ kind: 'anime' as const, animeId })),
]);
});
} finally {
db.close();
cleanupDbPath(dbPath);
}
});
@@ -0,0 +1,83 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { DeleteMaintenanceScheduler } from './delete-maintenance-scheduler';
import type { DeleteMaintenanceTask } from './delete-maintenance';
test('scheduler batches same-turn requests and balances busy state', async () => {
const tasks: DeleteMaintenanceTask[] = [];
const states: string[] = [];
const scheduler = new DeleteMaintenanceScheduler({
batchWindowMs: 0,
runTask: async (task) => {
tasks.push(task);
},
onBusy: () => states.push('busy'),
onIdle: () => states.push('idle'),
});
const first = scheduler.enqueue(() => ({ kind: 'session', sessionId: 1 }));
const second = scheduler.enqueue(() => ({ kind: 'sessions', sessionIds: [2, 3] }));
const third = scheduler.enqueue(() => null);
await Promise.all([first, second, third]);
assert.deepEqual(tasks, [
{
kind: 'batch',
tasks: [
{ kind: 'session', sessionId: 1 },
{ kind: 'sessions', sessionIds: [2, 3] },
],
},
]);
assert.deepEqual(states, ['busy', 'idle']);
});
test('scheduler rejects enqueue after destruction without entering busy state', async () => {
let busyCalls = 0;
let runCalls = 0;
const scheduler = new DeleteMaintenanceScheduler({
batchWindowMs: 0,
runTask: async () => {
runCalls += 1;
},
onBusy: () => {
busyCalls += 1;
},
onIdle: () => {},
});
scheduler.destroy();
await assert.rejects(
scheduler.enqueue(() => ({ kind: 'session', sessionId: 1 })),
/shutting down/,
);
assert.equal(busyCalls, 0);
assert.equal(runCalls, 0);
});
test('scheduler serializes batches and rejects requests queued at destruction', async () => {
const releases: Array<() => void> = [];
let activeTasks = 0;
let maxActiveTasks = 0;
const scheduler = new DeleteMaintenanceScheduler({
batchWindowMs: 0,
runTask: async () => {
activeTasks += 1;
maxActiveTasks = Math.max(maxActiveTasks, activeTasks);
await new Promise<void>((resolve) => releases.push(resolve));
activeTasks -= 1;
},
onBusy: () => {},
onIdle: () => {},
});
const first = scheduler.enqueue(() => ({ kind: 'session', sessionId: 1 }));
while (releases.length === 0) await new Promise<void>((resolve) => setTimeout(resolve, 0));
const queued = scheduler.enqueue(() => ({ kind: 'session', sessionId: 2 }));
scheduler.destroy();
await assert.rejects(queued, /shutting down/);
releases[0]?.();
await first;
assert.equal(maxActiveTasks, 1);
});
@@ -0,0 +1,103 @@
import type { DeleteMaintenanceOperation, DeleteMaintenanceTask } from './delete-maintenance';
type ResolveDeleteMaintenanceOperation = () =>
| DeleteMaintenanceOperation
| null
| Promise<DeleteMaintenanceOperation | null>;
interface PendingDeleteMaintenanceRequest {
resolveTask: ResolveDeleteMaintenanceOperation;
resolve: () => void;
reject: (error: unknown) => void;
}
interface DeleteMaintenanceSchedulerOptions {
batchWindowMs: number;
runTask: (task: DeleteMaintenanceTask) => Promise<void>;
onBusy: () => void;
onIdle: () => void;
}
export class DeleteMaintenanceScheduler {
private readonly pendingRequests: PendingDeleteMaintenanceRequest[] = [];
private running = false;
private drainTimer: ReturnType<typeof setTimeout> | null = null;
private pendingTaskCount = 0;
private destroyed = false;
constructor(private readonly options: DeleteMaintenanceSchedulerOptions) {}
enqueue(resolveTask: ResolveDeleteMaintenanceOperation): Promise<void> {
if (this.destroyed) {
return Promise.reject(new Error('Immersion tracker is shutting down'));
}
if (this.pendingTaskCount === 0) this.options.onBusy();
this.pendingTaskCount += 1;
const result = new Promise<void>((resolve, reject) => {
this.pendingRequests.push({ resolveTask, resolve, reject });
this.scheduleDrain();
});
return result.finally(() => {
this.pendingTaskCount -= 1;
if (this.pendingTaskCount === 0) this.options.onIdle();
});
}
destroy(): void {
if (this.destroyed) return;
this.destroyed = true;
if (this.drainTimer) {
clearTimeout(this.drainTimer);
this.drainTimer = null;
}
const error = new Error('Immersion tracker is shutting down');
for (const request of this.pendingRequests.splice(0)) request.reject(error);
}
private scheduleDrain(): void {
if (this.destroyed || this.running || this.drainTimer) return;
this.drainTimer = setTimeout(() => {
this.drainTimer = null;
void this.drain();
}, this.options.batchWindowMs);
}
private async drain(): Promise<void> {
if (this.running || this.pendingRequests.length === 0) return;
this.running = true;
const requests = this.pendingRequests.splice(0);
const runnable: Array<{
request: PendingDeleteMaintenanceRequest;
task: DeleteMaintenanceOperation;
}> = [];
for (const request of requests) {
try {
const task = await request.resolveTask();
if (task) runnable.push({ request, task });
else request.resolve();
} catch (error) {
request.reject(error);
}
}
if (runnable.length > 0) {
const task: DeleteMaintenanceTask =
runnable.length === 1
? runnable[0]!.task
: { kind: 'batch', tasks: runnable.map((entry) => entry.task) };
try {
await this.options.runTask(task);
for (const { request } of runnable) request.resolve();
} catch (error) {
for (const { request } of runnable) request.reject(error);
}
}
this.running = false;
this.scheduleDrain();
}
}
@@ -3,7 +3,10 @@ import fs from 'node:fs';
import os from 'node:os';
import path from 'node:path';
import test from 'node:test';
import { DeleteMaintenanceWorkerRuntime } from './delete-maintenance-worker-runtime';
import {
DeleteMaintenanceWorkerRuntime,
resolveDeleteMaintenanceWorkerPath,
} from './delete-maintenance-worker-runtime';
import { executeDeleteMaintenanceTask } from './delete-maintenance';
import { startSessionRecord } from './session';
import { Database } from './sqlite';
@@ -79,7 +82,7 @@ test('a delete batch rebuilds lifetime summaries once', () => {
test(
'compiled delete worker removes data through its separate database connection',
{ skip: path.extname(__filename) !== '.js' },
{ skip: resolveDeleteMaintenanceWorkerPath() === null },
async () => {
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'subminer-delete-worker-test-'));
const dbPath = path.join(tempDir, 'immersion.sqlite');
@@ -123,3 +126,73 @@ test(
}
},
);
test('worker runtime warns before falling back when no emitted worker is available', async () => {
const warnings: unknown[][] = [];
const fallbackTasks: unknown[] = [];
const runtime = new DeleteMaintenanceWorkerRuntime({
resolveWorkerPath: () => null,
warn: (...args) => warnings.push(args),
executeFallback: (_dbPath, task) => fallbackTasks.push(task),
});
await runtime.run('/tmp/fallback.sqlite', { kind: 'session', sessionId: 1 });
assert.equal(warnings.length, 1);
assert.match(String(warnings[0]?.[0]), /worker unavailable/i);
assert.deepEqual(fallbackTasks, [{ kind: 'session', sessionId: 1 }]);
});
test('worker runtime terminates a worker after successful settlement', async () => {
type Listener = (value: never) => void;
const listeners = new Map<string, Listener>();
let terminateCalls = 0;
const worker = {
once(event: string, listener: Listener) {
listeners.set(event, listener);
return this;
},
terminate: async () => {
terminateCalls += 1;
return 0;
},
};
const runtime = new DeleteMaintenanceWorkerRuntime({
resolveWorkerPath: () => '/tmp/delete-worker.js',
createWorker: async () => worker,
});
const result = runtime.run('/tmp/test.sqlite', { kind: 'session', sessionId: 1 });
await new Promise<void>((resolve) => setTimeout(resolve, 0));
listeners.get('message')?.({ ok: true } as never);
await result;
assert.equal(terminateCalls, 1);
});
test('worker runtime terminates a worker after failed settlement', async () => {
type Listener = (value: never) => void;
const listeners = new Map<string, Listener>();
let terminateCalls = 0;
const worker = {
once(event: string, listener: Listener) {
listeners.set(event, listener);
return this;
},
terminate: async () => {
terminateCalls += 1;
return 0;
},
};
const runtime = new DeleteMaintenanceWorkerRuntime({
resolveWorkerPath: () => '/tmp/delete-worker.js',
createWorker: async () => worker,
});
const result = runtime.run('/tmp/test.sqlite', { kind: 'session', sessionId: 1 });
await new Promise<void>((resolve) => setTimeout(resolve, 0));
listeners.get('error')?.(new Error('worker failed') as never);
await assert.rejects(result, /worker failed/);
assert.equal(terminateCalls, 1);
});
@@ -1,3 +1,6 @@
import fs from 'node:fs';
import path from 'node:path';
import { createLogger } from '../../../logger';
import { executeDeleteMaintenanceTask, type DeleteMaintenanceTask } from './delete-maintenance';
interface DeleteMaintenanceWorkerResponse {
@@ -10,28 +13,63 @@ export type RunDeleteMaintenanceTask = (
task: DeleteMaintenanceTask,
) => Promise<void>;
interface DeleteMaintenanceWorkerHandle {
once(event: 'message', listener: (message: DeleteMaintenanceWorkerResponse) => void): this;
once(event: 'error', listener: (error: Error) => void): this;
once(event: 'exit', listener: (code: number) => void): this;
terminate(): Promise<number>;
}
interface DeleteMaintenanceWorkerRuntimeOptions {
resolveWorkerPath?: () => string | null;
createWorker?: (
workerPath: string,
workerData: { dbPath: string; task: DeleteMaintenanceTask },
) => Promise<DeleteMaintenanceWorkerHandle>;
executeFallback?: typeof executeDeleteMaintenanceTask;
warn?: (message: string, ...meta: unknown[]) => void;
}
export function resolveDeleteMaintenanceWorkerPath(): string | null {
const workerPath = path.join(__dirname, 'delete-maintenance-worker-thread.js');
return fs.existsSync(workerPath) ? workerPath : null;
}
const logger = createLogger('main:immersion-tracker:delete-worker');
export class DeleteMaintenanceWorkerRuntime {
private readonly activeWorkers = new Set<import('node:worker_threads').Worker>();
private readonly activeWorkers = new Set<DeleteMaintenanceWorkerHandle>();
private destroyed = false;
constructor(private readonly options: DeleteMaintenanceWorkerRuntimeOptions = {}) {}
async run(dbPath: string, task: DeleteMaintenanceTask): Promise<void> {
if (this.destroyed) {
throw new Error('Delete maintenance worker is shut down');
}
let workerThreads: typeof import('node:worker_threads');
let workerPath: string;
let worker: DeleteMaintenanceWorkerHandle;
try {
workerThreads = await import('node:worker_threads');
workerPath = require.resolve('./delete-maintenance-worker-thread.js');
} catch {
executeDeleteMaintenanceTask(dbPath, task);
const workerPath = (this.options.resolveWorkerPath ?? resolveDeleteMaintenanceWorkerPath)();
if (!workerPath) throw new Error('Emitted delete-maintenance worker module was not found');
const createWorker =
this.options.createWorker ??
(async (resolvedPath, workerData) => {
const { Worker } = await import('node:worker_threads');
return new Worker(resolvedPath, { workerData });
});
worker = await createWorker(workerPath, { dbPath, task });
} catch (error) {
(this.options.warn ?? logger.warn)(
'Delete maintenance worker unavailable; running maintenance on the current thread',
error,
);
(this.options.executeFallback ?? executeDeleteMaintenanceTask)(dbPath, task);
return;
}
await new Promise<void>((resolve, reject) => {
let settled = false;
const worker = new workerThreads.Worker(workerPath, { workerData: { dbPath, task } });
this.activeWorkers.add(worker);
const settle = (error?: Error) => {
@@ -40,6 +78,7 @@ export class DeleteMaintenanceWorkerRuntime {
this.activeWorkers.delete(worker);
if (error) reject(error);
else resolve();
void worker.terminate();
};
worker.once('message', (message: DeleteMaintenanceWorkerResponse) => {
@@ -7,8 +7,11 @@ import {
deleteSessionsByIds,
makePlaceholders,
planLexicalRemovalsForSessions,
type LexicalRemovalPlan,
} from './query-shared';
const SQLITE_ID_CHUNK_SIZE = 1_000;
export type DeleteMaintenanceOperation =
| { kind: 'session'; sessionId: number }
| { kind: 'sessions'; sessionIds: number[] }
@@ -39,11 +42,65 @@ function addOperationTargets(
}
}
function selectIds(db: DatabaseSync, sql: string, params: number[], column: string): number[] {
function forEachIdChunk(ids: number[], callback: (chunk: number[]) => void): void {
for (let start = 0; start < ids.length; start += SQLITE_ID_CHUNK_SIZE) {
callback(ids.slice(start, start + SQLITE_ID_CHUNK_SIZE));
}
}
function selectIds(
db: DatabaseSync,
buildSql: (placeholders: string) => string,
params: number[],
column: string,
): number[] {
if (params.length === 0) return [];
return (db.prepare(sql).all(...params) as Array<Record<string, number>>).map(
(row) => row[column]!,
);
const ids: number[] = [];
forEachIdChunk(params, (chunk) => {
const rows = db.prepare(buildSql(makePlaceholders(chunk))).all(...chunk) as Array<
Record<string, number>
>;
for (const row of rows) ids.push(row[column]!);
});
return ids;
}
function planLexicalRemovalsInChunks(db: DatabaseSync, sessionIds: number[]): LexicalRemovalPlan {
const combined: LexicalRemovalPlan = { words: [], kanji: [] };
const merge = (target: LexicalRemovalPlan['words'], source: LexicalRemovalPlan['words']) => {
const byId = new Map(target.map((entry) => [entry.id, entry]));
for (const entry of source) {
const existing = byId.get(entry.id);
if (!existing) {
const added = { ...entry };
target.push(added);
byId.set(entry.id, added);
continue;
}
existing.removedFrequency += entry.removedFrequency;
if (
entry.removedFirstSeenMs !== null &&
(existing.removedFirstSeenMs === null ||
entry.removedFirstSeenMs < existing.removedFirstSeenMs)
) {
existing.removedFirstSeenMs = entry.removedFirstSeenMs;
}
if (
entry.removedLastSeenMs !== null &&
(existing.removedLastSeenMs === null ||
entry.removedLastSeenMs > existing.removedLastSeenMs)
) {
existing.removedLastSeenMs = entry.removedLastSeenMs;
}
}
};
forEachIdChunk(sessionIds, (chunk) => {
const plan = planLexicalRemovalsForSessions(db, chunk);
merge(combined.words, plan.words);
merge(combined.kanji, plan.kanji);
});
return combined;
}
export function deleteMaintenanceBatch(
@@ -62,7 +119,7 @@ export function deleteMaintenanceBatch(
const animeIdList = [...animeIds];
for (const videoId of selectIds(
db,
`SELECT video_id FROM imm_videos WHERE anime_id IN (${makePlaceholders(animeIdList)})`,
(placeholders) => `SELECT video_id FROM imm_videos WHERE anime_id IN (${placeholders})`,
animeIdList,
'video_id',
)) {
@@ -72,7 +129,7 @@ export function deleteMaintenanceBatch(
const videoIdList = [...videoIds];
for (const sessionId of selectIds(
db,
`SELECT session_id FROM imm_sessions WHERE video_id IN (${makePlaceholders(videoIdList)})`,
(placeholders) => `SELECT session_id FROM imm_sessions WHERE video_id IN (${placeholders})`,
videoIdList,
'session_id',
)) {
@@ -80,36 +137,43 @@ export function deleteMaintenanceBatch(
}
const sessionIdList = [...sessionIds];
const lexicalRemovals = planLexicalRemovalsForSessions(db, sessionIdList);
const affectedRollupGroups = getRollupGroupsForSessions(db, sessionIdList).filter(
(group) => !videoIds.has(group.videoId),
);
const lexicalRemovals = planLexicalRemovalsInChunks(db, sessionIdList);
const affectedRollupGroups = sessionIdList
.flatMap((_, index) =>
index % SQLITE_ID_CHUNK_SIZE === 0
? getRollupGroupsForSessions(db, sessionIdList.slice(index, index + SQLITE_ID_CHUNK_SIZE))
: [],
)
.filter((group) => !videoIds.has(group.videoId));
const coverBlobHashes = new Set<string>();
if (videoIdList.length > 0) {
const placeholders = makePlaceholders(videoIdList);
const artRows = db
.prepare(
`SELECT cover_blob_hash AS coverBlobHash
FROM imm_media_art
WHERE video_id IN (${placeholders}) AND cover_blob_hash IS NOT NULL`,
)
.all(...videoIdList) as Array<{ coverBlobHash: string }>;
for (const row of artRows) coverBlobHashes.add(row.coverBlobHash);
forEachIdChunk(videoIdList, (chunk) => {
const placeholders = makePlaceholders(chunk);
const artRows = db
.prepare(
`SELECT cover_blob_hash AS coverBlobHash
FROM imm_media_art
WHERE video_id IN (${placeholders}) AND cover_blob_hash IS NOT NULL`,
)
.all(...chunk) as Array<{ coverBlobHash: string }>;
for (const row of artRows) coverBlobHashes.add(row.coverBlobHash);
});
deleteSessionsByIds(db, sessionIdList);
db.prepare(`DELETE FROM imm_subtitle_lines WHERE video_id IN (${placeholders})`).run(
...videoIdList,
);
db.prepare(`DELETE FROM imm_daily_rollups WHERE video_id IN (${placeholders})`).run(
...videoIdList,
);
db.prepare(`DELETE FROM imm_monthly_rollups WHERE video_id IN (${placeholders})`).run(
...videoIdList,
);
db.prepare(`DELETE FROM imm_media_art WHERE video_id IN (${placeholders})`).run(
...videoIdList,
);
db.prepare(`DELETE FROM imm_videos WHERE video_id IN (${placeholders})`).run(...videoIdList);
forEachIdChunk(videoIdList, (chunk) => {
const placeholders = makePlaceholders(chunk);
db.prepare(`DELETE FROM imm_subtitle_lines WHERE video_id IN (${placeholders})`).run(
...chunk,
);
db.prepare(`DELETE FROM imm_daily_rollups WHERE video_id IN (${placeholders})`).run(
...chunk,
);
db.prepare(`DELETE FROM imm_monthly_rollups WHERE video_id IN (${placeholders})`).run(
...chunk,
);
db.prepare(`DELETE FROM imm_media_art WHERE video_id IN (${placeholders})`).run(...chunk);
db.prepare(`DELETE FROM imm_videos WHERE video_id IN (${placeholders})`).run(...chunk);
});
} else {
deleteSessionsByIds(db, sessionIdList);
}
@@ -118,11 +182,13 @@ export function deleteMaintenanceBatch(
cleanupUnusedCoverArtBlobHash(db, coverBlobHash);
}
if (animeIdList.length > 0) {
const placeholders = makePlaceholders(animeIdList);
db.prepare(`DELETE FROM imm_lifetime_anime WHERE anime_id IN (${placeholders})`).run(
...animeIdList,
);
db.prepare(`DELETE FROM imm_anime WHERE anime_id IN (${placeholders})`).run(...animeIdList);
forEachIdChunk(animeIdList, (chunk) => {
const placeholders = makePlaceholders(chunk);
db.prepare(`DELETE FROM imm_lifetime_anime WHERE anime_id IN (${placeholders})`).run(
...chunk,
);
db.prepare(`DELETE FROM imm_anime WHERE anime_id IN (${placeholders})`).run(...chunk);
});
}
applyLexicalRemovals(db, lexicalRemovals);
@@ -490,17 +490,21 @@ export function deleteSessionsByIds(db: DatabaseSync, sessionIds: number[]): voi
return;
}
const placeholders = makePlaceholders(sessionIds);
db.prepare(`DELETE FROM imm_subtitle_lines WHERE session_id IN (${placeholders})`).run(
...sessionIds,
);
db.prepare(`DELETE FROM imm_session_telemetry WHERE session_id IN (${placeholders})`).run(
...sessionIds,
);
db.prepare(`DELETE FROM imm_session_events WHERE session_id IN (${placeholders})`).run(
...sessionIds,
);
db.prepare(`DELETE FROM imm_sessions WHERE session_id IN (${placeholders})`).run(...sessionIds);
const chunkSize = 1_000;
for (let start = 0; start < sessionIds.length; start += chunkSize) {
const chunk = sessionIds.slice(start, start + chunkSize);
const placeholders = makePlaceholders(chunk);
db.prepare(`DELETE FROM imm_subtitle_lines WHERE session_id IN (${placeholders})`).run(
...chunk,
);
db.prepare(`DELETE FROM imm_session_telemetry WHERE session_id IN (${placeholders})`).run(
...chunk,
);
db.prepare(`DELETE FROM imm_session_events WHERE session_id IN (${placeholders})`).run(
...chunk,
);
db.prepare(`DELETE FROM imm_sessions WHERE session_id IN (${placeholders})`).run(...chunk);
}
}
export function toDbMs(ms: number | bigint): bigint {