mirror of
https://github.com/ksyasuda/SubMiner.git
synced 2026-08-13 01:55:50 -07:00
fix(stats): drain full write queue before anime merge/move rebuilds
- Replace single flushNow() with drainWriteQueue loop so forced telemetry appended after a full batch isn't left unwritten before merge/move/rebuild summaries recompute - Add dialog a11y to AnimeMergeDialog/LibraryEntryPicker: aria-modal, labelled headings, alert roles for errors, labelled search input, close button labels
This commit is contained in:
@@ -2498,6 +2498,63 @@ test('Jellyfin link repair removes merged leaked anime rows and sanitizes orphan
|
||||
}
|
||||
});
|
||||
|
||||
test('mergeAnime drains a queue larger than one batch before rebuilding summaries', async () => {
|
||||
const dbPath = makeDbPath();
|
||||
let tracker: ImmersionTrackerService | null = null;
|
||||
|
||||
try {
|
||||
const Ctor = await loadTrackerCtor();
|
||||
// A batch size well below the queued write count: one flushNow() pass would
|
||||
// leave the forced telemetry sample, appended last, unwritten.
|
||||
tracker = new Ctor({ dbPath, policy: { batchSize: 2 } });
|
||||
const privateApi = tracker as unknown as {
|
||||
db: DatabaseSync;
|
||||
queue: unknown[];
|
||||
mergeAnime: (targetAnimeId: number, sourceAnimeIds: number[]) => Promise<unknown>;
|
||||
};
|
||||
|
||||
privateApi.db.exec(`
|
||||
INSERT INTO imm_anime (anime_id, normalized_title_key, canonical_title, CREATED_DATE, LAST_UPDATE_DATE)
|
||||
VALUES (1, 'show', 'Show', 1000, 1000), (2, 'show season 1', 'Show Season 1', 1000, 1000);
|
||||
INSERT INTO imm_videos (video_id, video_key, canonical_title, anime_id, source_type, watched, duration_ms, CREATED_DATE, LAST_UPDATE_DATE)
|
||||
VALUES (1, 'local:/tmp/a.mkv', 'A', 1, 1, 0, 1440000, 1000, 1000),
|
||||
(2, 'local:/tmp/b.mkv', 'B', 2, 1, 0, 1440000, 1000, 1000);
|
||||
INSERT INTO imm_sessions (session_id, session_uuid, video_id, started_at_ms, ended_at_ms, status, active_watched_ms, CREATED_DATE, LAST_UPDATE_DATE)
|
||||
VALUES (1, 'drain-session', 2, '1000', '2000', 2, 1000, 1000, 2000);
|
||||
`);
|
||||
|
||||
for (let index = 0; index < 8; index += 1) {
|
||||
(tracker as unknown as { recordWrite: (write: Record<string, unknown>) => void }).recordWrite(
|
||||
{
|
||||
kind: 'subtitleLine',
|
||||
sessionId: 1,
|
||||
videoId: 2,
|
||||
lineIndex: index,
|
||||
segmentStartMs: index * 1000,
|
||||
segmentEndMs: index * 1000 + 900,
|
||||
text: `line ${index}`,
|
||||
wordOccurrences: [],
|
||||
kanjiOccurrences: [],
|
||||
firstSeen: 1000,
|
||||
lastSeen: 2000,
|
||||
},
|
||||
);
|
||||
}
|
||||
assert.ok(privateApi.queue.length > 2, 'expected more queued writes than one batch');
|
||||
|
||||
await privateApi.mergeAnime(1, [2]);
|
||||
|
||||
assert.equal(privateApi.queue.length, 0);
|
||||
const lines = privateApi.db
|
||||
.prepare('SELECT COUNT(*) AS total FROM imm_subtitle_lines WHERE anime_id = 1')
|
||||
.get() as { total: number };
|
||||
assert.equal(Number(lines.total), 8);
|
||||
} finally {
|
||||
tracker?.destroy();
|
||||
cleanupDbPath(dbPath);
|
||||
}
|
||||
});
|
||||
|
||||
test('applies configurable queue, flush, and retention policy', async () => {
|
||||
const dbPath = makeDbPath();
|
||||
let tracker: ImmersionTrackerService | null = null;
|
||||
|
||||
@@ -602,8 +602,7 @@ export class ImmersionTrackerService {
|
||||
}
|
||||
|
||||
async rebuildLifetimeSummaries(): Promise<LifetimeRebuildSummary> {
|
||||
this.flushTelemetry(true);
|
||||
this.flushNow();
|
||||
this.drainWriteQueue('rebuilding lifetime summaries');
|
||||
return rebuildLifetimeSummaryTables(this.db);
|
||||
}
|
||||
|
||||
@@ -772,21 +771,47 @@ export class ImmersionTrackerService {
|
||||
if (pendingVideoId !== undefined) {
|
||||
await this.pendingAnimeMetadataUpdates.get(pendingVideoId);
|
||||
}
|
||||
// Both of these rebuild the lifetime summaries, which recompute from the
|
||||
// database: queued telemetry has to land first or the active session's
|
||||
// watch time is dropped from the merged totals.
|
||||
this.flushTelemetry(true);
|
||||
this.flushNow();
|
||||
// This rebuilds the lifetime summaries, which recompute from the database:
|
||||
// queued writes have to land first or the active session is dropped from
|
||||
// the merged totals.
|
||||
this.drainWriteQueue('merging library entries');
|
||||
return mergeAnimeRecords(this.db, targetAnimeId, sourceAnimeIds);
|
||||
}
|
||||
|
||||
async moveVideoToAnime(videoId: number, targetAnimeId: number): Promise<VideoMoveSummary> {
|
||||
await this.pendingAnimeMetadataUpdates.get(videoId);
|
||||
this.flushTelemetry(true);
|
||||
this.flushNow();
|
||||
this.drainWriteQueue('moving an episode');
|
||||
return moveVideoToAnimeQuery(this.db, videoId, targetAnimeId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist every queued write before a caller recomputes summaries from the
|
||||
* database.
|
||||
*
|
||||
* A single `flushNow()` is not enough: forced telemetry is appended to the
|
||||
* back of the queue while `flushNow()` writes at most `batchSize` entries off
|
||||
* the front, so a busy session leaves the newest sample unwritten. Stops as
|
||||
* soon as a pass makes no progress — a rolled-back batch is pushed back onto
|
||||
* the queue, and looping on that would spin forever.
|
||||
*
|
||||
* Returns false when the queue could not be emptied, in which case the
|
||||
* rebuild runs against a database still missing those writes.
|
||||
*/
|
||||
private drainWriteQueue(context: string): boolean {
|
||||
this.flushTelemetry(true);
|
||||
while (this.queue.length > 0) {
|
||||
const pending = this.queue.length;
|
||||
this.flushNow();
|
||||
if (this.queue.length >= pending) {
|
||||
this.logger.warn(
|
||||
`Immersion tracker queue did not drain before ${context}; summaries may lag by ${this.queue.length} writes`,
|
||||
);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
async reassignAnimeAnilist(
|
||||
animeId: number,
|
||||
info: {
|
||||
|
||||
Reference in New Issue
Block a user