diff --git a/workers/src/index.ts b/workers/src/index.ts index 2c92b0b..1a3f61f 100644 --- a/workers/src/index.ts +++ b/workers/src/index.ts @@ -113,9 +113,13 @@ async function initWorker() { console.log('[Integrity] Starting integrity sweep'); const integrityService = new IntegrityService(pgPool, queue); const summary = await integrityService.runSweep(); - console.log( - `[Integrity] Sweep complete: detected=${summary.detected} fixed=${summary.fixed} needsReview=${summary.needsReview}` - ); + if (summary.aborted) { + console.error(`[Integrity] Sweep ABORTED by safety guard: ${summary.abortReason}`); + } else { + console.log( + `[Integrity] Sweep complete: detected=${summary.detected} fixed=${summary.fixed} needsReview=${summary.needsReview}` + ); + } break; } case 'cleanup_sweep': { diff --git a/workers/src/integrity.service.ts b/workers/src/integrity.service.ts index 9866c03..6fb0d86 100644 --- a/workers/src/integrity.service.ts +++ b/workers/src/integrity.service.ts @@ -5,6 +5,24 @@ import { ScannerService } from './scanner.service.js'; const MUSIC_DIR = process.env.MUSIC_DIR || '/mnt/hdd1/media/Music'; +// Safety rails for the missing-file pass. `MISSING` is written by an automated +// 03:00 cron and there is no automatic path back, so an unmounted /mnt/hdd1 +// would otherwise flip the ENTIRE library to MISSING in one sweep. +// +// INTEGRITY_MISSING_ABORT_PCT: if more than this share of the tracks actually +// checked are unreachable, the pass writes nothing and logs an error. A genuine +// bulk deletion is rare and is better handled by an explicit re-scan than by a +// silent cron; a vanished mount is common and catastrophic. +const MISSING_ABORT_PCT = Number(process.env.INTEGRITY_MISSING_ABORT_PCT ?? 10); +// Below this many checked tracks the percentage is statistically meaningless +// (1 of 3 missing is 33%), so the threshold is not applied. +const MISSING_ABORT_MIN_SAMPLE = Number( + process.env.INTEGRITY_MISSING_ABORT_MIN_SAMPLE ?? 20 +); +// Tracks examined per sweep. The sweep is paginated by a persisted keyset +// cursor, so consecutive runs deterministically cover the whole library. +const MISSING_PAGE_SIZE = Number(process.env.INTEGRITY_MISSING_PAGE_SIZE ?? 5000); + // Detection regexes for the known (now-fixed) corruption bug. These are the // exact patterns used by the original one-off repair script: // title : trailing ' (Enriched)' (possibly stacked) @@ -31,6 +49,10 @@ export interface SweepSummary { detected: number; fixed: number; needsReview: number; + /** True when a guard stopped the sweep before it wrote anything. */ + aborted?: boolean; + /** Human-readable reason when `aborted` is set. */ + abortReason?: string; } export class IntegrityService { @@ -57,6 +79,64 @@ export class IntegrityService { UNIQUE(track_id, issue_type) )` ); + // Persisted keyset cursor for the paginated missing-file pass, so the sweep + // resumes where the previous run stopped instead of re-checking the same + // arbitrary 5,000 rows forever. + await this.pgClient.query( + `CREATE TABLE IF NOT EXISTS integrity_sweep_state ( + key TEXT PRIMARY KEY, + cursor_path TEXT, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + )` + ); + } + + /** + * Verify MUSIC_DIR actually looks like the mounted library before any pass + * that treats "file not on disk" as authoritative. + * + * An unmounted bind leaves either a missing path or an EMPTY directory — + * both indistinguishable, from `fs.access` on individual tracks, from "every + * file was deleted". Requiring at least one entry catches the empty-mountpoint + * case, which is the one that would otherwise destroy the library's state. + */ + private async checkMusicDirLive(): Promise<{ live: boolean; reason?: string }> { + try { + const st = await fs.stat(MUSIC_DIR); + if (!st.isDirectory()) { + return { live: false, reason: `${MUSIC_DIR} exists but is not a directory` }; + } + } catch (err) { + return { live: false, reason: `${MUSIC_DIR} is not accessible: ${(err as Error).message}` }; + } + try { + const entries = await fs.readdir(MUSIC_DIR); + if (entries.length === 0) { + return { + live: false, + reason: `${MUSIC_DIR} is empty — the library bind mount is almost certainly not mounted`, + }; + } + } catch (err) { + return { live: false, reason: `${MUSIC_DIR} is not readable: ${(err as Error).message}` }; + } + return { live: true }; + } + + private async getSweepCursor(): Promise { + const res = await this.pgClient.query<{ cursor_path: string | null }>( + `SELECT cursor_path FROM integrity_sweep_state WHERE key = 'missing_files'` + ); + return res.rows[0]?.cursor_path ?? null; + } + + private async setSweepCursor(cursor: string | null): Promise { + await this.pgClient.query( + `INSERT INTO integrity_sweep_state (key, cursor_path, updated_at) + VALUES ('missing_files', $1, NOW()) + ON CONFLICT (key) DO UPDATE SET cursor_path = EXCLUDED.cursor_path, updated_at = NOW()`, + [cursor] + ); } /** Tracks whose title or artist still carry the corruption markers. */ @@ -70,14 +150,28 @@ export class IntegrityService { return res.rows; } - /** Tracks (excluding ones already flagged DELETED) whose file is gone from disk. */ - async detectMissingFiles(limit = 5000): Promise { + /** + * Tracks (excluding ones already flagged DELETED) whose file is gone from disk. + * + * Paginated by keyset on `path` (UNIQUE, hence a total order): each call takes + * the next `limit` rows after `afterPath`. Callers persist the returned + * `nextCursor` so successive sweeps walk the whole library and wrap around, + * instead of re-checking an arbitrary unordered `LIMIT 5000` forever. + * + * Returns the checked count too, so the caller can apply a ratio guard. + */ + async detectMissingFiles( + limit = MISSING_PAGE_SIZE, + afterPath: string | null = null + ): Promise<{ missing: MissingRow[]; checked: number; nextCursor: string | null }> { const res = await this.pgClient.query( `SELECT id, path, title, artist FROM tracks WHERE state <> 'DELETED' + AND ($2::text IS NULL OR path > $2) + ORDER BY path LIMIT $1`, - [limit] + [limit, afterPath] ); const missing: MissingRow[] = []; @@ -88,7 +182,13 @@ export class IntegrityService { missing.push(row); } } - return missing; + + // Short page (or empty) means we reached the end: wrap to the start so the + // next sweep begins a fresh cycle. + const nextCursor = + res.rows.length < limit ? null : res.rows[res.rows.length - 1].path; + + return { missing, checked: res.rows.length, nextCursor }; } /** Upsert an issue row keyed on (track_id, issue_type), refreshing on re-detection. */ @@ -137,6 +237,22 @@ export class IntegrityService { const summary: SweepSummary = { detected: 0, fixed: 0, needsReview: 0 }; + // --- Liveness guard ------------------------------------------------------ + // Both passes treat the filesystem as authoritative (PASS 1 re-scans it, + // the missing-file pass writes state='MISSING' from it). If the library + // mount is gone, every conclusion the sweep draws is wrong and there is no + // automatic way back, so refuse to run at all. + const live = await this.checkMusicDirLive(); + if (!live.live) { + summary.aborted = true; + summary.abortReason = live.reason; + console.error( + `[Integrity] ABORTING sweep: music library not available — ${live.reason}. ` + + `No tracks were examined and nothing was written.` + ); + return summary; + } + // --- Corrupted metadata -------------------------------------------------- const corrupt = await this.detectCorruptedMetadata(); summary.detected = corrupt.length; @@ -185,7 +301,34 @@ export class IntegrityService { } // --- Missing files ------------------------------------------------------- - const missing = await this.detectMissingFiles(); + const cursor = await this.getSweepCursor(); + const { missing, checked, nextCursor } = await this.detectMissingFiles( + MISSING_PAGE_SIZE, + cursor + ); + + // Ratio guard: a large fraction of the page being unreachable means the + // filesystem, not the library, changed (partially-mounted disk, permission + // loss, a moved music root). Write nothing and leave the cursor untouched so + // the same page is re-examined once the cause is fixed. + const missingPct = checked > 0 ? (missing.length / checked) * 100 : 0; + if ( + checked >= MISSING_ABORT_MIN_SAMPLE && + missingPct > MISSING_ABORT_PCT + ) { + summary.aborted = true; + summary.abortReason = + `${missing.length}/${checked} checked tracks unreachable ` + + `(${missingPct.toFixed(1)}% > ${MISSING_ABORT_PCT}% threshold)`; + console.error( + `[Integrity] ABORTING missing-file pass: ${summary.abortReason}. ` + + `Nothing was marked MISSING and the sweep cursor was not advanced. ` + + `Verify ${MUSIC_DIR} is fully mounted, then re-run; raise ` + + `INTEGRITY_MISSING_ABORT_PCT only if this bulk removal is genuine.` + ); + return summary; + } + for (const row of missing) { summary.detected++; summary.needsReview++; @@ -203,8 +346,12 @@ export class IntegrityService { ); } + await this.setSweepCursor(nextCursor); + console.log( - `[Integrity] Sweep summary: detected=${summary.detected} fixed=${summary.fixed} needsReview=${summary.needsReview}` + `[Integrity] Sweep summary: detected=${summary.detected} fixed=${summary.fixed} ` + + `needsReview=${summary.needsReview} checked=${checked} missing=${missing.length} ` + + `nextCursor=${nextCursor === null ? '' : nextCursor}` ); return summary; } diff --git a/workers/src/scripts/repair-corrupted-metadata.ts b/workers/src/scripts/repair-corrupted-metadata.ts index 2710739..447264f 100644 --- a/workers/src/scripts/repair-corrupted-metadata.ts +++ b/workers/src/scripts/repair-corrupted-metadata.ts @@ -29,9 +29,13 @@ async function main() { const integrity = new IntegrityService(pgClient, queue); const summary = await integrity.runSweep(); - console.log( - `[Repair] Done: detected=${summary.detected} fixed=${summary.fixed} needsReview=${summary.needsReview}` - ); + if (summary.aborted) { + console.error(`[Repair] Sweep ABORTED by safety guard: ${summary.abortReason}`); + } else { + console.log( + `[Repair] Done: detected=${summary.detected} fixed=${summary.fixed} needsReview=${summary.needsReview}` + ); + } await pgClient.end(); console.log('[Repair] Connection closed.');