fix(stats): batch deletes off the main thread

- Keep stats and playback responsive during delete maintenance
- Serialize concurrent deletes and rebuild summaries once
This commit is contained in:
2026-08-11 22:23:38 -07:00
parent 7b0fbdf254
commit 320db591ae
10 changed files with 986 additions and 39 deletions
@@ -50,6 +50,7 @@ import {
updateAnimeAnilistInfo,
upsertCoverArt,
} from '../query-maintenance.js';
import { deleteMaintenanceBatch } from '../query-delete-maintenance.js';
import { getLocalEpochDay } from '../query-shared.js';
import { EVENT_CARD_MINED, EVENT_SUBTITLE_LINE, SOURCE_TYPE_LOCAL } from '../types.js';
@@ -985,3 +986,172 @@ test('split maintenance helpers delete multiple sessions and whole videos with d
cleanupDbPath(dbPath);
}
});
test('delete maintenance batch preserves retained data across overlapping session, video, and anime targets', () => {
const { db, dbPath, stmts } = createDb();
try {
const retainedAnimeId = getOrCreateAnimeRecord(db, {
parsedTitle: 'Retained Anime',
canonicalTitle: 'Retained Anime',
anilistId: null,
titleRomaji: null,
titleEnglish: null,
titleNative: null,
metadataJson: null,
});
const deletedAnimeId = getOrCreateAnimeRecord(db, {
parsedTitle: 'Deleted Anime',
canonicalTitle: 'Deleted Anime',
anilistId: null,
titleRomaji: null,
titleEnglish: null,
titleNative: null,
metadataJson: null,
});
const retainedVideoId = getOrCreateVideoRecord(db, 'local:/tmp/batch-retain.mkv', {
canonicalTitle: 'Batch Retain',
sourcePath: '/tmp/batch-retain.mkv',
sourceUrl: null,
sourceType: SOURCE_TYPE_LOCAL,
});
const deletedVideoId = getOrCreateVideoRecord(db, 'local:/tmp/batch-video.mkv', {
canonicalTitle: 'Batch Video',
sourcePath: '/tmp/batch-video.mkv',
sourceUrl: null,
sourceType: SOURCE_TYPE_LOCAL,
});
const animeVideoId = getOrCreateVideoRecord(db, 'local:/tmp/batch-anime.mkv', {
canonicalTitle: 'Batch Anime',
sourcePath: '/tmp/batch-anime.mkv',
sourceUrl: null,
sourceType: SOURCE_TYPE_LOCAL,
});
for (const [videoId, animeId, episode] of [
[retainedVideoId, retainedAnimeId, 1],
[deletedVideoId, retainedAnimeId, 2],
[animeVideoId, deletedAnimeId, 1],
] as const) {
linkVideoToAnimeRecord(db, videoId, {
animeId,
parsedBasename: `batch-${episode}.mkv`,
parsedTitle: animeId === retainedAnimeId ? 'Retained Anime' : 'Deleted Anime',
parsedSeason: 1,
parsedEpisode: episode,
parserSource: 'test',
parserConfidence: 1,
parseMetadataJson: null,
});
}
const startedAtMs = 1_700_000_000_000;
const deletedSessionId = startSessionRecord(db, retainedVideoId, startedAtMs).sessionId;
const retainedSessionId = startSessionRecord(
db,
retainedVideoId,
startedAtMs + 1_000,
).sessionId;
const videoSessionId = startSessionRecord(db, deletedVideoId, startedAtMs + 2_000).sessionId;
const animeSessionId = startSessionRecord(db, animeVideoId, startedAtMs + 3_000).sessionId;
for (const [sessionId, sessionStartedAtMs] of [
[deletedSessionId, startedAtMs],
[retainedSessionId, startedAtMs + 1_000],
[videoSessionId, startedAtMs + 2_000],
[animeSessionId, startedAtMs + 3_000],
] as const) {
finalizeSessionMetrics(db, sessionId, sessionStartedAtMs);
}
for (const [index, sessionId, videoId, animeId] of [
[1, deletedSessionId, retainedVideoId, retainedAnimeId],
[2, retainedSessionId, retainedVideoId, retainedAnimeId],
[3, videoSessionId, deletedVideoId, retainedAnimeId],
[4, animeSessionId, animeVideoId, deletedAnimeId],
] as const) {
insertWordOccurrence(db, stmts, {
sessionId,
videoId,
animeId,
lineIndex: index,
text: '猫日',
word: { headword: '猫', word: '猫', reading: 'ねこ' },
});
insertKanjiOccurrence(db, stmts, {
sessionId,
videoId,
animeId,
lineIndex: index + 10,
text: '猫日',
kanji: '日',
});
}
const rollupDay = getLocalEpochDay(db, startedAtMs);
const rollupMonth = 202311;
for (const videoId of [retainedVideoId, deletedVideoId, animeVideoId]) {
db.prepare(
`INSERT INTO imm_daily_rollups (
rollup_day, video_id, total_sessions, total_active_min, total_lines_seen,
total_tokens_seen, total_cards, CREATED_DATE, LAST_UPDATE_DATE
) VALUES (?, ?, 99, 99, 99, 99, 99, ?, ?)`,
).run(rollupDay, videoId, startedAtMs, startedAtMs);
db.prepare(
`INSERT INTO imm_monthly_rollups (
rollup_month, video_id, total_sessions, total_active_min, total_lines_seen,
total_tokens_seen, total_cards, CREATED_DATE, LAST_UPDATE_DATE
) VALUES (?, ?, 99, 99, 99, 99, 99, ?, ?)`,
).run(rollupMonth, videoId, startedAtMs, startedAtMs);
}
deleteMaintenanceBatch(db, [
{ kind: 'session', sessionId: deletedSessionId },
{ kind: 'session', sessionId: videoSessionId },
{ kind: 'video', videoId: deletedVideoId },
{ kind: 'video', videoId: animeVideoId },
{ kind: 'anime', animeId: deletedAnimeId },
]);
assert.deepEqual(db.prepare('SELECT session_id FROM imm_sessions').all(), [
{ session_id: retainedSessionId },
]);
assert.deepEqual(db.prepare('SELECT video_id FROM imm_videos').all(), [
{ video_id: retainedVideoId },
]);
assert.deepEqual(db.prepare('SELECT anime_id FROM imm_anime').all(), [
{ anime_id: retainedAnimeId },
]);
assert.equal(
(
db.prepare(`SELECT frequency FROM imm_words WHERE headword = '猫'`).get() as {
frequency: number;
}
).frequency,
1,
);
assert.equal(
(
db.prepare(`SELECT frequency FROM imm_kanji WHERE kanji = '日'`).get() as {
frequency: number;
}
).frequency,
1,
);
assert.deepEqual(
db.prepare('SELECT video_id, total_sessions FROM imm_daily_rollups').all() as Array<{
video_id: number;
total_sessions: number;
}>,
[{ video_id: retainedVideoId, total_sessions: 1 }],
);
assert.deepEqual(
db.prepare('SELECT video_id, total_sessions FROM imm_monthly_rollups').all() as Array<{
video_id: number;
total_sessions: number;
}>,
[{ video_id: retainedVideoId, total_sessions: 1 }],
);
} finally {
db.close();
cleanupDbPath(dbPath);
}
});
@@ -0,0 +1,125 @@
import assert from 'node:assert/strict';
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 { executeDeleteMaintenanceTask } from './delete-maintenance';
import { startSessionRecord } from './session';
import { Database } from './sqlite';
import { applyPragmas, ensureSchema, getOrCreateVideoRecord } from './storage';
test('a delete batch rebuilds lifetime summaries once', () => {
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'subminer-delete-batch-test-'));
const dbPath = path.join(tempDir, 'immersion.sqlite');
let db = new Database(dbPath);
try {
applyPragmas(db);
ensureSchema(db);
const videoId = getOrCreateVideoRecord(db, 'local:/tmp/batch-delete.mkv', {
canonicalTitle: 'Batch Delete',
sourcePath: '/tmp/batch-delete.mkv',
sourceUrl: null,
sourceType: 1,
});
const firstSessionId = startSessionRecord(db, videoId, 1_000).sessionId;
const secondSessionId = startSessionRecord(db, videoId, 2_000).sessionId;
const deletedVideoId = getOrCreateVideoRecord(db, 'local:/tmp/batch-delete-video.mkv', {
canonicalTitle: 'Batch Delete Video',
sourcePath: '/tmp/batch-delete-video.mkv',
sourceUrl: null,
sourceType: 1,
});
startSessionRecord(db, deletedVideoId, 3_000);
db.exec(`
CREATE TABLE delete_rebuild_audit (id INTEGER PRIMARY KEY);
CREATE TRIGGER count_delete_lifetime_rebuild
AFTER UPDATE OF last_rebuilt_ms ON imm_lifetime_global
BEGIN
INSERT INTO delete_rebuild_audit (id) VALUES (NULL);
END;
`);
db.close();
executeDeleteMaintenanceTask(dbPath, {
kind: 'batch',
tasks: [
{ kind: 'session', sessionId: firstSessionId },
{ kind: 'video', videoId: deletedVideoId },
],
});
db = new Database(dbPath);
const audit = db.prepare('SELECT COUNT(*) AS total FROM delete_rebuild_audit').get() as {
total: number;
};
const retainedSession = db
.prepare('SELECT session_id AS sessionId FROM imm_sessions WHERE video_id = ?')
.get(videoId) as { sessionId: number } | null;
const deletedVideo = db
.prepare('SELECT video_id AS videoId FROM imm_videos WHERE video_id = ?')
.get(deletedVideoId) as { videoId: number } | null;
assert.equal(retainedSession?.sessionId, secondSessionId);
assert.equal(deletedVideo, undefined);
assert.equal(
audit.total,
2,
'one rebuild performs exactly its reset and final global summary writes',
);
} finally {
try {
db.close();
} catch {
// The setup connection closes before maintenance runs.
}
fs.rmSync(tempDir, { recursive: true, force: true });
}
});
test(
'compiled delete worker removes data through its separate database connection',
{ skip: path.extname(__filename) !== '.js' },
async () => {
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), 'subminer-delete-worker-test-'));
const dbPath = path.join(tempDir, 'immersion.sqlite');
const runtime = new DeleteMaintenanceWorkerRuntime();
let db = new Database(dbPath);
try {
applyPragmas(db);
ensureSchema(db);
const videoId = getOrCreateVideoRecord(db, 'local:/tmp/worker-delete.mkv', {
canonicalTitle: 'Worker Delete',
sourcePath: '/tmp/worker-delete.mkv',
sourceUrl: null,
sourceType: 1,
});
const firstSessionId = startSessionRecord(db, videoId, 1_000).sessionId;
const secondSessionId = startSessionRecord(db, videoId, 2_000).sessionId;
db.close();
await runtime.run(dbPath, {
kind: 'batch',
tasks: [
{ kind: 'session', sessionId: firstSessionId },
{ kind: 'session', sessionId: secondSessionId },
],
});
db = new Database(dbPath);
const row = db
.prepare('SELECT COUNT(*) AS total FROM imm_sessions WHERE video_id = ?')
.get(videoId) as { total: number };
assert.equal(row.total, 0);
} finally {
runtime.destroy();
try {
db.close();
} catch {
// The setup connection is already closed before the worker starts.
}
fs.rmSync(tempDir, { recursive: true, force: true });
}
},
);
@@ -0,0 +1,74 @@
import { executeDeleteMaintenanceTask, type DeleteMaintenanceTask } from './delete-maintenance';
interface DeleteMaintenanceWorkerResponse {
ok?: unknown;
error?: unknown;
}
export type RunDeleteMaintenanceTask = (
dbPath: string,
task: DeleteMaintenanceTask,
) => Promise<void>;
export class DeleteMaintenanceWorkerRuntime {
private readonly activeWorkers = new Set<import('node:worker_threads').Worker>();
private destroyed = false;
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;
try {
workerThreads = await import('node:worker_threads');
workerPath = require.resolve('./delete-maintenance-worker-thread.js');
} catch {
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) => {
if (settled) return;
settled = true;
this.activeWorkers.delete(worker);
if (error) reject(error);
else resolve();
};
worker.once('message', (message: DeleteMaintenanceWorkerResponse) => {
if (message.ok === true) {
settle();
return;
}
const detail = typeof message.error === 'string' ? message.error : 'unknown worker error';
settle(new Error(`Delete maintenance failed: ${detail}`));
});
worker.once('error', (error) => settle(error));
worker.once('exit', (code) => {
settle(
new Error(
code === 0
? 'Delete maintenance worker exited without a response'
: `Delete maintenance worker exited with code ${code}`,
),
);
});
});
}
destroy(): void {
if (this.destroyed) return;
this.destroyed = true;
for (const worker of this.activeWorkers) {
void worker.terminate();
}
this.activeWorkers.clear();
}
}
@@ -0,0 +1,22 @@
import { parentPort, workerData } from 'node:worker_threads';
import { executeDeleteMaintenanceTask, type DeleteMaintenanceTask } from './delete-maintenance';
interface DeleteMaintenanceWorkerData {
dbPath: string;
task: DeleteMaintenanceTask;
}
if (!parentPort) {
throw new Error('delete maintenance worker missing parent port');
}
const port = parentPort;
const request = workerData as DeleteMaintenanceWorkerData;
try {
executeDeleteMaintenanceTask(request.dbPath, request.task);
port.postMessage({ ok: true });
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
port.postMessage({ ok: false, error: message });
}
@@ -0,0 +1,47 @@
import { Database } from './sqlite';
import { applyPragmas } from './storage';
import { deleteAnime, deleteSession, deleteSessions, deleteVideo } from './query-maintenance';
import {
deleteMaintenanceBatch,
type DeleteMaintenanceOperation,
} from './query-delete-maintenance';
export type { DeleteMaintenanceOperation } from './query-delete-maintenance';
export type DeleteMaintenanceTask =
| DeleteMaintenanceOperation
| { kind: 'batch'; tasks: DeleteMaintenanceOperation[] };
function executeDeleteMaintenanceOperation(
db: InstanceType<typeof Database>,
task: DeleteMaintenanceOperation,
): void {
switch (task.kind) {
case 'session':
deleteSession(db, task.sessionId);
return;
case 'sessions':
deleteSessions(db, task.sessionIds);
return;
case 'video':
deleteVideo(db, task.videoId);
return;
case 'anime':
deleteAnime(db, task.animeId);
return;
}
}
export function executeDeleteMaintenanceTask(dbPath: string, task: DeleteMaintenanceTask): void {
const db = new Database(dbPath);
try {
applyPragmas(db);
if (task.kind === 'batch') {
deleteMaintenanceBatch(db, task.tasks);
return;
}
executeDeleteMaintenanceOperation(db, task);
} finally {
db.close();
}
}
@@ -0,0 +1,136 @@
import type { DatabaseSync } from './sqlite';
import { rebuildLifetimeSummariesInTransaction } from './lifetime';
import { getRollupGroupsForSessions, refreshRollupsForGroupsInTransaction } from './maintenance';
import {
applyLexicalRemovals,
cleanupUnusedCoverArtBlobHash,
deleteSessionsByIds,
makePlaceholders,
planLexicalRemovalsForSessions,
} from './query-shared';
export type DeleteMaintenanceOperation =
| { kind: 'session'; sessionId: number }
| { kind: 'sessions'; sessionIds: number[] }
| { kind: 'video'; videoId: number }
| { kind: 'anime'; animeId: number };
function addOperationTargets(
operations: DeleteMaintenanceOperation[],
sessionIds: Set<number>,
videoIds: Set<number>,
animeIds: Set<number>,
): void {
for (const operation of operations) {
switch (operation.kind) {
case 'session':
sessionIds.add(operation.sessionId);
break;
case 'sessions':
for (const sessionId of operation.sessionIds) sessionIds.add(sessionId);
break;
case 'video':
videoIds.add(operation.videoId);
break;
case 'anime':
animeIds.add(operation.animeId);
break;
}
}
}
function selectIds(db: DatabaseSync, sql: 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]!,
);
}
export function deleteMaintenanceBatch(
db: DatabaseSync,
operations: DeleteMaintenanceOperation[],
): void {
if (operations.length === 0) return;
db.exec('BEGIN IMMEDIATE');
try {
const sessionIds = new Set<number>();
const videoIds = new Set<number>();
const animeIds = new Set<number>();
addOperationTargets(operations, sessionIds, videoIds, animeIds);
const animeIdList = [...animeIds];
for (const videoId of selectIds(
db,
`SELECT video_id FROM imm_videos WHERE anime_id IN (${makePlaceholders(animeIdList)})`,
animeIdList,
'video_id',
)) {
videoIds.add(videoId);
}
const videoIdList = [...videoIds];
for (const sessionId of selectIds(
db,
`SELECT session_id FROM imm_sessions WHERE video_id IN (${makePlaceholders(videoIdList)})`,
videoIdList,
'session_id',
)) {
sessionIds.add(sessionId);
}
const sessionIdList = [...sessionIds];
const lexicalRemovals = planLexicalRemovalsForSessions(db, sessionIdList);
const affectedRollupGroups = getRollupGroupsForSessions(db, sessionIdList).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);
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);
} else {
deleteSessionsByIds(db, sessionIdList);
}
for (const coverBlobHash of coverBlobHashes) {
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);
}
applyLexicalRemovals(db, lexicalRemovals);
rebuildLifetimeSummariesInTransaction(db);
refreshRollupsForGroupsInTransaction(db, affectedRollupGroups);
db.exec('COMMIT');
} catch (error) {
db.exec('ROLLBACK');
throw error;
}
}