diff --git a/CHANGELOG.MD b/CHANGELOG.MD index c60803b..52755bd 100644 --- a/CHANGELOG.MD +++ b/CHANGELOG.MD @@ -5,6 +5,59 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0). +## [2.6.64] - 2026-07-09 + +### Added + +- **Rolling-integrity coverage is now visible.** Because budgeted integrity runs sweep the library in slices, no single run report answers "has every file been verified?". `/api/stats` now returns an `integrity` block - files checked ever and within the last 30 days, percent of library covered, never-checked count, the oldest check date (every checked file has been verified at least once since then), and the current bitrot-suspected count. The dashboard gains an "Integrity Checked" stat card showing the coverage percentage, with the full breakdown in its tooltip. + +## [2.6.63] - 2026-07-09 + +### Fixed + +- **Created or edited schedules now take effect within a minute instead of requiring a container restart.** Schedule CRUD dispatched `reload_schedules_task` through Celery, but prefork workers execute tasks in pool children while APScheduler runs only in the scheduler-lock-holding main process - the reload always landed in a child, logged "Scheduler not running in this worker", and was skipped, so no schedule change since the mechanism was added ever reached the running scheduler without a restart (verified live: a freshly created `file_changes` schedule never fired at its cron time). The scheduler process now re-syncs its jobs directly from the database every 60 seconds, gated on a fingerprint of the schedule definitions so jobs are only rebuilt when something actually changed. New schedules also get `next_run` computed at creation so the UI shows the upcoming fire immediately instead of after the first run. + +## [2.6.62] - 2026-07-09 + +### Added + +- **Bitrot UI.** Results list shows a distinct "Bitrot?" badge (violet) taking precedence over corrupted/warning; a "Bitrot Suspected" filter button joins the corrupted/warning/healthy set (`bitrot_suspected=true` on `/api/scan-results`); the file details modal gains a bitrot section (detection date, baseline vs current hash/mtime/size, stable-check count) with an "Accept Current State" action; the same accept action appears in each flagged file's Actions dropdown (desktop and mobile), confirming before adopting the current content as the new baseline via `POST /api/bitrot/accept`. + +## [2.6.61] - 2026-07-09 + +### Added + +- **The integrity check now distinguishes bitrot from legitimate modification.** Previously any hash mismatch was uniformly marked `modified` and rescanned - and the rescan adopted the new content's hash as baseline, silently laundering corruption (a bit flip in non-critical stream data passes decode checks). The hash task now classifies: hash mismatch with CHANGED mtime = `modified` (normal edit/replace, rescans as before); hash mismatch with UNCHANGED mtime (2-second epsilon) = `bitrot_suspected` - no legitimate write path alters content without touching mtime. Suspected files are flagged (`bitrot_suspected` + permanent `bitrot_detected_date`/`bitrot_details`, first detection date preserved across re-detections), one aggregated high-priority `bitrot_suspected` notification fires per run (a dying disk flags files in bulk; per-file dispatch would mean thousands of serial HTTP round-trips), the file is rescanned for decodability, and critically NO scan path overwrites the stored hash/mtime baseline for flagged files: the guard lives in a single shared writer (`apply_scan_baseline`) used by both the serial scan save and the parallel chunk workers that perform all normal-scan and rescan persistence. +- **Auto-expire state machine for flagged files.** Flagged files jump the integrity queue. Each subsequent check compares against the candidate hash: stable content for `BITROT_STABLE_CHECKS_TO_EXPIRE` consecutive checks (default 2) plus a clean rescan auto-adopts the content as the new baseline and clears the flag (detection record kept permanently - a per-disk history of "files that ever tripped bitrot" is a dying-disk signal). Content matching the ORIGINAL baseline again = transient read anomaly, flag clears. A THIRD hash = active rot: counter resets, re-notification fires. Manual override: `POST /api/bitrot/accept` (file_ids) adopts the current state immediately via the same shared transition as auto-expire; the adopted baseline mtime is the one recorded with the candidate hash, not a fresh stat, so the stored hash/mtime pair always describes content that actually existed. Stale queued results for a row whose flag was cleared mid-run (accept races Phase 3 on budgeted runs) are ignored instead of resurrecting state-machine fields. +- **Notification rules now actually fire.** `NotificationRule`/`NotificationProvider` existed as CRUD-only models - no backend code ever evaluated rules. New `dispatch_event()` in the notification service joins active rules to active providers, delivers, and records per-provider status. `bitrot_suspected` is the first wired event type. +- `bitrot_suspected` filter on `/api/scan-results`; bitrot columns in CSV/JSON exports and `ScanResult.to_dict`. + +### Fixed + +- **`get_file_info` stored naive local mtimes; all baselines are now UTC.** `datetime.fromtimestamp(st_mtime)` without a timezone poisoned every stored `last_modified` on hosts where TZ != UTC (its error fallback even stored aware UTC, mixing both semantics in one column). Bitrot classification requires trusted baselines, so a new `mtime_baseline_utc` flag (false for all pre-upgrade rows) gates it: untrusted rows whose hash mismatches classify as `modified` and re-baseline via rescan, and untrusted rows whose hash MATCHES (the cold-archive common case) re-baseline directly during the integrity check - no false bitrot alarms from the old data, detection reliable from each file's second check. The scan-result cache validity check now treats a content-hash match as sufficient (mtime equality only as fallback for hash-less rows), so the UTC change cannot force a one-time full-library re-decode on hosts where TZ is not UTC. Scan error paths no longer fabricate a scan-time mtime for unreadable files; they leave the stored baseline untouched. + +## [2.6.60] - 2026-07-09 + +### Added + +- **Integrity check schedules can now carry a time budget (`time_budget_minutes`).** A budgeted run dispatches hash tasks until the soft deadline expires, drains in-flight tasks (which still stamp `last_integrity_check_date`), and completes with a coverage message ("Budget reached: N of M files verified this run"). Because the rolling queue (2.6.59) is ordered stalest-first, successive budgeted runs sweep the whole library in disjoint slices - a 50-hour full pass becomes ~100 min/day for a monthly cadence - without any single run monopolizing IO. NULL budget = unlimited (current behavior); the field is valid only on `file_changes` schedules and manual runs stay unlimited unless the start request passes `time_budget_minutes` explicitly. Budgeted runs dispatch largest files first within each batch so a 70GB file is not started at minute 59; dispatch skips past saturated size tiers instead of head-of-line blocking, so a run of large files at the front of a batch cannot cap the whole window at the huge-file concurrency limit while small-file slots idle. The budget is resolved inside the service (not the HTTP route), so every entry point honors a schedule's configured budget. Schedule create/edit forms expose the field with sizing guidance. + +## [2.6.59] - 2026-07-09 + +### Changed + +- **The integrity check (file changes) is now a rolling queue instead of a single full-table pass.** The producer previously loaded every `scan_results` row into memory in one query (hundreds of MB at 1M+ files) and lost all progress on crash or cancellation. It now re-fetches the queue in ordered batches (`INTEGRITY_BATCH_SIZE`, default 10,000): never-checked files first (`last_integrity_check_date` NULL), then stalest-first. `last_integrity_check_date` - previously written but never read - is now the queue cursor: every terminal outcome (result received, hash error after retries, abandoned task) stamps the file, pushing it to the back of the ordering, so an interrupted run resumes where it left off by construction. In-flight tasks, rows whose timestamp write failed, and rows whose task dispatch failed (broker errors) are excluded from re-fetches explicitly - ordering alone would re-hash the first two and re-fail the third forever; failure counts are surfaced in the completion message. Stamps are written in bulk (one UPDATE per batch fetch instead of one SELECT plus UPDATE per file), a queue re-fetch only runs after a dispatch, completion, or exclusion changed the queue (no tight query loops when every size tier is saturated or the broker is down), and a successfully hashed file gets `file_exists` restored, recovering rows stranded by a past mount outage. A composite index supporting the queue ordering ships in the same migration. + +### Fixed + +- **Producer no longer accumulates one AsyncResult handle per dispatched task.** `task_results` collected every dispatched task's result handle for the lifetime of the run and was never read - a per-run memory leak of 1M+ objects on large libraries. Removed. + +## [2.6.58] - 2026-07-09 + +### Removed + +- **Legacy serial `MediaChecker.check_file_changes` deleted.** It had zero callers (all file-changes routes go through `MaintenanceService`'s Celery-based path) and was a divergent second implementation: it only hashed a file when its mtime differed, so it structurally could not detect bitrot (content change with unchanged mtime); it compared naive local timestamps; and it ran serially with no progress reporting or cancellation. Removing it ahead of the bitrot classification work ensures the integrity-check logic exists in exactly one place. + ## [2.6.57] - 2026-06-11 ### Fixed diff --git a/README.md b/README.md index 357add9..64995cf 100644 --- a/README.md +++ b/README.md @@ -20,6 +20,7 @@ PixelProbe detects corrupted video, image, and audio files across your media lib - FFmpeg-based deep video analysis - Video freeze detection (stuck frames while audio continues) via FFmpeg freezedetect filter with black frame false positive filtering - ImageMagick and PIL image validation +- Bitrot detection: a content hash change without a matching mtime change flags the file for review instead of silently adopting the new hash - Smart warning system for minor issues vs critical corruption - Multi-stage detection with configurable thresholds - Automatic retry logic for transient failures @@ -28,6 +29,7 @@ PixelProbe detects corrupted video, image, and audio files across your media lib - **Parallel multi-threaded scanning**: Configurable worker threads (10-24 workers recommended) with thread-safe database access - **Real-time progress**: Live updates with ETA calculations and phase tracking - **Multiple scan types**: Full scan, cleanup, integrity check +- **Rolling integrity queue**: Integrity checks sweep the library stalest-first in batches and resume where they left off; schedules can carry a per-run time budget so no single run monopolizes IO - **Scheduled automated scans**: Cron expressions or simple intervals for hands-free monitoring - **Smart exclusions**: Configure paths and file extensions to skip - **Phase-based scanning**: Discovery → Database → Validation workflow @@ -342,8 +344,9 @@ PixelProbe provides a REST API with OpenAPI/Swagger documentation. #### Maintenance Endpoints - `POST /api/cleanup` - Remove orphaned database entries - `GET /api/cleanup-status` - Get cleanup operation status -- `POST /api/file-changes` - Detect file system changes +- `POST /api/file-changes` - Detect file system changes (optional `time_budget_minutes`) - `GET /api/file-changes-status` - Get file changes scan status +- `POST /api/bitrot/accept` - Accept a bitrot-suspected file's current content as the new baseline - `POST /api/reset-for-rescan` - Reset files for rescanning - `POST /api/reset-files-by-path` - Reset specific files by path diff --git a/docs/api/SCAN_TYPES_DOCUMENTATION.md b/docs/api/SCAN_TYPES_DOCUMENTATION.md index a191c4a..8e31203 100644 --- a/docs/api/SCAN_TYPES_DOCUMENTATION.md +++ b/docs/api/SCAN_TYPES_DOCUMENTATION.md @@ -95,27 +95,27 @@ POST /api/force-scan-pending ### 4. File Changes Scan (`file_changes`) **Endpoint**: `GET/POST /api/file-changes` -**Purpose**: Detect and scan modified files +**Purpose**: Verify stored content hashes and classify what changed (rolling integrity queue) **When to use**: -- Regular maintenance scans -- After media file edits -- To catch file corruption over time +- Scheduled integrity sweeps (pair with a per-schedule time budget) +- After bulk media edits +- To catch silent corruption (bitrot) over time **How it works**: -1. **Phase 1 - Integrity Check**: Compares file modification times with database -2. **Phase 2 - Scan Changed**: Re-scans files that have been modified -3. Updates database with new scan results +1. Pulls files from a rolling queue ordered stalest-first by `last_integrity_check_date`; files flagged as suspected bitrot jump the queue +2. Re-hashes each file and classifies the result: hash match = unchanged; hash and mtime both changed = modified (queued for rescan); hash changed while mtime did not = suspected bitrot (flagged and notified, stored baseline preserved) +3. Stamps every processed file, so interrupted or budget-limited runs resume where they left off **Features**: -- Detects file modifications via timestamps -- Optionally detects file size changes -- Re-scans only changed files -- Reports list of changed files -- Efficient for regular checks +- Optional `time_budget_minutes` (request body, or the schedule's field): dispatch stops at the deadline, in-flight hashes drain, and the queue resumes at the next run +- Bitrot flags auto-expire after consecutive stable checks plus a clean rescan; accept manually via `POST /api/bitrot/accept` +- Cumulative coverage exposed in `/api/stats` (`integrity` block) and the dashboard's Integrity Checked card +- Reports the changed-file list with hash, mtime, and size detail **Example**: ```json POST /api/file-changes +{"time_budget_minutes": 10} ``` ### 5. Cleanup (`cleanup`) diff --git a/openapi.yaml b/openapi.yaml index 8dd910d..188178b 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -674,6 +674,13 @@ paths: type: string enum: [all, 'true', 'false'] default: all + - name: bitrot_suspected + in: query + description: Filter files flagged as suspected bitrot (hash changed, mtime unchanged) + schema: + type: string + enum: [all, 'true', 'false'] + default: all - name: search in: query schema: @@ -1259,7 +1266,12 @@ paths: get: summary: Get statistics summary description: | - Retrieve comprehensive statistics about scanned files and corruption. + Retrieve statistics about scanned files and corruption. The response + includes an `integrity` block with rolling-coverage numbers: files + integrity-checked ever and within the last 30 days, coverage percent, + never-checked count, the oldest check date (every checked file has + been verified at least once since then), and the current + bitrot-suspected count. **Example:** ```bash @@ -1522,6 +1534,41 @@ paths: '200': description: Files marked as good successfully + /bitrot/accept: + post: + summary: Accept current state of bitrot-suspected files + description: | + Adopt each flagged file's current content as the new integrity + baseline (the candidate hash plus the mtime recorded when that hash + was computed), clearing the bitrot_suspected flag and stability + counter. The detection record (bitrot_detected_date, bitrot_details) + is preserved permanently. Files that are not flagged are skipped and + reported back. + + **Example:** + ```bash + curl -X POST http://your-server/api/bitrot/accept \ + -H "Authorization: Bearer YOUR_TOKEN" \ + -H "Content-Type: application/json" \ + -d '{"file_ids": [123, 456]}' + ``` + tags: [Maintenance] + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [file_ids] + properties: + file_ids: + type: array + items: + type: integer + responses: + '200': + description: Accepted; response lists accepted count and skipped ids + /schedules: get: summary: List scheduled scans @@ -1580,6 +1627,14 @@ paths: force_rescan: type: boolean default: false + time_budget_minutes: + type: integer + nullable: true + minimum: 1 + description: | + file_changes schedules only. Soft per-run deadline in + minutes; dispatch stops when it expires and the rolling + integrity queue resumes at the next run. Null = unlimited. responses: '201': description: Schedule created successfully @@ -2423,9 +2478,19 @@ paths: type: array items: type: string + time_budget_minutes: + type: integer + minimum: 1 + description: | + Optional soft deadline in minutes. Dispatch stops when it + expires; in-flight hashes drain and the rolling queue + resumes at the next run. Omitted = unlimited (scheduled + runs inherit their schedule's budget). responses: '200': description: File changes check started + '400': + description: Invalid time_budget_minutes '409': description: Check already in progress diff --git a/pixelprobe/api/admin_routes.py b/pixelprobe/api/admin_routes.py index 1c95b06..ed32134 100644 --- a/pixelprobe/api/admin_routes.py +++ b/pixelprobe/api/admin_routes.py @@ -9,6 +9,8 @@ from pixelprobe.models import db, ScanResult, IgnoredErrorPattern, ScanConfiguration, ScanSchedule from pixelprobe.scheduler import MediaScheduler from pixelprobe.utils.security import validate_json_input, AuditLogger, validate_directory_path +from pixelprobe.utils.validators import validate_time_budget +from pixelprobe.utils.integrity import adopt_bitrot_baseline from pixelprobe.auth import auth_required logger = logging.getLogger(__name__) @@ -76,6 +78,20 @@ def calculate_next_run(cron_expression: str, last_run=None): return trigger.get_next_fire_time(None, now) +def _parse_file_ids(data): + """Coerce and bound the file_ids list shared by batch file endpoints. + + Returns (file_ids, error_response). + """ + try: + file_ids = [int(fid) for fid in data.get('file_ids', [])] + except (ValueError, TypeError): + return None, ({'error': 'Invalid file ID format'}, 400) + if len(file_ids) > 1000: # Prevent excessive updates + return None, ({'error': 'Too many file IDs (max 1000)'}, 400) + return file_ids, None + + @admin_bp.route('/mark-as-good', methods=['POST']) @rate_limit("10 per minute") @auth_required @@ -85,17 +101,10 @@ def calculate_next_run(cron_expression: str, last_run=None): def mark_as_good(): """Mark files as good/healthy""" data = request.get_json() - file_ids = data.get('file_ids', []) - - # Validate file IDs are integers - try: - file_ids = [int(fid) for fid in file_ids] - except (ValueError, TypeError): - return {'error': 'Invalid file ID format'}, 400 - - if len(file_ids) > 1000: # Prevent excessive updates - return {'error': 'Too many file IDs (max 1000)'}, 400 - + file_ids, id_error = _parse_file_ids(data) + if id_error: + return id_error + try: for file_id in file_ids: result = db.session.get(ScanResult, file_id) @@ -118,6 +127,66 @@ def mark_as_good(): db.session.rollback() return {'error': 'Internal server error'}, 500 +@admin_bp.route('/bitrot/accept', methods=['POST']) +@rate_limit("10 per minute") +@auth_required +@validate_json_input({ + 'file_ids': {'required': True, 'type': list} +}) +def accept_bitrot_current_state(): + """Accept a bitrot-suspected file's current content as the new baseline. + + Adopts the candidate hash and current on-disk mtime, clears the flag and + stability counter. bitrot_detected_date/bitrot_details are preserved so + files that ever tripped bitrot stay queryable (dying-disk signal). + """ + data = request.get_json() + file_ids, id_error = _parse_file_ids(data) + if id_error: + return id_error + + try: + rows = ScanResult.query.filter(ScanResult.id.in_(file_ids)).all() if file_ids else [] + rows_by_id = {row.id: row for row in rows} + accepted = 0 + skipped = [] + for file_id in file_ids: + result = rows_by_id.get(file_id) + if not result or not result.bitrot_suspected: + skipped.append(file_id) + continue + + # Baseline mtime: the one recorded when the candidate hash was + # computed. A fresh os.stat could pair the candidate hash with a + # NEWER mtime if the file changed again after the last check, + # storing a baseline pair that never described real content. If + # no recorded mtime exists the row re-baselines its mtime on the + # next hash-match integrity check. + mtime = None + try: + details = json.loads(result.bitrot_details or '{}') + if details.get('current_modified'): + mtime = datetime.fromisoformat(details['current_modified']) + except (ValueError, TypeError): + mtime = None + + adopt_bitrot_baseline(result, mtime) + accepted += 1 + logger.info(f"Accepted current state for bitrot-suspected file: {result.file_path}") + AuditLogger.log_action('bitrot_accept', {'file_id': file_id, 'file_path': result.file_path}) + + db.session.commit() + return { + 'message': f'Accepted current state for {accepted} file(s)', + 'accepted': accepted, + 'skipped': skipped + } + except Exception as e: + logger.error(f"Error accepting bitrot state: {str(e)}", exc_info=True) + db.session.rollback() + return {'error': 'Internal server error'}, 500 + + @admin_bp.route('/ignored-patterns') @auth_required def get_ignored_patterns(): @@ -277,28 +346,49 @@ def get_schedule(schedule_id): schedule = db.get_or_404(ScanSchedule, schedule_id) return jsonify(schedule.to_dict()) +def _validate_time_budget(data, scan_type): + """Validate time_budget_minutes from a schedule payload. Returns (value, error_response).""" + value, error = validate_time_budget(data.get('time_budget_minutes'), scan_type) + if error: + return None, ({'error': error}, 400) + return value, None + + @admin_bp.route('/schedules', methods=['POST']) @auth_required def create_schedule(): """Create a new scan schedule""" data = request.get_json() - + try: # Check for duplicate name name = data.get('name', 'Unnamed Schedule') existing = ScanSchedule.query.filter_by(name=name, is_active=True).first() if existing: return {'error': f'Schedule with name "{name}" already exists'}, 400 - + + scan_type = data.get('scan_type', 'full') + time_budget, budget_error = _validate_time_budget(data, scan_type) + if budget_error: + return budget_error + schedule = ScanSchedule( name=name, cron_expression=data['cron_expression'], scan_paths=json.dumps(data.get('scan_paths', [])), - scan_type=data.get('scan_type', 'full'), + scan_type=scan_type, force_rescan=data.get('force_rescan', False), + time_budget_minutes=time_budget, is_active=True, created_at=datetime.now(timezone.utc) ) + # Populate next_run immediately so the UI shows it before the first + # fire (the scheduler's db-sync job registers the actual APScheduler + # job within a minute) + try: + schedule.next_run = calculate_next_run(schedule.cron_expression) + except Exception as e: + logger.warning(f"Could not calculate next_run for new schedule: {e}") db.session.add(schedule) db.session.commit() @@ -333,12 +423,23 @@ def update_schedule(schedule_id): new_cron = data.get('cron_expression', schedule.cron_expression) cron_changed = new_cron != schedule.cron_expression + new_scan_type = data.get('scan_type', schedule.scan_type) + if 'time_budget_minutes' in data: + time_budget, budget_error = _validate_time_budget(data, new_scan_type) + if budget_error: + return budget_error + schedule.time_budget_minutes = time_budget + elif new_scan_type != 'file_changes' and schedule.time_budget_minutes is not None: + # Type changed away from file_changes: the budget no longer applies + logger.info(f"Clearing time_budget_minutes on schedule {schedule_id} (scan_type now {new_scan_type})") + schedule.time_budget_minutes = None + # Update fields schedule.name = data.get('name', schedule.name) schedule.cron_expression = new_cron if 'scan_paths' in data: schedule.scan_paths = json.dumps(data['scan_paths']) - schedule.scan_type = data.get('scan_type', schedule.scan_type) + schedule.scan_type = new_scan_type schedule.force_rescan = data.get('force_rescan', schedule.force_rescan) schedule.is_active = new_is_active diff --git a/pixelprobe/api/maintenance_routes.py b/pixelprobe/api/maintenance_routes.py index 893161a..02e7890 100644 --- a/pixelprobe/api/maintenance_routes.py +++ b/pixelprobe/api/maintenance_routes.py @@ -11,6 +11,7 @@ from pixelprobe.media_checker import PixelProbe from pixelprobe.auth import auth_required from pixelprobe.utils.helpers import ProgressTracker +from pixelprobe.utils.validators import validate_time_budget from pixelprobe.services.maintenance_service import MaintenanceService from pixelprobe.progress_utils import get_file_changes_progress_redis @@ -506,6 +507,16 @@ def check_file_changes(): file_paths = data.get('file_paths', []) schedule_id = data.get('schedule_id') # For healthcheck integration + # Time budget: explicit request value wins (manual runs are unlimited by + # default). Scheduled runs inherit their schedule's budget inside + # MaintenanceService, so every entry point resolves it identically. + time_budget_minutes = data.get('time_budget_minutes') + if time_budget_minutes is not None: + time_budget_minutes, budget_error = validate_time_budget(time_budget_minutes, 'file_changes') + if budget_error: + db.session.rollback() # release the advisory lock + return {'error': budget_error}, 400 + # Create unique check ID check_id = str(uuid.uuid4()) @@ -555,7 +566,7 @@ def check_file_changes(): app = current_app._get_current_object() current_file_changes_thread = threading.Thread( target=check_file_changes_async, - args=(app, check_id, file_paths, schedule_id) + args=(app, check_id, file_paths, schedule_id, time_budget_minutes) ) current_file_changes_thread.start() @@ -612,7 +623,7 @@ def cleanup_orphaned_async(app, cleanup_id, file_paths=None, schedule_id=None): except Exception as commit_error: logger.error(f"Failed to update cleanup record on error: {str(commit_error)}") -def check_file_changes_async(app, check_id, file_paths=None, schedule_id=None): +def check_file_changes_async(app, check_id, file_paths=None, schedule_id=None, time_budget_minutes=None): """Async function to check file changes Args: @@ -620,6 +631,8 @@ def check_file_changes_async(app, check_id, file_paths=None, schedule_id=None): check_id: Unique ID for this check file_paths: Optional list of specific file paths to check (if None, checks all files) schedule_id: Optional schedule ID for healthcheck integration + time_budget_minutes: Optional soft deadline; dispatch stops when it + expires and the rolling queue resumes at the next run """ try: with app.app_context(): @@ -637,7 +650,9 @@ def check_file_changes_async(app, check_id, file_paths=None, schedule_id=None): maintenance_service = MaintenanceService(app.config['SQLALCHEMY_DATABASE_URI']) # Run the file changes check using the maintenance service logic with optional file_paths filter - maintenance_service._run_file_changes_check(check_record.check_id, file_paths=file_paths, schedule_id=schedule_id) + maintenance_service._run_file_changes_check( + check_record.check_id, file_paths=file_paths, schedule_id=schedule_id, + time_budget_minutes=time_budget_minutes) except Exception as e: logger.error(f"Error in check_file_changes_async: {str(e)}", exc_info=True) diff --git a/pixelprobe/api/notification_routes.py b/pixelprobe/api/notification_routes.py index 792610f9..1e2b668 100644 --- a/pixelprobe/api/notification_routes.py +++ b/pixelprobe/api/notification_routes.py @@ -21,7 +21,7 @@ VALID_PROVIDER_TYPES = ['pushover', 'ntfy', 'webhook'] VALID_EVENT_TYPES = [ 'scan_start', 'scan_complete', 'scan_failed', 'scan_missed', - 'corruption_found', 'user_added', 'user_deleted', + 'corruption_found', 'bitrot_suspected', 'user_added', 'user_deleted', 'api_key_added', 'api_key_deleted', 'auth_failed' ] VALID_PRIORITIES = ['low', 'normal', 'high'] diff --git a/pixelprobe/api/scan_routes.py b/pixelprobe/api/scan_routes.py index 8cd37d4..6e1fd25 100644 --- a/pixelprobe/api/scan_routes.py +++ b/pixelprobe/api/scan_routes.py @@ -209,6 +209,7 @@ def get_scan_results(): scan_status = request.args.get('scan_status', 'all') is_corrupted = request.args.get('is_corrupted', 'all') has_warnings = request.args.get('has_warnings', 'all') + bitrot_suspected = request.args.get('bitrot_suspected', 'all') search_query = request.args.get('search', '').strip() sort_field = request.args.get('sort_field', 'scan_date') sort_order = request.args.get('sort_order', 'desc') @@ -255,7 +256,13 @@ def get_scan_results(): (ScanResult.has_warnings == False) | (ScanResult.has_warnings == None) ) - + + # Apply bitrot filter + if bitrot_suspected == 'true': + query = query.filter(ScanResult.bitrot_suspected == True) + elif bitrot_suspected == 'false': + query = query.filter(ScanResult.bitrot_suspected == False) + # Apply sorting # Map frontend field names to model attributes field_mapping = { diff --git a/pixelprobe/api/stats_routes.py b/pixelprobe/api/stats_routes.py index 8f956c6..70f7f1d 100644 --- a/pixelprobe/api/stats_routes.py +++ b/pixelprobe/api/stats_routes.py @@ -51,6 +51,9 @@ def get_stats(): 'marked_as_good': stats[7] or 0, 'warning_files': stats[8] or 0 } + # Rolling-integrity coverage: answers "how much of the library has + # been verified over time" - no single budgeted run report can + result['integrity'] = StatsService().get_integrity_coverage() return result @@ -104,7 +107,8 @@ def get_stats(): 'corrupted_files': corrupted_files, 'healthy_files': healthy_files, 'marked_as_good': marked_as_good, - 'warning_files': warning_files + 'warning_files': warning_files, + 'integrity': StatsService().get_integrity_coverage() } except Exception as e2: logger.error(f"Fallback stats query also failed: {str(e2)}") diff --git a/pixelprobe/media_checker.py b/pixelprobe/media_checker.py index 14a1a41..d0d9b27 100644 --- a/pixelprobe/media_checker.py +++ b/pixelprobe/media_checker.py @@ -23,6 +23,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed import threading from pixelprobe.utils.security import safe_subprocess_run, validate_file_path, ensure_cli_safe_path +from pixelprobe.utils.integrity import apply_scan_baseline from pixelprobe.utils.paths import is_path_under logger = logging.getLogger(__name__) @@ -507,8 +508,11 @@ def get_file_info(self, file_path): try: file_stats = os.stat(file_path) file_size = file_stats.st_size - creation_date = datetime.fromtimestamp(file_stats.st_ctime) - last_modified = datetime.fromtimestamp(file_stats.st_mtime) + # UTC-aware: bitrot classification compares this stored baseline + # against a UTC mtime, and naive local values poison it (the old + # naive form is why mtime_baseline_utc exists). + creation_date = datetime.fromtimestamp(file_stats.st_ctime, timezone.utc) + last_modified = datetime.fromtimestamp(file_stats.st_mtime, timezone.utc) file_type = magic.from_file(file_path, mime=True) return { @@ -640,8 +644,10 @@ def _scan_files_single_pool(self, file_paths, progress_callback=None, force_resc 'file_path': file_path, 'file_size': 0, 'file_type': 'unknown', - 'creation_date': datetime.now(), - 'last_modified': datetime.now(), + 'creation_date': datetime.now(timezone.utc), + # No real file mtime available here; None keeps the + # stored baseline instead of poisoning it with scan time + 'last_modified': None, 'is_corrupted': True, 'corruption_details': f"Scan error: {str(e)}" }) @@ -718,8 +724,10 @@ def scan_path_files(path, path_files): 'file_path': file_path, 'file_size': 0, 'file_type': 'unknown', - 'creation_date': datetime.now(), - 'last_modified': datetime.now(), + 'creation_date': datetime.now(timezone.utc), + # No real file mtime available here; None keeps the + # stored baseline instead of poisoning it with scan time + 'last_modified': None, 'is_corrupted': True, 'corruption_details': f"Scan error: {str(e)}" } @@ -2342,42 +2350,6 @@ def _check_strict_error_detection(self, file_path): return is_corrupted, corruption_details - def check_file_changes(self, scan_results_db): - """Check for file changes by comparing current hashes with stored hashes""" - changed_files = [] - - for result in scan_results_db: - file_path = result.file_path - stored_hash = result.file_hash - stored_modified = result.last_modified - - if not os.path.exists(file_path): - changed_files.append({ - 'file_path': file_path, - 'change_type': 'deleted', - 'stored_hash': stored_hash, - 'current_hash': None - }) - continue - - # Check if file was modified - current_stats = os.stat(file_path) - current_modified = datetime.fromtimestamp(current_stats.st_mtime) - - if stored_modified and current_modified != stored_modified: - current_hash = self.calculate_file_hash(file_path) - if current_hash != stored_hash: - changed_files.append({ - 'file_path': file_path, - 'change_type': 'modified', - 'stored_hash': stored_hash, - 'current_hash': current_hash, - 'stored_modified': stored_modified, - 'current_modified': current_modified - }) - - return changed_files - def find_orphaned_records(self, scan_results_db): """Find database records for files that no longer exist""" orphaned_records = [] @@ -2425,10 +2397,16 @@ def _check_cache(self, file_path, file_hash, last_modified): session.close() return None - # Check if file hasn't changed (same hash and modification time) - if (result.file_hash == file_hash and - result.last_modified and - result.last_modified.replace(tzinfo=None) == last_modified.replace(tzinfo=None)): + # Content hash equality alone proves the cached result is + # current. The old additional mtime-equality requirement broke + # after the UTC mtime fix: pre-upgrade rows store naive LOCAL + # time, so on any TZ != UTC host every cached row would miss + # and force a full library re-decode. mtime remains as the + # fallback for rows scanned before hashes were stored. + if ((result.file_hash and file_hash and result.file_hash == file_hash) or + (not result.file_hash and + result.last_modified and last_modified and + result.last_modified.replace(tzinfo=None) == last_modified.replace(tzinfo=None))): # Convert database result to expected format cached_data = { @@ -2500,10 +2478,16 @@ def _save_to_cache(self, file_path, scan_result): db_result.file_size = scan_result.get('file_size') db_result.file_type = scan_result.get('file_type') db_result.creation_date = scan_result.get('creation_date') - db_result.last_modified = scan_result.get('last_modified') + if not apply_scan_baseline(db_result, scan_result.get('file_hash'), + scan_result.get('last_modified')): + # Anti-laundering: a rescan of a bitrot-suspected file must not + # adopt its current content as the baseline - a bit flip can + # pass decode checks and would silently become the new "good" + # hash. Baseline updates for flagged files happen only via + # auto-expire or the manual accept action. + logger.info(f"Preserving hash/mtime baseline for bitrot-suspected file: {file_path}") db_result.is_corrupted = scan_result.get('is_corrupted', False) db_result.corruption_details = scan_result.get('corruption_details') - db_result.file_hash = scan_result.get('file_hash') db_result.scan_tool = scan_result.get('scan_tool') db_result.scan_duration = scan_result.get('scan_duration') db_result.scan_output = scan_result.get('scan_output') diff --git a/pixelprobe/migrations/startup.py b/pixelprobe/migrations/startup.py index e93f893..c5427c2 100644 --- a/pixelprobe/migrations/startup.py +++ b/pixelprobe/migrations/startup.py @@ -339,6 +339,77 @@ def run_v2_6_53_migrations(db): logger.error(f"Migration v2.6.53 failed: {e}") +def run_v2_6_60_migrations(db): + """Add scan_schedules.time_budget_minutes for budgeted integrity runs. + + NULL = unlimited (current behavior). Only meaningful for + scan_type='file_changes'; the API rejects it on other types. + """ + try: + with migration_connection(db) as conn: + exists = conn.execute(text( + "SELECT 1 FROM information_schema.columns " + "WHERE table_name = 'scan_schedules' AND column_name = 'time_budget_minutes'" + )).fetchone() + if not exists: + logger.info("Adding column scan_schedules.time_budget_minutes (INTEGER)") + conn.execute(text( + "ALTER TABLE scan_schedules ADD COLUMN time_budget_minutes INTEGER" + )) + conn.commit() + logger.info("v2.6.60 schedule budget migration completed") + except Exception as e: + logger.error(f"Migration v2.6.60 failed: {e}") + + +def run_v2_6_61_migrations(db): + """Bitrot classification columns on scan_results. + + bitrot_suspected: hash changed while mtime did not - flagged for review. + bitrot_detected_date/bitrot_details: permanent detection record. + bitrot_candidate_hash/bitrot_stable_checks: auto-expire state machine. + mtime_baseline_utc: false for all pre-upgrade rows, whose last_modified + was written as naive local time; bitrot classification requires a + trusted (UTC) baseline, so those rows re-baseline on first check. + """ + columns = [ + ("bitrot_suspected", "BOOLEAN NOT NULL DEFAULT FALSE"), + ("bitrot_detected_date", "TIMESTAMP"), + ("bitrot_details", "TEXT"), + ("bitrot_candidate_hash", "VARCHAR(64)"), + ("bitrot_stable_checks", "INTEGER NOT NULL DEFAULT 0"), + ("mtime_baseline_utc", "BOOLEAN NOT NULL DEFAULT FALSE"), + ] + try: + with migration_connection(db) as conn: + for name, ddl in columns: + exists = conn.execute(text( + "SELECT 1 FROM information_schema.columns " + "WHERE table_name = 'scan_results' AND column_name = :col" + ), {'col': name}).fetchone() + if not exists: + logger.info(f"Adding column scan_results.{name}") + conn.execute(text( + f"ALTER TABLE scan_results ADD COLUMN {name} {ddl}" + )) + conn.execute(text( + "CREATE INDEX IF NOT EXISTS idx_scan_results_bitrot_suspected " + "ON scan_results(bitrot_suspected)" + )) + # Supports the rolling-queue fetch ordering exactly; without it + # every ~10k-row batch fetch is a full-table scan + top-N sort + # (~100 scans per run at 1M files). + conn.execute(text( + "CREATE INDEX IF NOT EXISTS idx_scan_results_integrity_queue " + "ON scan_results (bitrot_suspected DESC, " + "last_integrity_check_date ASC NULLS FIRST, id ASC)" + )) + conn.commit() + logger.info("v2.6.61 bitrot classification migration completed") + except Exception as e: + logger.error(f"Migration v2.6.61 failed: {e}") + + def create_performance_indexes(db): """Create performance indexes""" indexes = [ @@ -430,6 +501,18 @@ def _run_all_migrations(db): except Exception as e: logger.error(f"v2.6.53 migration failed: {e}") + logger.info("Running v2.6.60 migration (schedule time budget)...") + try: + run_v2_6_60_migrations(db) + except Exception as e: + logger.error(f"v2.6.60 migration failed: {e}") + + logger.info("Running v2.6.61 migration (bitrot classification)...") + try: + run_v2_6_61_migrations(db) + except Exception as e: + logger.error(f"v2.6.61 migration failed: {e}") + logger.info("Creating performance indexes...") try: create_performance_indexes(db) diff --git a/pixelprobe/models.py b/pixelprobe/models.py index 4cebe74..0f78413 100644 --- a/pixelprobe/models.py +++ b/pixelprobe/models.py @@ -59,6 +59,21 @@ class ScanResult(db.Model): error_message = db.Column(db.Text, nullable=True) # Error message from scan media_info = db.Column(db.Text, nullable=True) # JSON string of media metadata file_exists = db.Column(db.Boolean, nullable=False, default=True, index=True) # Whether file exists on disk + + # Bitrot classification: a hash mismatch with unchanged mtime is suspected + # bitrot, not a legitimate edit (see calculate_file_hash_task). The flag + # auto-expires after BITROT_STABLE_CHECKS_TO_EXPIRE stable checks plus a + # clean rescan; detection date/details are permanent once set so "files + # that ever tripped bitrot" stays queryable (dying-disk signal). + bitrot_suspected = db.Column(db.Boolean, nullable=False, default=False, server_default='false', index=True) + bitrot_detected_date = db.Column(db.DateTime, nullable=True) + bitrot_details = db.Column(db.Text, nullable=True) # JSON: stored vs current hash/mtime/size + bitrot_candidate_hash = db.Column(db.String(64), nullable=True) # stability reference for auto-expire + bitrot_stable_checks = db.Column(db.Integer, nullable=False, default=0, server_default='0') + # True once last_modified was written as UTC by post-v2.6.61 code. + # Pre-upgrade baselines are naive local time and must not trigger bitrot + # classification; their first check re-baselines instead. + mtime_baseline_utc = db.Column(db.Boolean, nullable=False, default=False, server_default='false') # Legacy column kept for backward compatibility with older database schemas deep_scan = db.Column(db.Boolean, nullable=True, default=False, server_default='false') @@ -109,7 +124,12 @@ def to_dict(self): 'discovered_date': convert_to_tz(self.discovered_date), 'error_message': self.error_message, 'media_info': self.media_info, - 'file_exists': self.file_exists + 'file_exists': self.file_exists, + 'bitrot_suspected': self.bitrot_suspected, + 'bitrot_detected_date': convert_to_tz(self.bitrot_detected_date), + 'bitrot_details': self.bitrot_details, + 'bitrot_candidate_hash': self.bitrot_candidate_hash, + 'bitrot_stable_checks': self.bitrot_stable_checks } def append_output(self, new_output): @@ -205,6 +225,9 @@ class ScanSchedule(db.Model): scan_paths = db.Column(db.Text) # JSON array of paths to scan scan_type = db.Column(db.String(20), nullable=False, default='normal') # normal, orphan, file_changes force_rescan = db.Column(db.Boolean, nullable=False, default=False) + # file_changes only: stop dispatching new hash tasks after this many + # minutes; the rolling queue resumes at the next run. NULL = unlimited. + time_budget_minutes = db.Column(db.Integer, nullable=True) is_active = db.Column(db.Boolean, nullable=False, default=True) last_run = db.Column(db.DateTime, nullable=True) next_run = db.Column(db.DateTime, nullable=True) @@ -219,6 +242,7 @@ def to_dict(self): 'scan_paths': json.loads(self.scan_paths) if self.scan_paths else [], 'scan_type': self.scan_type, 'force_rescan': self.force_rescan, + 'time_budget_minutes': self.time_budget_minutes, 'is_active': self.is_active, 'last_run': convert_to_tz(self.last_run), 'next_run': convert_to_tz(self.next_run), @@ -1026,7 +1050,8 @@ class NotificationRule(db.Model): nullable=False, index=True) event_type = db.Column(db.String(50), nullable=False, index=True) # Event types: 'scan_start', 'scan_complete', 'scan_failed', 'scan_missed', - # 'user_added', 'user_deleted', 'api_key_added', 'api_key_deleted', 'auth_failed' + # 'corruption_found', 'bitrot_suspected', 'user_added', 'user_deleted', + # 'api_key_added', 'api_key_deleted', 'auth_failed' is_active = db.Column(db.Boolean, nullable=False, default=True) priority = db.Column(db.String(10), nullable=False, default='normal') # 'low', 'normal', 'high' conditions = db.Column(db.JSON, nullable=True) # Optional conditions (e.g., {"corrupted_count": ">0"}) diff --git a/pixelprobe/scheduler.py b/pixelprobe/scheduler.py index 3648f2c..d4afcc4 100644 --- a/pixelprobe/scheduler.py +++ b/pixelprobe/scheduler.py @@ -62,6 +62,9 @@ def __init__(self, app=None): self.pending_retries: Dict[str, int] = {} self._retry_lock = threading.Lock() + # Snapshot of saved-schedule definitions; the db-sync job reloads jobs + # only when this changes (see _sync_schedules_from_db) + self._schedules_fp = None self.retry_delay_minutes = self._load_positive_int_env( 'SCHEDULE_RETRY_DELAY_MINUTES', self.DEFAULT_RETRY_DELAY_MINUTES, min_value=1 ) @@ -277,10 +280,31 @@ def init_app(self, app): misfire_grace_time=60 ) logger.info("Scheduled stuck scan detection to run every 5 minutes") - + + # Re-sync saved schedules from the database every 60 seconds. + # Schedule create/update/delete happens in gunicorn workers, but + # APScheduler runs only in the process holding the scheduler lock; the + # Celery reload task always executes in a prefork pool child and can + # never reach it, so changes only took effect after a restart. The id + # must not start with 'schedule_' or update_schedules would remove it. + self.scheduler.add_job( + func=self._sync_schedules_from_db, + trigger="interval", + seconds=60, + id="db_schedule_sync", + name="Sync saved schedules from database", + misfire_grace_time=30, + coalesce=True + ) + logger.info("Scheduled database schedule sync every 60 seconds") + # Load saved schedules from database with app.app_context(): self._load_saved_schedules() + try: + self._schedules_fp = self._schedule_fingerprint() + except Exception as e: + logger.warning(f"Could not compute initial schedule fingerprint: {e}") def _load_exclusions(self): """Load path and extension exclusions from environment variables""" @@ -743,6 +767,38 @@ def update_schedules(self): # Reload from database with self.app.app_context(): self._load_saved_schedules() + self._schedules_fp = self._schedule_fingerprint() + + def _schedule_fingerprint(self): + """Snapshot of every field that affects job registration. + + Must be called inside an app context. + """ + return tuple(db.session.query( + ScanSchedule.id, ScanSchedule.name, ScanSchedule.cron_expression, + ScanSchedule.scan_type, ScanSchedule.scan_paths, + ScanSchedule.force_rescan, ScanSchedule.time_budget_minutes, + ScanSchedule.is_active + ).order_by(ScanSchedule.id).all()) + + def _sync_schedules_from_db(self): + """Reload saved schedules when their definitions change in the database. + + Schedule CRUD runs in gunicorn workers; APScheduler runs only in the + process holding the scheduler lock. The Celery reload task executes in + a prefork pool child where the scheduler is never running, so it was + always skipped and new/edited schedules never registered until a + restart. Polling a fingerprint keeps the job store in sync without + cross-process messaging, and only churns jobs when something changed. + """ + try: + with self.app.app_context(): + fingerprint = self._schedule_fingerprint() + if fingerprint != self._schedules_fp: + logger.info("Schedule definitions changed in database; reloading scheduler jobs") + self.update_schedules() + except Exception as e: + logger.error(f"Schedule database sync failed: {e}") def _check_stuck_scans(self): """Check for stuck scans and mark them as crashed""" diff --git a/pixelprobe/services/export_service.py b/pixelprobe/services/export_service.py index d1f44ff..8933edf 100644 --- a/pixelprobe/services/export_service.py +++ b/pixelprobe/services/export_service.py @@ -72,9 +72,11 @@ def export_to_csv(self, filter_type: str = 'all', search: str = '', 'Scan Date', 'Scan Status', 'Discovered Date', - 'Marked as Good' + 'Marked as Good', + 'Bitrot Suspected', + 'Bitrot Detected Date' ]) - + # Write data rows for result in results: writer.writerow([ @@ -87,9 +89,11 @@ def export_to_csv(self, filter_type: str = 'all', search: str = '', result.corruption_details or '', result.scan_date.isoformat() if result.scan_date else '', getattr(result, 'scan_status', 'completed'), - getattr(result, 'discovered_date', result.scan_date).isoformat() + getattr(result, 'discovered_date', result.scan_date).isoformat() if getattr(result, 'discovered_date', result.scan_date) else '', - 'Yes' if result.marked_as_good else 'No' + 'Yes' if result.marked_as_good else 'No', + 'Yes' if result.bitrot_suspected else 'No', + result.bitrot_detected_date.isoformat() if result.bitrot_detected_date else '' ]) # Prepare response @@ -150,7 +154,9 @@ def export_to_json(self, filter_type: str = 'all', search: str = '', 'corruption_details': result.corruption_details, 'scan_date': result.scan_date.isoformat() if result.scan_date else None, 'scan_status': getattr(result, 'scan_status', 'completed'), - 'marked_as_good': result.marked_as_good + 'marked_as_good': result.marked_as_good, + 'bitrot_suspected': result.bitrot_suspected, + 'bitrot_detected_date': result.bitrot_detected_date.isoformat() if result.bitrot_detected_date else None }) return export_data diff --git a/pixelprobe/services/maintenance_service.py b/pixelprobe/services/maintenance_service.py index f55e5f8..940b4f2 100644 --- a/pixelprobe/services/maintenance_service.py +++ b/pixelprobe/services/maintenance_service.py @@ -3,6 +3,7 @@ """ import os +import json import threading import time import logging @@ -13,10 +14,12 @@ from celery import states from celery.exceptions import TimeoutError as CeleryTimeoutError -from sqlalchemy import text +from sqlalchemy import text, or_ from pixelprobe.media_checker import PixelProbe, load_exclusions, load_exclusions_with_patterns -from pixelprobe.models import db, ScanResult, CleanupState, FileChangesState, ScanReport, LogEntry, AppConfig +from pixelprobe.models import db, ScanResult, ScanSchedule, CleanupState, FileChangesState, ScanReport, LogEntry, AppConfig +from pixelprobe.services.notification_service import dispatch_event from pixelprobe.utils.helpers import ProgressTracker, env_int +from pixelprobe.utils.integrity import adopt_bitrot_baseline from pixelprobe.constants import CONFIG_LOG_RETENTION_DAYS from pixelprobe.progress_utils import ( update_file_changes_progress_redis, @@ -46,6 +49,23 @@ # producer loop cannot pin at max concurrency and spin forever. CLEANUP_TASK_TIMEOUT_SECS = env_int('CLEANUP_TASK_TIMEOUT_SECS', 600, floor=60) +# Consecutive stable-hash integrity checks (plus a clean rescan) required +# before a bitrot flag auto-expires and the stable content becomes the new +# baseline. Override via the BITROT_STABLE_CHECKS_TO_EXPIRE environment +# variable. +BITROT_STABLE_CHECKS_TO_EXPIRE = env_int('BITROT_STABLE_CHECKS_TO_EXPIRE', 2, floor=1) + + +def _size_tier(size_bytes): + """Concurrency tier for a file size; must match the MAX_CONCURRENT_* gating.""" + if size_bytes < 10 * 1024 * 1024: + return 'small' + if size_bytes < 100 * 1024 * 1024: + return 'medium' + if size_bytes < 1024 * 1024 * 1024: + return 'large' + return 'huge' + def forget_task_result(task): """Delete a consumed result from the backend (best-effort). @@ -171,6 +191,70 @@ def safe_task_get(task, timeout=1, max_retries=5, base_delay=1.0): logger.error(f"Failed to get task result after {max_retries} attempts: {type(e).__name__}: {e}") raise + +def fetch_integrity_batch(file_paths, run_watermark, excluded_ids, batch_size, + largest_first=False): + """Fetch the stalest slice of the integrity queue as plain dicts. + + Queue order: never-checked files first (NULL last_integrity_check_date), + then oldest check date, id as tiebreak. Rows stamped at or after + run_watermark were already processed by this run and are skipped (the + table never empties, so fetch-until-empty alone would loop forever). + excluded_ids covers rows the ordering cannot exclude: in-flight tasks and + rows whose timestamp write failed, neither of which have a committed + fresh timestamp yet. + + Returned dicts are sorted smallest-first by default: small files backfill + the wide small-file concurrency slots while the few large-file slots + grind. Budgeted runs pass largest_first=True so big files start hashing + early in the window instead of at the deadline. + Plain dicts (not ORM rows) are immune to session expiration during the + long-running producer loop and to concurrent row deletion (v2.5.61). + """ + query = db.session.query( + ScanResult.id, + ScanResult.file_path, + ScanResult.file_hash, + ScanResult.file_size, + ScanResult.last_modified, + ScanResult.mtime_baseline_utc, + ScanResult.bitrot_suspected, + ScanResult.bitrot_candidate_hash + ).filter( + or_( + ScanResult.last_integrity_check_date.is_(None), + ScanResult.last_integrity_check_date < run_watermark + ) + ) + if file_paths: + query = query.filter(ScanResult.file_path.in_(file_paths)) + if excluded_ids: + query = query.filter(ScanResult.id.notin_(excluded_ids)) + rows = query.order_by( + # Flagged files jump the queue so auto-expire resolves in a few runs + # instead of a few full sweep cycles. + ScanResult.bitrot_suspected.desc(), + ScanResult.last_integrity_check_date.asc().nullsfirst(), + ScanResult.id.asc() + ).limit(batch_size).all() + batch = [ + { + 'id': r.id, + 'file_path': r.file_path, + 'file_hash': r.file_hash, + 'file_size': r.file_size, + 'last_modified': r.last_modified, + 'mtime_baseline_utc': r.mtime_baseline_utc, + 'bitrot_suspected': r.bitrot_suspected, + 'bitrot_candidate_hash': r.bitrot_candidate_hash + } + for r in rows + ] + batch.sort(key=lambda x: x['file_size'] if x['file_size'] else 0, + reverse=largest_first) + return batch + + class MaintenanceService: """Service for maintenance operations like cleanup and file monitoring""" @@ -748,7 +832,6 @@ def _create_cleanup_report(self, cleanup_record: CleanupState, orphaned_files_li # Store the list of orphaned files in directories_scanned field as JSON # This field is repurposed for cleanup reports to store the orphaned files list if orphaned_files_list: - import json report.directories_scanned = json.dumps(orphaned_files_list) db.session.add(report) @@ -771,16 +854,31 @@ def _create_cleanup_report(self, cleanup_record: CleanupState, orphaned_files_li # Don't fail the cleanup operation if report creation fails return None - def _run_file_changes_check(self, check_id: str, file_paths=None, schedule_id=None): + def _run_file_changes_check(self, check_id: str, file_paths=None, schedule_id=None, + time_budget_minutes=None): """Run the file changes check operation Args: check_id: Unique ID for this check file_paths: Optional list of specific file paths to check (if None, checks all files) schedule_id: Optional schedule ID for healthcheck integration + time_budget_minutes: Optional soft deadline. When it expires, no + new hash tasks are dispatched; in-flight tasks drain (still + stamping last_integrity_check_date) and the run completes. + The rolling queue resumes at the next run. """ # Store schedule_id for report creation self._file_changes_schedule_id = schedule_id + + # Resolve the budget here, not in the HTTP route, so every entry point + # (scheduler-triggered, manual API, service-level) honors a schedule's + # configured budget. An explicit argument always wins. + if time_budget_minutes is None and schedule_id: + schedule = db.session.get(ScanSchedule, schedule_id) + if schedule and schedule.time_budget_minutes: + time_budget_minutes = schedule.time_budget_minutes + logger.info(f"Using schedule {schedule_id} time budget: {time_budget_minutes} minutes") + try: # Use READ COMMITTED isolation level to reduce lock contention # This allows reads to see committed data without holding locks @@ -829,55 +927,35 @@ def _run_file_changes_check(self, check_id: str, file_paths=None, schedule_id=No logger.info(f"Starting Phase 2a: parallel hash calculation for {total_files} files using Celery workers") - # Import Celery task + # Imported here, not at module top: tasks.py pulls in the Celery + # app (and on the worker path, the Flask app), which must not load + # on service import from pixelprobe.tasks import calculate_file_hash_task - from celery import group - # Track dispatched tasks and changed files - task_results = [] + # Track changed files for the report changed_files = [] - # OPTIMIZATION: Load all files at once (like cleanup does) to avoid pagination issues - # This is more memory-intensive but eliminates the progressive slowdown problem - # with ID-based pagination on large datasets - logger.info(f"Loading all {total_files} files from database in a single query...") - - # Get database entries - either all or filtered by file_paths (same as cleanup) - # CRITICAL FIX v2.5.61: Load only needed columns as dictionaries to avoid - # detached instance errors during long-running scans (20+ hours for 1M+ files). - # ORM objects held in memory get expired by db.session.commit() calls, and if - # concurrent jobs delete rows, subsequent attribute access crashes with - # "Instance has been deleted" errors. - if file_paths: - results_query = db.session.query( - ScanResult.id, - ScanResult.file_path, - ScanResult.file_hash, - ScanResult.file_size, - ScanResult.last_modified - ).filter(ScanResult.file_path.in_(file_paths)).all() - logger.info(f"Loaded {len(results_query)} specific files from database") - else: - results_query = db.session.query( - ScanResult.id, - ScanResult.file_path, - ScanResult.file_hash, - ScanResult.file_size, - ScanResult.last_modified - ).all() - logger.info(f"Loaded all {len(results_query)} files from database") - - # Convert to list of dicts immediately - immune to session expiration - all_results = [ - { - 'id': r.id, - 'file_path': r.file_path, - 'file_hash': r.file_hash, - 'file_size': r.file_size, - 'last_modified': r.last_modified - } - for r in results_query - ] + # Rolling integrity queue: instead of loading every row into + # memory in one query (hundreds of MB at 1M+ files), the queue is + # re-fetched in ordered batches, stalest first. Files are stamped + # with last_integrity_check_date on ANY terminal outcome, pushing + # them behind run_watermark, so a crashed or cancelled run resumes + # where it left off by construction. + batch_size = env_int('INTEGRITY_BATCH_SIZE', 10000, floor=100) + failed_stamp_ids = set() + timestamp_write_failures = 0 + dispatch_failures = 0 + run_watermark = datetime.now(timezone.utc) + + # Soft budget deadline: stops NEW dispatch only; in-flight tasks + # drain to completion, so overshoot is bounded by the size-aware + # concurrency slots. Budgeted runs dispatch largest-first within + # each batch so a huge file is not started at the last minute. + budget_deadline = None + budget_expired = False + if time_budget_minutes: + budget_deadline = time.monotonic() + time_budget_minutes * 60 + logger.info(f"Integrity run budget: {time_budget_minutes} minutes") # CRITICAL FIX v2.4.60: Adaptive memory-aware task management # Problem: Files vary from 1KB to 40GB, systems vary in resources @@ -922,6 +1000,8 @@ def _run_file_changes_check(self, check_id: str, file_paths=None, schedule_id=No MAX_CONCURRENT_MEDIUM = min(max_safe_tasks // 10, int(os.environ.get('MAX_CONCURRENT_MEDIUM', '500'))) MAX_CONCURRENT_LARGE = min(max_safe_tasks // 100, int(os.environ.get('MAX_CONCURRENT_LARGE', '50'))) MAX_CONCURRENT_HUGE = min(max_safe_tasks // 1000, int(os.environ.get('MAX_CONCURRENT_HUGE', '5'))) + tier_limits = {'small': MAX_CONCURRENT_SMALL, 'medium': MAX_CONCURRENT_MEDIUM, + 'large': MAX_CONCURRENT_LARGE, 'huge': MAX_CONCURRENT_HUGE} # Track active tasks active_tasks = [] @@ -929,30 +1009,23 @@ def _run_file_changes_check(self, check_id: str, file_paths=None, schedule_id=No files_queued = 0 files_abandoned = 0 last_progress_update = 0 - task_results = [] last_heartbeat_time = time.time() - logger.info(f"Processing {len(all_results)} files with size-aware batching...") + logger.info(f"Processing {total_files} files with size-aware batching (queue batch size {batch_size})...") # Set initial progress to show we're starting file_changes_record.phase_current = 0 - file_changes_record.phase_total = len(all_results) + file_changes_record.phase_total = total_files file_changes_record.files_processed = 0 - file_changes_record.progress_message = f'Phase 2 of 3: Starting to process {len(all_results):,} files...' + file_changes_record.progress_message = f'Phase 2 of 3: Starting to process {total_files:,} files...' db.session.commit() # Force a small delay to allow UI to see initial progress time.sleep(0.1) - # Sort files by size for better batch management (small files first) - # Handle NULL file_size by treating as 0 for sorting - all_results_sorted = sorted(all_results, key=lambda x: x['file_size'] if x['file_size'] else 0) - - total_count = len(all_results) - def write_progress_snapshot(set_heartbeat: bool): - pct = int((total_files_processed / total_count * 100)) if total_count > 0 else 0 + pct = int((total_files_processed / total_files * 100)) if total_files > 0 else 0 msg = ( - f'Processing files: {total_files_processed:,}/{total_count:,} ({pct}%) - ' + f'Processing files: {total_files_processed:,}/{total_files:,} ({pct}%) - ' f'{len(changed_files)} changes found, {len(active_tasks)} active tasks' ) if set_heartbeat: @@ -964,13 +1037,89 @@ def write_progress_snapshot(set_heartbeat: bool): update_file_changes_progress_redis( check_id=check_id, files_processed=total_files_processed, - total_files=total_count, + total_files=total_files, phase=file_changes_record.phase or '', progress_message=msg, ) + # Terminal-outcome buffers, flushed in bulk before each queue + # re-fetch and at the end of the run: two UPDATE statements per + # flush instead of one SELECT + UPDATE per file (the dominant DB + # chatter at 1M-file scale). + stamp_ids = [] # rotate to the back of the queue + seen_ids = [] # file was readable again -> restore file_exists + rebaseline = [] # (id, mtime_iso): first check after the UTC fix + + def flush_stamps(): + # A row whose stamp fails would re-queue at the front forever, + # so failed ids are excluded from further fetches and counted. + nonlocal timestamp_write_failures + if not (stamp_ids or seen_ids or rebaseline): + return + try: + if stamp_ids: + db.session.query(ScanResult).filter( + ScanResult.id.in_(stamp_ids) + ).update({'last_integrity_check_date': datetime.now(timezone.utc)}, + synchronize_session=False) + if seen_ids: + # A file we just hashed exists, whatever a past run + # recorded (recovers rows stranded by a mount outage) + db.session.query(ScanResult).filter( + ScanResult.id.in_(seen_ids), + ScanResult.file_exists == False + ).update({'file_exists': True}, synchronize_session=False) + for file_id, mtime_iso in rebaseline: + # Hash matched but the stored mtime predates the UTC + # fix: refresh it so bitrot classification can trust + # this baseline from the next check onward. + file_record = db.session.get(ScanResult, file_id) + if file_record and not file_record.bitrot_suspected: + file_record.last_modified = datetime.fromisoformat(mtime_iso) + file_record.mtime_baseline_utc = True + db.session.commit() + except Exception as e: + db.session.rollback() + timestamp_write_failures += len(stamp_ids) + failed_stamp_ids.update(stamp_ids) + logger.error(f"Failed to flush {len(stamp_ids)} integrity stamps: {e}") + finally: + stamp_ids.clear() + seen_ids.clear() + rebaseline.clear() + + def fetch_next_batch(): + # Flush first so completed files' fresh timestamps are visible + # to the ordering and they drop out of the queue front. + flush_stamps() + db.session.commit() + in_flight = {t['id'] for t in active_tasks} + return fetch_integrity_batch( + file_paths, run_watermark, + in_flight | failed_stamp_ids, batch_size, + largest_first=budget_deadline is not None + ) + + pending = [] file_index = 0 - while file_index < len(all_results_sorted) or active_tasks: + queue_exhausted = False + # A refill is productive only after a dispatch, a completion, or a + # failure-exclusion changed the queue; otherwise re-fetching would + # return the same rows in a tight query loop (e.g. every size tier + # saturated, or a broker outage failing every dispatch). + refill_allowed = True + while True: + # Refill from the queue once the current batch is fully dispatched + if file_index >= len(pending) and not queue_exhausted and refill_allowed: + pending = fetch_next_batch() + file_index = 0 + refill_allowed = False + if not pending: + queue_exhausted = True + + if queue_exhausted and file_index >= len(pending) and not active_tasks: + break + # Heartbeat every 10 s: write-through to PostgreSQL + Redis so the # UI sees motion even when the periodic-delta block below hasn't # fired in the current iteration. phase_total is invariant for @@ -979,17 +1128,30 @@ def write_progress_snapshot(set_heartbeat: bool): if current_time - last_heartbeat_time >= 10: write_progress_snapshot(set_heartbeat=True) logger.info( - f"Progress: {total_files_processed}/{total_count} processed, " - f"{len(active_tasks)} active, {len(all_results_sorted) - file_index} remaining, " + f"Progress: {total_files_processed}/{total_files} processed, " + f"{len(active_tasks)} active, {len(pending) - file_index} remaining in batch, " f"{files_abandoned} abandoned" ) last_heartbeat_time = current_time # Check for cancellation if self._is_cancelled_file_changes(file_changes_record): - logger.info(f"Cancelled at {total_files_processed}/{len(all_results)} files") + logger.info(f"Cancelled at {total_files_processed}/{total_files} files") break + # Budget deadline: stop dispatching, let in-flight tasks drain + # (they still stamp timestamps), then complete normally. + if budget_deadline and not budget_expired and time.monotonic() >= budget_deadline: + budget_expired = True + pending = [] + file_index = 0 + queue_exhausted = True + logger.info( + f"Integrity run budget of {time_budget_minutes} minutes reached at " + f"{total_files_processed}/{total_files} files; draining " + f"{len(active_tasks)} in-flight tasks" + ) + # Collect completed tasks and free up slots still_active = [] for task_info in active_tasks: @@ -998,17 +1160,19 @@ def write_progress_snapshot(set_heartbeat: bool): try: result = safe_task_get(task, timeout=1) total_files_processed += 1 + refill_allowed = True - # Update last integrity check timestamp for this file - try: - file_record = ScanResult.query.filter_by(file_path=result['file_path']).first() - if file_record: - file_record.last_integrity_check_date = datetime.now(timezone.utc) - except Exception as e: - logger.error(f"Error updating last_integrity_check_date for {result['file_path']}: {e}") + # Terminal outcome: rotate to the back of the queue + stamp_ids.append(result['file_id']) + if result.get('current_hash'): + seen_ids.append(result['file_id']) + if (result['change_type'] == 'unchanged' + and not task_info.get('mtime_trusted') + and result.get('current_modified')): + rebaseline.append((result['file_id'], result['current_modified'])) # For single file scans, update progress immediately so UI can see it - if len(all_results) == 1: + if total_files == 1: file_changes_record.phase_current = total_files_processed file_changes_record.phase_total = 1 file_changes_record.progress_message = f'Phase 2 of 3: Completed checking file' @@ -1032,24 +1196,37 @@ def write_progress_snapshot(set_heartbeat: bool): if result.get('changed'): changed_files.append({ + 'file_id': result['file_id'], 'file_path': result['file_path'], 'change_type': result['change_type'], 'stored_hash': result['stored_hash'], - 'current_hash': result['current_hash'] + 'current_hash': result['current_hash'], + 'stored_modified': result.get('stored_modified'), + 'current_modified': result.get('current_modified'), + 'stored_size': result.get('stored_size'), + 'current_size': result.get('current_size') }) except Exception as e: logger.error(f"Error getting task result: {e}") + # Unreadable result is still a terminal outcome for + # this run - stamp it or it re-queues at the front. + if task_info.get('id') is not None: + stamp_ids.append(task_info['id']) + refill_allowed = True finally: forget_task_result(task) elif abandon_if_stuck(task_info, INTEGRITY_TASK_TIMEOUT_SECS, 'integrity'): files_abandoned += 1 + if task_info.get('id') is not None: + stamp_ids.append(task_info['id']) + refill_allowed = True else: still_active.append(task_info) active_tasks = still_active # Submit new tasks based on available slots - while file_index < len(all_results_sorted): - result = all_results_sorted[file_index] + while file_index < len(pending): + result = pending[file_index] # Use file_size from DB (it's a BigInteger field, might be NULL) file_size = result['file_size'] if result['file_size'] else 0 @@ -1068,14 +1245,8 @@ def write_progress_snapshot(set_heartbeat: bool): file_size = 10 * 1024 * 1024 # 10MB default # Determine max concurrent based on file size - if file_size < 10 * 1024 * 1024: # < 10MB - max_concurrent = MAX_CONCURRENT_SMALL - elif file_size < 100 * 1024 * 1024: # < 100MB - max_concurrent = MAX_CONCURRENT_MEDIUM - elif file_size < 1024 * 1024 * 1024: # < 1GB - max_concurrent = MAX_CONCURRENT_LARGE - else: # >= 1GB - max_concurrent = MAX_CONCURRENT_HUGE + tier = _size_tier(file_size) + max_concurrent = tier_limits[tier] # Count how many tasks of this size category are active size_category_active = sum(1 for t in active_tasks @@ -1083,27 +1254,40 @@ def write_progress_snapshot(set_heartbeat: bool): # Check if we can submit this task if len(active_tasks) >= max_concurrent or size_category_active >= max_concurrent: - # Wait for tasks to complete before submitting more - break + # This size tier is saturated: skip past it instead of + # head-of-line blocking the whole batch. Under the + # budgeted largest-first ordering the huge tier would + # otherwise cap the entire window at MAX_CONCURRENT_HUGE + # while thousands of small-file slots sit idle. Skipped + # files are never stamped, so later fetches return them. + file_index += 1 + while (file_index < len(pending) + and _size_tier(pending[file_index]['file_size'] or 0) == tier): + file_index += 1 + continue # Submit the task stored_modified_iso = result['last_modified'].isoformat() if result['last_modified'] else None try: task_result = calculate_file_hash_task.apply_async( - args=[result['id'], result['file_path'], result['file_hash'], stored_modified_iso] + args=[result['id'], result['file_path'], result['file_hash'], stored_modified_iso, + result['file_size'], bool(result['mtime_baseline_utc']), + bool(result['bitrot_suspected']), result['bitrot_candidate_hash']] ) active_tasks.append({ 'task': task_result, + 'id': result['id'], 'size': file_size, 'path': result['file_path'], + 'mtime_trusted': bool(result['mtime_baseline_utc']), 'submitted_at': time.monotonic(), }) - task_results.append(task_result) files_queued += 1 file_index += 1 + refill_allowed = True # Update progress immediately when processing single files - if len(all_results) == 1: + if total_files == 1: file_changes_record.progress_message = f'Phase 2 of 3: Processing {result["file_path"].split("/")[-1]}...' db.session.commit() except Exception as e: @@ -1111,23 +1295,29 @@ def write_progress_snapshot(set_heartbeat: bool): if "maxmemory" in str(e): logger.warning("Redis memory full, waiting for tasks to complete...") break # Wait for active tasks to complete + # Exclude from re-fetch: the rolling queue would + # otherwise return this row forever (e.g. broker + # outage), spinning the producer without progress. + failed_stamp_ids.add(result['id']) + dispatch_failures += 1 + refill_allowed = True file_index += 1 # Skip this file - # If no new tasks submitted and active tasks exist, wait a bit - if file_index < len(all_results_sorted) and len(active_tasks) > 0: + # If slots are full or the queue is draining, wait a bit + if len(active_tasks) > 0 and (file_index < len(pending) or queue_exhausted): time.sleep(0.1) # Brief sleep to avoid busy waiting # Update progress while waiting for single file - if len(all_results) == 1 and len(active_tasks) > 0: + if total_files == 1: file_changes_record.progress_message = f'Phase 2 of 3: Checking file for changes...' file_changes_record.phase_current = 0 file_changes_record.phase_total = 1 db.session.commit() # Update progress periodically - every file for tiny sets, every 10 for small, every 100 for larger - if len(all_results) <= 10: + if total_files <= 10: update_interval = 1 # Update after every file for 10 or fewer files - elif len(all_results) < 1000: + elif total_files < 1000: update_interval = 10 else: update_interval = 100 @@ -1139,21 +1329,34 @@ def write_progress_snapshot(set_heartbeat: bool): # 5000-active steady state almost never does. if total_files_processed > 0 and ( total_files_processed - last_progress_update >= update_interval or - total_files_processed == total_count + total_files_processed == total_files ): write_progress_snapshot(set_heartbeat=False) last_progress_update = total_files_processed + # Persist any buffered terminal outcomes before finalizing + flush_stamps() + # Final update file_changes_record.phase_current = total_files_processed - file_changes_record.phase_total = len(all_results) - pct = int((total_files_processed / len(all_results) * 100)) if len(all_results) > 0 else 0 + file_changes_record.phase_total = total_files file_changes_record.progress_message = ( - f'Completed: {total_files_processed:,}/{len(all_results):,} files - ' + f'Completed: {total_files_processed:,}/{total_files:,} files - ' f'{len(changed_files)} changes found' ) db.session.commit() + if timestamp_write_failures: + logger.warning( + f"{timestamp_write_failures} last_integrity_check_date updates failed; " + f"affected files will re-queue at the front of the next run" + ) + if dispatch_failures: + logger.warning( + f"{dispatch_failures} files could not be dispatched (broker errors); " + f"they remain queued for the next run" + ) + logger.info(f"Phase 2a complete: Processed {total_files_processed} files, found {len(changed_files)} changed files") # Store changed files for report generation @@ -1169,11 +1372,13 @@ def write_progress_snapshot(set_heartbeat: bool): file_changes_record.last_heartbeat = datetime.now(timezone.utc) db.session.commit() - logger.info(f"Starting Phase 3: Marking {len(changed_files)} changed files as pending for rescan") + logger.info(f"Starting Phase 3: Applying classification for {len(changed_files)} changed files") files_marked = 0 modified_count = 0 deleted_count = 0 + category_counts = {} + flagged_paths = [] last_heartbeat_time = time.time() for i, change_info in enumerate(changed_files): @@ -1181,7 +1386,7 @@ def write_progress_snapshot(set_heartbeat: bool): current_time = time.time() if current_time - last_heartbeat_time >= 30: file_changes_record.last_heartbeat = datetime.now(timezone.utc) - logger.info(f"Phase 3 heartbeat: {files_marked}/{len(changed_files)} files marked") + logger.info(f"Phase 3 heartbeat: {files_marked}/{len(changed_files)} files processed") db.session.commit() last_heartbeat_time = current_time @@ -1190,24 +1395,31 @@ def write_progress_snapshot(set_heartbeat: bool): file_changes_record.phase_current = i + 1 - # Mark the file as pending for rescan by the regular scan workers + # Apply per-classification state transitions (rescan marking, + # bitrot flagging/expiry, deletion recording) try: - file_record = ScanResult.query.filter_by(file_path=change_info['file_path']).first() + file_record = None + if change_info.get('file_id') is not None: + file_record = db.session.get(ScanResult, change_info['file_id']) + if not file_record: + file_record = ScanResult.query.filter_by(file_path=change_info['file_path']).first() if file_record: - file_record.scan_status = 'pending' + category = self._apply_change_classification(file_record, change_info) + category_counts[category] = category_counts.get(category, 0) + 1 files_marked += 1 + if category in ('bitrot_new', 'bitrot_active'): + flagged_paths.append(change_info['file_path']) - # Count by type - if change_info['change_type'] == 'deleted': + if category == 'deleted': deleted_count += 1 - logger.info(f"Marked deleted file for cleanup: {change_info['file_path']}") - elif change_info['change_type'] == 'modified': + logger.info(f"Recorded deleted file: {change_info['file_path']}") + elif category == 'modified': modified_count += 1 logger.info(f"Marked modified file for rescan: {change_info['file_path']} (hash: {change_info['stored_hash'][:16]}... -> {change_info['current_hash'][:16]}...)") else: - logger.info(f"Marked file for rescan: {change_info['file_path']} (type: {change_info['change_type']})") + logger.info(f"Processed changed file: {change_info['file_path']} (type: {change_info['change_type']}, category: {category})") except Exception as e: - logger.error(f"Error marking file for rescan {change_info['file_path']}: {e}") + logger.error(f"Error processing changed file {change_info['file_path']}: {e}") # Commit every 10 files for real-time progress updates if (i + 1) % 10 == 0: @@ -1216,14 +1428,21 @@ def write_progress_snapshot(set_heartbeat: bool): logger.info(f"Phase 3: Marked {files_marked}/{len(changed_files)} files for rescan") # Final commit for remaining files and update counts - file_changes_record.progress_message = f'Phase 3 of 3: Marked {files_marked}/{len(changed_files)} files as pending for rescan' + file_changes_record.progress_message = f'Phase 3 of 3: Processed {files_marked}/{len(changed_files)} changed files' file_changes_record.changes_found = modified_count db.session.commit() - logger.info(f"Phase 3 complete: Marked {files_marked} files for rescan ({modified_count} modified, {deleted_count} deleted). They will be processed by parallel scan workers.") + logger.info(f"Phase 3 complete: Processed {files_marked} changed files " + f"({modified_count} modified, {deleted_count} deleted, " + f"categories: {category_counts}). Rescans run via parallel scan workers.") + + # After the final commit: a delivery-status rollback inside + # dispatch_event can no longer clobber uncommitted flag state + self._notify_bitrot_summary(flagged_paths) else: # No changes found - set counts to 0 modified_count = 0 deleted_count = 0 + category_counts = {} # Complete check if self._is_cancelled_file_changes(file_changes_record): @@ -1231,11 +1450,34 @@ def write_progress_snapshot(set_heartbeat: bool): file_changes_record.progress_message = 'File changes check cancelled by user' else: file_changes_record.phase = 'complete' - file_changes_record.progress_message = ( - f'Check complete. Found {len(changed_files)} changed files ' - f'({modified_count} modified, {deleted_count} deleted), ' - f'{file_changes_record.corrupted_found} newly corrupted.' - ) + if budget_expired: + file_changes_record.progress_message = ( + f'Budget reached: {total_files_processed:,} of {total_files:,} files ' + f'verified this run ({len(changed_files)} changed, {modified_count} modified, ' + f'{deleted_count} deleted). Queue resumes at next run.' + ) + else: + file_changes_record.progress_message = ( + f'Check complete. Found {len(changed_files)} changed files ' + f'({modified_count} modified, {deleted_count} deleted), ' + f'{file_changes_record.corrupted_found} newly corrupted.' + ) + bitrot_flagged = (category_counts.get('bitrot_new', 0) + + category_counts.get('bitrot_active', 0)) + if bitrot_flagged: + file_changes_record.progress_message += ( + f' WARNING: {bitrot_flagged} file(s) flagged as suspected bitrot.' + ) + if timestamp_write_failures: + file_changes_record.progress_message += ( + f' Warning: {timestamp_write_failures} integrity timestamp ' + f'updates failed; those files re-queue first next run.' + ) + if dispatch_failures: + file_changes_record.progress_message += ( + f' Warning: {dispatch_failures} files could not be dispatched ' + f'(broker errors); they remain queued for the next run.' + ) file_changes_record.is_active = False file_changes_record.end_time = datetime.now(timezone.utc) @@ -1288,6 +1530,117 @@ def _is_cancelled(self, cleanup_record: CleanupState) -> bool: logger.warning(f"Error checking cancel status from DB: {e}") return self.cleanup_state.get('cancel_requested', False) + def _apply_change_classification(self, file_record, change_info): + """Phase 3 state transitions for one changed file. + + Returns the counting category ('modified', 'deleted', 'bitrot_new', + 'bitrot_active', 'bitrot_stable', 'bitrot_expired', + 'bitrot_self_healed', 'other'). + """ + change_type = change_info['change_type'] + now = datetime.now(timezone.utc) + + if change_type == 'deleted': + # Row deletion stays exclusively with orphan cleanup (its + # mount-outage mass-delete guard is the single gate for + # destructive action); the integrity check only records absence. + file_record.file_exists = False + return 'deleted' + + # The classification was computed at dispatch time; on budgeted runs + # hours can pass before this applies. If the flag was cleared in the + # meantime (manual accept, auto-expire), the queued bitrot-state + # result is stale - do not resurrect state-machine fields. + if change_type in ('bitrot_stable', 'bitrot_self_healed', 'bitrot_active') \ + and not file_record.bitrot_suspected: + return 'other' + + if change_type == 'bitrot_suspected': + file_record.bitrot_suspected = True + # First detection date is permanent; re-detections refresh details only + file_record.bitrot_detected_date = file_record.bitrot_detected_date or now + file_record.bitrot_details = json.dumps({ + 'stored_hash': change_info.get('stored_hash'), + 'current_hash': change_info.get('current_hash'), + 'stored_modified': change_info.get('stored_modified'), + 'current_modified': change_info.get('current_modified'), + 'stored_size': change_info.get('stored_size'), + 'current_size': change_info.get('current_size'), + 'detected': now.isoformat(), + }) + file_record.bitrot_candidate_hash = change_info.get('current_hash') + file_record.bitrot_stable_checks = 0 + file_record.scan_status = 'pending' # rescan checks decodability + return 'bitrot_new' + + if change_type == 'bitrot_active': + # Content changed AGAIN while flagged: active rot. New stability + # reference, counter restarts. + file_record.bitrot_candidate_hash = change_info.get('current_hash') + file_record.bitrot_stable_checks = 0 + file_record.scan_status = 'pending' + return 'bitrot_active' + + if change_type == 'bitrot_stable': + file_record.bitrot_stable_checks = (file_record.bitrot_stable_checks or 0) + 1 + rescan_clean = (file_record.scan_status == 'completed' + and file_record.is_corrupted is False) + if (file_record.bitrot_stable_checks >= BITROT_STABLE_CHECKS_TO_EXPIRE + and rescan_clean and file_record.bitrot_candidate_hash): + # Auto-expire: adopt the stable content as the new baseline. + # Detection date/details stay - permanent record. + mtime = None + if change_info.get('current_modified'): + mtime = datetime.fromisoformat(change_info['current_modified']) + adopt_bitrot_baseline(file_record, mtime) + logger.info(f"Bitrot flag auto-expired for {file_record.file_path}: " + f"hash stable and rescan clean") + return 'bitrot_expired' + return 'bitrot_stable' + + if change_type == 'bitrot_self_healed': + # Content matches the ORIGINAL baseline again: transient read + # anomaly, not rot. Clear the flag, keep the detection record. + file_record.bitrot_suspected = False + file_record.bitrot_candidate_hash = None + file_record.bitrot_stable_checks = 0 + logger.info(f"Bitrot flag cleared for {file_record.file_path}: " + f"content matches original baseline (self-healed)") + return 'bitrot_self_healed' + + # 'modified', 'no_hash', 'error': rescan re-baselines via the scan path + file_record.scan_status = 'pending' + return 'modified' if change_type == 'modified' else 'other' + + def _notify_bitrot_summary(self, flagged_paths): + """One aggregated bitrot_suspected notification per run (best-effort). + + Per-file dispatch would mean thousands of serial HTTP round-trips and + mid-loop session commits in the dying-disk case this feature targets + (a dispatch failure rollback would also discard uncommitted flag + state). Called only after Phase 3's final commit. + """ + if not flagged_paths: + return + try: + shown = '\n'.join(flagged_paths[:10]) + if len(flagged_paths) > 10: + shown += f'\n... and {len(flagged_paths) - 10} more' + dispatch_event( + 'bitrot_suspected', + f'Bitrot suspected on {len(flagged_paths)} file(s)', + 'Content hash changed while file modification time did not - ' + 'possible silent corruption. Review the files, then accept ' + 'their current state or restore from backup.\n' + shown, + priority='high', + additional_data={ + 'count': len(flagged_paths), + 'file_paths': flagged_paths[:50], + } + ) + except Exception as e: + logger.error(f"Failed to dispatch bitrot notification: {e}") + def _is_cancelled_file_changes(self, record: FileChangesState) -> bool: """Check if file changes check has been cancelled""" try: @@ -1456,7 +1809,6 @@ def _create_file_changes_report(self, file_changes_record: FileChangesState, cha # Store the list of changed files with hash comparison details in directories_scanned field as JSON # This field is repurposed for file changes reports to store the changed files list with hash info if changed_files_list: - import json report.directories_scanned = json.dumps(changed_files_list) db.session.add(report) diff --git a/pixelprobe/services/notification_service.py b/pixelprobe/services/notification_service.py index 8f10ca3..4e21634 100644 --- a/pixelprobe/services/notification_service.py +++ b/pixelprobe/services/notification_service.py @@ -11,11 +11,68 @@ import requests from typing import Dict, Optional, List from datetime import datetime, timezone +from pixelprobe.models import db, NotificationRule from pixelprobe.utils.security import validate_safe_url, create_safe_session logger = logging.getLogger(__name__) +def dispatch_event(event_type, title, message, priority='normal', additional_data=None): + """Send an event through every active NotificationRule for event_type. + + This is the rule-evaluation layer the CRUD-only rules previously lacked: + it joins active rules to their active providers and delivers via + NotificationService, recording per-provider delivery status. Returns the + number of successful deliveries. Failures are logged, never raised - + notification must not break the operation that triggered it. + """ + try: + rules = NotificationRule.query.filter_by(event_type=event_type, is_active=True).all() + except Exception as e: + logger.error(f"Could not load notification rules for {event_type}: {e}") + return 0 + + if not rules: + logger.debug(f"No active notification rules for event {event_type}") + return 0 + + service = NotificationService() + sent = 0 + for rule in rules: + provider = rule.provider + if not provider or not provider.is_active: + continue + try: + success, error = service.send_notification( + provider_type=provider.provider_type, + provider_config=provider.configuration, + title=title, + message=message, + # rule.priority defaults to 'normal' (NOT NULL), so it cannot + # simply win over the event: an explicitly raised/lowered rule + # priority applies, otherwise the event's priority does (e.g. + # bitrot dispatches at 'high') + priority=rule.priority if rule.priority and rule.priority != 'normal' else priority, + additional_data=additional_data + ) + provider.last_notification_status = 'success' if success else 'failure' + provider.last_notification_time = datetime.now(timezone.utc) + if success: + sent += 1 + else: + logger.warning(f"Notification via {provider.name} failed for {event_type}: {error}") + except Exception as e: + logger.error(f"Notification via provider {provider.id} raised for {event_type}: {e}") + + try: + db.session.commit() + except Exception as e: + logger.error(f"Could not record notification delivery status: {e}") + db.session.rollback() + + return sent + + class NotificationService: """Service for sending notifications via various providers""" diff --git a/pixelprobe/services/stats_service.py b/pixelprobe/services/stats_service.py index 373cb68..35b09a0 100644 --- a/pixelprobe/services/stats_service.py +++ b/pixelprobe/services/stats_service.py @@ -5,7 +5,7 @@ import os import logging from typing import Dict, List -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from sqlalchemy import text, func from pixelprobe.models import db, ScanResult, ScanReport @@ -57,6 +57,49 @@ def get_file_statistics(self) -> Dict: # Fallback to individual queries return self._get_stats_fallback() + def get_integrity_coverage(self) -> Dict: + """Cumulative rolling-integrity coverage. + + The integrity check sweeps the library stalest-first in budgeted + slices, so no single run report answers "has every file been + verified?". Coverage counts files by last_integrity_check_date: + checked ever and within the last 30 days, plus the oldest check date + (every checked file has been verified at least once since then) and + how many files have never been checked. + """ + try: + cutoff = datetime.now(timezone.utc) - timedelta(days=30) + row = db.session.execute( + text(""" + SELECT + COUNT(*) as total_files, + SUM(CASE WHEN last_integrity_check_date IS NOT NULL THEN 1 ELSE 0 END) as checked_files, + SUM(CASE WHEN last_integrity_check_date >= :cutoff THEN 1 ELSE 0 END) as checked_last_30_days, + MIN(last_integrity_check_date) as oldest_check, + SUM(CASE WHEN bitrot_suspected = TRUE THEN 1 ELSE 0 END) as bitrot_suspected + FROM scan_results + """), + {'cutoff': cutoff} + ).fetchone() + + total = row[0] or 0 + checked = row[1] or 0 + oldest = row[3] + if oldest is not None and hasattr(oldest, 'isoformat'): + oldest = oldest.isoformat() + return { + 'total_files': total, + 'checked_files': checked, + 'checked_percent': round(checked / total * 100, 1) if total else 0.0, + 'checked_last_30_days': row[2] or 0, + 'never_checked': total - checked, + 'oldest_check_date': oldest, + 'bitrot_suspected': row[4] or 0, + } + except Exception as e: + logger.error(f"Error getting integrity coverage: {e}") + return {} + def get_system_info(self) -> Dict: """Get comprehensive system information""" try: diff --git a/pixelprobe/tasks.py b/pixelprobe/tasks.py index f6479bf..1cd75ce 100644 --- a/pixelprobe/tasks.py +++ b/pixelprobe/tasks.py @@ -16,6 +16,7 @@ from pixelprobe.services.scan_service import ScanService from pixelprobe.models import db, ScanState, ScanResult, ScanReport from pixelprobe.utils.celery_utils import is_db_connection_corruption +from pixelprobe.utils.integrity import classify_file_change from pixelprobe.utils.log_context import current_scan_id, current_celery_task_id @@ -372,21 +373,35 @@ def check_file_exists_task(self, file_id, file_path): @celery_app.task(bind=True, max_retries=2, soft_time_limit=None, time_limit=None, # No timeout - must complete hash regardless of file size priority=7) # Low priority - maintenance runs in background -def calculate_file_hash_task(self, file_id, file_path, stored_hash, stored_modified): +def calculate_file_hash_task(self, file_id, file_path, stored_hash, stored_modified, + stored_size=None, mtime_trusted=False, + bitrot_suspected=False, bitrot_candidate_hash=None): """ - Calculate hash for a single file and compare to stored hash + Calculate hash for a single file and classify any change. IMPORTANT: No timeouts on this task - we must always calculate hash regardless of how large the file is. File integrity checking requires accurate hashes. + Classification: a hash mismatch with a changed mtime is a legitimate + modification; a mismatch with an UNCHANGED mtime is suspected bitrot (no + legitimate write path alters content without touching mtime). Files + already flagged are measured against the candidate hash for the + auto-expire state machine (stable / self-healed / active rot). + Args: file_id (int): Database ID of the file file_path (str): Path to the file stored_hash (str): Expected hash from database stored_modified (str): Last modified timestamp from database (ISO format) + stored_size (int): File size from database (for the detection record) + mtime_trusted (bool): ScanResult.mtime_baseline_utc - pre-upgrade + naive-local baselines cannot be classified and fall back to + 'modified', which re-baselines in UTC + bitrot_suspected (bool): File is already flagged + bitrot_candidate_hash (str): Stability reference for flagged files Returns: - dict: Hash comparison result with change information + dict: Hash comparison result with change classification """ # ContextTask wrapper in celery_config.py automatically provides Flask app context # NOTE: Do NOT call db.session.remove() here - it corrupts the connection pool @@ -450,26 +465,14 @@ def calculate_file_hash_task(self, file_id, file_path, stored_hash, stored_modif # Get current modification time stat = os.stat(file_path) current_modified = datetime.fromtimestamp(stat.st_mtime, timezone.utc) + current_size = stat.st_size - # Parse stored modified time - if stored_modified: - try: - stored_mod_dt = datetime.fromisoformat(stored_modified.replace('Z', '+00:00')) - except: - stored_mod_dt = None - else: - stored_mod_dt = None - - # Determine if file changed - if not stored_hash: - change_type = 'no_hash' - changed = True - elif current_hash != stored_hash: - change_type = 'modified' - changed = True - else: - change_type = 'unchanged' - changed = False + change_type, changed = classify_file_change( + stored_hash, current_hash, stored_modified, current_modified, + mtime_trusted=mtime_trusted, + bitrot_suspected=bitrot_suspected, + bitrot_candidate_hash=bitrot_candidate_hash + ) return { 'file_id': file_id, @@ -479,7 +482,9 @@ def calculate_file_hash_task(self, file_id, file_path, stored_hash, stored_modif 'stored_hash': stored_hash, 'current_hash': current_hash, 'stored_modified': stored_modified, - 'current_modified': current_modified.isoformat() if current_modified else None + 'current_modified': current_modified.isoformat() if current_modified else None, + 'stored_size': stored_size, + 'current_size': current_size } except Exception as exc: diff --git a/pixelprobe/tasks_parallel.py b/pixelprobe/tasks_parallel.py index 9c8b26d..5f7b1dd 100644 --- a/pixelprobe/tasks_parallel.py +++ b/pixelprobe/tasks_parallel.py @@ -25,6 +25,7 @@ from pixelprobe.models import db, ScanState, ScanResult, ScanChunk from pixelprobe.media_checker import PixelProbe, load_exclusions_with_patterns from pixelprobe.progress_utils import clear_scan_progress_redis, update_scan_progress_redis +from pixelprobe.utils.integrity import apply_scan_baseline from pixelprobe.services.scan_engine import ( build_scan_chunks, claim_scan_slot, finalize_scan, maybe_finalize_scan, sync_progress_from_chunks @@ -197,7 +198,13 @@ def process_chunk_task(self, chunk_db_id: int, scan_id: str, force_rescan: bool db_result.scan_output = str(scan_result.get('scan_output', ''))[:10000] db_result.has_warnings = has_warnings db_result.warning_details = warning_details - db_result.file_hash = scan_result.get('file_hash') + # Guarded baseline write: never overwrite a + # bitrot-suspected file's stored hash/mtime (this is + # the writer for all chunked scans, including the + # rescans Phase 3 of the integrity check queues up) + if not apply_scan_baseline(db_result, scan_result.get('file_hash'), + scan_result.get('last_modified')): + logger.info(f"Preserving hash/mtime baseline for bitrot-suspected file: {file_path}") db_result.scan_tool = scan_result.get('scan_tool', 'unknown') db_result.scan_duration = scan_result.get('scan_duration') db_result.file_size = scan_result.get('file_size', 0) diff --git a/pixelprobe/utils/integrity.py b/pixelprobe/utils/integrity.py new file mode 100644 index 0000000..397a185 --- /dev/null +++ b/pixelprobe/utils/integrity.py @@ -0,0 +1,102 @@ +"""Pure classification logic for the integrity check (bitrot detection). + +Kept free of Celery/app imports so it is unit-testable and reusable from +both the hash task and any future callers. +""" + +from datetime import datetime, timezone + +# A hash mismatch counts as "mtime unchanged" (bitrot signal) only within +# this tolerance, absorbing filesystem timestamp granularity (FAT-family, +# some network mounts) and float truncation. +MTIME_EPSILON_SECONDS = 2 + + +def classify_file_change(stored_hash, current_hash, stored_modified, current_modified, + mtime_trusted=False, bitrot_suspected=False, + bitrot_candidate_hash=None): + """Classify a hash comparison result. Returns (change_type, changed). + + Matrix (not already flagged): + hash match -> unchanged + no stored hash -> no_hash (baseline) + mismatch, untrusted baseline -> modified (re-baselines in UTC) + mismatch, mtime changed -> modified (legitimate edit/replace) + mismatch, mtime unchanged -> bitrot_suspected (no legitimate write + path alters content without mtime) + + Already flagged (bitrot_suspected=True): + current == candidate hash -> bitrot_stable (toward auto-expire) + current == stored baseline -> bitrot_self_healed (transient anomaly) + current == neither -> bitrot_active (rot progressing) + + Args: + stored_modified: ISO string from the database (naive = UTC when + mtime_trusted, unclassifiable otherwise) + current_modified: tz-aware datetime from os.stat + """ + if bitrot_suspected: + if bitrot_candidate_hash and current_hash == bitrot_candidate_hash: + return 'bitrot_stable', True + if stored_hash and current_hash == stored_hash: + return 'bitrot_self_healed', True + return 'bitrot_active', True + + if not stored_hash: + return 'no_hash', True + + if current_hash == stored_hash: + return 'unchanged', False + + stored_mod_dt = None + if stored_modified: + try: + stored_mod_dt = datetime.fromisoformat(stored_modified.replace('Z', '+00:00')) + except (ValueError, TypeError): + stored_mod_dt = None + # Trusted baselines are UTC; attach the timezone if the ISO string was naive + if stored_mod_dt is not None and stored_mod_dt.tzinfo is None: + stored_mod_dt = stored_mod_dt.replace(tzinfo=timezone.utc) + + mtime_unchanged = ( + mtime_trusted and stored_mod_dt is not None and current_modified is not None and + abs((current_modified - stored_mod_dt).total_seconds()) <= MTIME_EPSILON_SECONDS + ) + return ('bitrot_suspected' if mtime_unchanged else 'modified'), True + + +def apply_scan_baseline(row, file_hash, last_modified): + """Write a scan's hash/mtime baseline onto a ScanResult row. + + Single choke point for baseline writes from scan paths: refuses to + overwrite the baseline of a bitrot-suspected file (anti-laundering - a + bit flip can pass decode checks, so a rescan must not adopt suspect + content), and marks the mtime trusted (UTC) whenever a real mtime is + written. Returns True when the baseline was written. + """ + if row.bitrot_suspected: + return False + row.file_hash = file_hash + if last_modified is not None: + row.last_modified = last_modified + row.mtime_baseline_utc = True + return True + + +def adopt_bitrot_baseline(row, last_modified): + """Adopt the candidate hash as the new baseline and clear the bitrot flag. + + Shared by auto-expire and the manual accept action so the two paths + cannot drift. The detection record (bitrot_detected_date/bitrot_details) + is preserved permanently. last_modified may be None when the caller has + no trustworthy mtime; the row then re-baselines its mtime on the next + hash-match integrity check. + """ + if row.bitrot_candidate_hash: + row.file_hash = row.bitrot_candidate_hash + if last_modified is not None: + row.last_modified = last_modified + row.mtime_baseline_utc = True + row.bitrot_suspected = False + row.bitrot_candidate_hash = None + row.bitrot_stable_checks = 0 diff --git a/pixelprobe/utils/validators.py b/pixelprobe/utils/validators.py index a88bfe7..4dc52bf 100644 --- a/pixelprobe/utils/validators.py +++ b/pixelprobe/utils/validators.py @@ -51,6 +51,21 @@ def validate_cron_expression(cron_expr): # More detailed validation could be added here return True, None +def validate_time_budget(value, scan_type): + """Validate a time_budget_minutes payload value. + + Positive integer minutes; only meaningful for file_changes runs and + schedules. None means unlimited. Returns (value, error_message). + """ + if value is None: + return None, None + if isinstance(value, bool) or not isinstance(value, int) or value <= 0: + return None, 'time_budget_minutes must be a positive integer (minutes) or null' + if scan_type != 'file_changes': + return None, 'time_budget_minutes is only valid for file_changes schedules' + return value, None + + def validate_export_format(format_type): """Validate export format""" valid_formats = ['csv', 'json', 'xml'] diff --git a/pixelprobe/version.py b/pixelprobe/version.py index 448c5fe..a10d272 100644 --- a/pixelprobe/version.py +++ b/pixelprobe/version.py @@ -4,7 +4,7 @@ # Default version - this is the single source of truth -_DEFAULT_VERSION = '2.6.57' +_DEFAULT_VERSION = '2.6.64' # Allow override via environment variable for CI/CD, but default to the hardcoded version diff --git a/static/css/desktop.css b/static/css/desktop.css index c455a21..502764a 100644 --- a/static/css/desktop.css +++ b/static/css/desktop.css @@ -966,6 +966,11 @@ input:checked + .toggle-slider:before { color: var(--danger-color); } +.badge-bitrot { + background-color: rgba(111, 66, 193, 0.12); + color: #6f42c1; +} + /* Scan Output Modal */ .scan-output-details h4 { color: var(--text-primary); diff --git a/static/js/app.js b/static/js/app.js index 549b178..e9a3b07 100644 --- a/static/js/app.js +++ b/static/js/app.js @@ -8,6 +8,16 @@ function escapeHtml(text) { return div.innerHTML; } +// Single source of truth for a file's status badge (used by the desktop +// table, mobile cards, and the details modal - keep them in lockstep) +function fileStatus(file) { + if (file.marked_as_good) return { cls: 'success', text: 'Healthy' }; + if (file.bitrot_suspected) return { cls: 'bitrot', text: 'Bitrot?' }; + if (file.is_corrupted) return { cls: 'danger', text: 'Corrupted' }; + if (file.has_warnings) return { cls: 'warning', text: 'Warning' }; + return { cls: 'success', text: 'Healthy' }; +} + // Theme Management class ThemeManager { constructor() { @@ -238,6 +248,13 @@ class APIClient { }); } + async acceptBitrot(fileIds) { + return this.request('/bitrot/accept', { + method: 'POST', + body: JSON.stringify({ file_ids: fileIds }) + }); + } + async resetForRescan(resetType = 'all') { return this.request('/reset-for-rescan', { method: 'POST', @@ -343,6 +360,25 @@ class StatsDashboard { this.updateStatCard('warning-files', stats.warning_files || 0); this.updateStatCard('pending-files', stats.pending_files); this.updateStatCard('scanning-files', stats.scanning_files); + this.renderIntegrityCoverage(stats.integrity); + } + + renderIntegrityCoverage(integrity) { + const element = document.querySelector('#integrity-checked'); + if (!element || !integrity || !integrity.total_files) return; + element.textContent = `${integrity.checked_percent}%`; + let title = `${integrity.checked_files.toLocaleString()} of ` + + `${integrity.total_files.toLocaleString()} files integrity-checked`; + if (integrity.never_checked > 0) { + title += `; ${integrity.never_checked.toLocaleString()} never checked`; + } + if (integrity.oldest_check_date) { + title += `; every checked file verified since ${new Date(integrity.oldest_check_date).toLocaleString()}`; + } + if (integrity.bitrot_suspected > 0) { + title += `; ${integrity.bitrot_suspected.toLocaleString()} bitrot suspected`; + } + element.title = title; } updateStatCard(id, value) { @@ -1170,6 +1206,9 @@ class TableManager { case 'warning': params.has_warnings = 'true'; break; + case 'bitrot': + params.bitrot_suspected = 'true'; + break; case 'all': default: params.is_corrupted = 'all'; @@ -1230,8 +1269,9 @@ class TableManager { } renderMobileCard(file) { - const statusClass = file.marked_as_good ? 'success' : (file.is_corrupted ? 'danger' : (file.has_warnings ? 'warning' : 'success')); - const statusText = file.marked_as_good ? 'HEALTHY' : (file.is_corrupted ? 'CORRUPTED' : (file.has_warnings ? 'WARNING' : 'HEALTHY')); + const status = fileStatus(file); + const statusClass = status.cls; + const statusText = status.text.toUpperCase(); return `
@@ -1275,6 +1315,11 @@ class TableManager {
  • Integrity Check
    • + ${file.bitrot_suspected ? ` +
  • + Accept Current State +
    • + ` : ''}
    ${file.corruption_details || file.scan_output || file.error_message || file.warning_details ? ` @@ -1295,8 +1340,7 @@ class TableManager { } renderRow(file) { - const statusClass = file.marked_as_good ? 'success' : (file.is_corrupted ? 'danger' : (file.has_warnings ? 'warning' : 'success')); - const statusText = file.marked_as_good ? 'Healthy' : (file.is_corrupted ? 'Corrupted' : (file.has_warnings ? 'Warning' : 'Healthy')); + const { cls: statusClass, text: statusText } = fileStatus(file); return ` @@ -1328,6 +1372,11 @@ class TableManager {
  • Integrity Check
    • + ${file.bitrot_suspected ? ` +
  • + Accept Current State +
    • + ` : ''} ${file.corruption_details || file.scan_output || file.error_message || file.warning_details ? ` @@ -2311,6 +2360,49 @@ class PixelProbeApp { } } + async acceptBitrot(fileId) { + if (!confirm("Accept this file's current content as the new baseline? The bitrot flag will be cleared (the detection record is kept).")) { + return; + } + try { + await this.api.acceptBitrot([fileId]); + this.showNotification('Current state accepted as new baseline', 'success'); + await this.table.loadData(); + } catch (error) { + this.showNotification('Failed to accept current state', 'error'); + } + } + + renderBitrotDetails(file) { + if (!file.bitrot_suspected && !file.bitrot_detected_date) return ''; + let details = {}; + try { details = JSON.parse(file.bitrot_details || '{}'); } catch (e) { details = {}; } + const esc = (v) => this.escapeHtml(String(v == null || v === '' ? 'N/A' : v)); + const rows = [ + ['Detected', file.bitrot_detected_date ? new Date(file.bitrot_detected_date).toLocaleString() : null], + ['Baseline Hash', details.stored_hash], + ['Current Hash', details.current_hash || file.bitrot_candidate_hash], + ['Baseline mtime', details.stored_modified], + ['Current mtime', details.current_modified], + ['Baseline Size', details.stored_size], + ['Current Size', details.current_size], + ['Stable Checks', file.bitrot_stable_checks || 0] + ]; + return ` +
    +

    ${file.bitrot_suspected ? 'Bitrot Suspected' : 'Past Bitrot Detection (resolved)'}

    +

    Content hash changed while the file's modification time did not - possible silent corruption. Review the file, then accept its current state or restore from backup and rescan.

    + + ${rows.map(([label, value]) => ``).join('')} +
    ${esc(label)}:${esc(value)}
    + ${file.bitrot_suspected ? ` + + ` : ''} + `; + } + async markSelectedAsGood() { if (this.table.selectedFiles.size === 0) { this.showNotification('No files selected', 'warning'); @@ -3985,10 +4077,11 @@ class PixelProbeApp { modalBody.innerHTML = `

    File: ${this.escapeHtml(file.file_path)}

    -

    Status: ${file.marked_as_good ? 'Healthy' : (file.is_corrupted ? 'Corrupted' : (file.has_warnings ? 'Warning' : 'Healthy'))}

    +

    Status: ${fileStatus(file).text}

    Tool: ${file.scan_tool || 'N/A'}

    Scanned: ${file.scan_date ? new Date(file.scan_date).toLocaleString() : 'N/A'}

    ${file.last_integrity_check_date ? `

    Last Integrity Check: ${new Date(file.last_integrity_check_date).toLocaleString()}

    ` : ''} + ${this.renderBitrotDetails(file)}
    ${detailsHtml}
    @@ -4094,6 +4187,7 @@ class PixelProbeApp {

    Schedule: ${this.escapeHtml(schedule.cron_expression)}

    Type: ${this.formatScanType(schedule.scan_type || 'normal')}

    + ${schedule.time_budget_minutes ? `

    Time Budget: ${schedule.time_budget_minutes} min/run

    ` : ''}

    Next Run: ${nextRun}

    Last Run: ${lastRun}

    ${schedule.scan_paths && schedule.scan_paths.length > 0 ? `

    Paths: ${this.escapeHtml(schedule.scan_paths.join(', '))}

    ` : ''} @@ -4115,10 +4209,11 @@ class PixelProbeApp { const modal = document.querySelector('#add-schedule-modal'); if (modal) { modal.style.display = 'block'; - + // Reset form const form = document.querySelector('#add-schedule-form'); if (form) form.reset(); + this.toggleBudgetInput(); } } @@ -4150,6 +4245,12 @@ class PixelProbeApp { } } + toggleBudgetInput(prefix = '') { + const scanType = document.querySelector(`#${prefix}scan-type`).value; + document.querySelector(`#${prefix}budget-input`).style.display = + scanType === 'file_changes' ? 'block' : 'none'; + } + async showEditSchedule(scheduleId) { try { // Load schedule data @@ -4165,6 +4266,8 @@ class PixelProbeApp { document.getElementById('edit-schedule-paths').value = schedule.scan_paths ? schedule.scan_paths.join('\n') : ''; document.getElementById('edit-force-rescan').checked = schedule.force_rescan || false; document.getElementById('edit-is-active').checked = schedule.is_active !== undefined ? schedule.is_active : true; + document.getElementById('edit-time-budget-minutes').value = schedule.time_budget_minutes || ''; + this.toggleBudgetInput('edit-'); // Determine if it's cron or interval const cronExpression = schedule.cron_expression || ''; @@ -4220,6 +4323,9 @@ class PixelProbeApp { } const scanPaths = paths.trim() ? paths.split('\n').map(p => p.trim()).filter(p => p) : []; + const budgetValue = document.getElementById('edit-time-budget-minutes').value; + const timeBudget = (scanType === 'file_changes' && budgetValue) + ? parseInt(budgetValue, 10) : null; const response = await fetch(`/api/schedules/${scheduleId}`, { method: 'PUT', @@ -4230,7 +4336,8 @@ class PixelProbeApp { scan_type: scanType, scan_paths: scanPaths, force_rescan: forceRescan, - is_active: isActive + is_active: isActive, + time_budget_minutes: timeBudget }) }); @@ -4607,7 +4714,10 @@ document.addEventListener('DOMContentLoaded', () => { const pathsText = document.querySelector('#schedule-paths').value; const scanPaths = pathsText.trim() ? pathsText.split('\n').filter(p => p.trim()) : []; const scanType = document.querySelector('#scan-type').value; - + const budgetValue = document.querySelector('#time-budget-minutes').value; + const timeBudget = (scanType === 'file_changes' && budgetValue) + ? parseInt(budgetValue, 10) : null; + try { const response = await fetch('/api/schedules', { method: 'POST', @@ -4616,7 +4726,8 @@ document.addEventListener('DOMContentLoaded', () => { name: name, cron_expression: cronExpression, scan_paths: scanPaths, - scan_type: scanType + scan_type: scanType, + time_budget_minutes: timeBudget }) }); diff --git a/templates/index.html b/templates/index.html index 07d6a33..766f87d 100644 --- a/templates/index.html +++ b/templates/index.html @@ -193,6 +193,10 @@

    Scan Progress

    0
    Pending Files
    +
    +
    -
    +
    Integrity Checked
    +
    @@ -210,6 +214,9 @@

    Scan Progress

    + @@ -647,13 +654,18 @@

    Add Scan Schedule

    - Choose what type of operation to perform
    +
    @@ -707,13 +719,18 @@

    Edit Scan Schedule

    - Choose what type of operation to perform
    +
    diff --git a/tests/integration/test_bitrot_endpoints.py b/tests/integration/test_bitrot_endpoints.py new file mode 100644 index 0000000..914d1fa --- /dev/null +++ b/tests/integration/test_bitrot_endpoints.py @@ -0,0 +1,146 @@ +"""Integration tests for the bitrot filter and accept-current-state endpoint.""" + +from datetime import datetime, timezone + +from pixelprobe.models import ScanResult + +HASH_A = 'a' * 64 +HASH_B = 'b' * 64 + + +def seed(db, path, **kwargs): + defaults = dict( + file_path=path, + file_size=1000, + file_type='video/mp4', + scan_status='completed', + file_hash=HASH_A, + is_corrupted=False, + scan_date=datetime.now(timezone.utc), + ) + defaults.update(kwargs) + row = ScanResult(**defaults) + db.session.add(row) + db.session.commit() + return row + + +class TestBitrotFilter: + + def test_filter_returns_only_flagged(self, authenticated_client, app, db): + with app.app_context(): + seed(db, '/m/clean.mkv') + flagged = seed(db, '/m/rotten.mkv', bitrot_suspected=True, + bitrot_candidate_hash=HASH_B) + + response = authenticated_client.get('/api/scan-results?bitrot_suspected=true') + assert response.status_code == 200 + data = response.get_json() + paths = [r['file_path'] for r in data['results']] + assert paths == ['/m/rotten.mkv'] + assert data['results'][0]['bitrot_suspected'] is True + assert data['results'][0]['bitrot_candidate_hash'] == HASH_B + assert flagged.id == data['results'][0]['id'] + + def test_filter_false_excludes_flagged(self, authenticated_client, app, db): + with app.app_context(): + seed(db, '/m/clean.mkv') + seed(db, '/m/rotten.mkv', bitrot_suspected=True) + + response = authenticated_client.get('/api/scan-results?bitrot_suspected=false') + assert response.status_code == 200 + paths = [r['file_path'] for r in response.get_json()['results']] + assert paths == ['/m/clean.mkv'] + + +class TestIntegrityCoverageStats: + + def test_stats_include_rolling_coverage(self, authenticated_client, app, db): + with app.app_context(): + now = datetime.now(timezone.utc).replace(tzinfo=None) + seed(db, '/m/checked_recent.mkv', last_integrity_check_date=now) + seed(db, '/m/checked_old.mkv', + last_integrity_check_date=datetime(2025, 1, 1, 12, 0, 0)) + seed(db, '/m/never.mkv') + seed(db, '/m/rotten.mkv', bitrot_suspected=True, + last_integrity_check_date=now) + + response = authenticated_client.get('/api/stats') + assert response.status_code == 200 + integrity = response.get_json()['integrity'] + + assert integrity['total_files'] == 4 + assert integrity['checked_files'] == 3 + assert integrity['checked_percent'] == 75.0 + assert integrity['never_checked'] == 1 + assert integrity['checked_last_30_days'] == 2 + assert integrity['bitrot_suspected'] == 1 + assert integrity['oldest_check_date'].startswith('2025-01-01') + + +class TestBitrotAccept: + + def test_accept_adopts_candidate_and_clears_flag(self, authenticated_client, app, db): + with app.app_context(): + detected = datetime(2026, 7, 1, 12, 0, 0) + check_mtime = datetime(2026, 7, 1, 11, 0, 0) + details = ('{"stored_hash": "' + HASH_A + '", "current_hash": "' + HASH_B + '", ' + '"current_modified": "' + check_mtime.isoformat() + '"}') + row = seed(db, '/m/rotten.mkv', bitrot_suspected=True, + bitrot_candidate_hash=HASH_B, bitrot_stable_checks=1, + bitrot_detected_date=detected, bitrot_details=details) + + response = authenticated_client.post('/api/bitrot/accept', + json={'file_ids': [row.id]}) + assert response.status_code == 200 + assert response.get_json()['accepted'] == 1 + + db.session.refresh(row) + assert row.bitrot_suspected is False + assert row.file_hash == HASH_B + assert row.bitrot_candidate_hash is None + assert row.bitrot_stable_checks == 0 + # Baseline mtime is the one recorded WITH the candidate hash, not a + # fresh stat (the file may have changed again since the check) + assert row.last_modified == check_mtime + assert row.mtime_baseline_utc is True + # Detection record is permanent + assert row.bitrot_detected_date == detected + assert row.bitrot_details == details + + def test_accept_without_recorded_mtime_still_clears_flag(self, authenticated_client, app, db): + with app.app_context(): + original_mtime = datetime(2026, 6, 1, 8, 0, 0) + row = seed(db, '/m/rotten2.mkv', bitrot_suspected=True, + bitrot_candidate_hash=HASH_B, last_modified=original_mtime, + bitrot_details='{"x": 1}') + + response = authenticated_client.post('/api/bitrot/accept', + json={'file_ids': [row.id]}) + assert response.status_code == 200 + assert response.get_json()['accepted'] == 1 + + db.session.refresh(row) + assert row.bitrot_suspected is False + assert row.file_hash == HASH_B + # No recorded check mtime: stored mtime untouched and still + # untrusted; the next hash-match integrity check re-baselines it + assert row.last_modified == original_mtime + assert row.mtime_baseline_utc is False + + def test_accept_skips_unflagged_files(self, authenticated_client, app, db): + with app.app_context(): + row = seed(db, '/m/clean.mkv') + + response = authenticated_client.post('/api/bitrot/accept', + json={'file_ids': [row.id]}) + assert response.status_code == 200 + data = response.get_json() + assert data['accepted'] == 0 + assert data['skipped'] == [row.id] + + def test_accept_rejects_bad_ids(self, authenticated_client, app, db): + with app.app_context(): + response = authenticated_client.post('/api/bitrot/accept', + json={'file_ids': ['abc']}) + assert response.status_code == 400 diff --git a/tests/integration/test_schedule_budget.py b/tests/integration/test_schedule_budget.py new file mode 100644 index 0000000..31e85a2 --- /dev/null +++ b/tests/integration/test_schedule_budget.py @@ -0,0 +1,104 @@ +"""Integration tests for per-schedule integrity time budgets (time_budget_minutes).""" + +from pixelprobe.models import ScanSchedule + + +class TestScheduleBudgetValidation: + + def test_create_file_changes_schedule_with_budget(self, authenticated_client, app, db): + with app.app_context(): + response = authenticated_client.post('/api/schedules', json={ + 'name': 'Nightly Integrity', + 'cron_expression': '0 2 * * *', + 'scan_type': 'file_changes', + 'time_budget_minutes': 120 + }) + assert response.status_code == 201 + assert response.get_json()['time_budget_minutes'] == 120 + + schedule = ScanSchedule.query.filter_by(name='Nightly Integrity').first() + assert schedule.time_budget_minutes == 120 + + def test_budget_rejected_on_non_file_changes_schedule(self, authenticated_client, app, db): + with app.app_context(): + response = authenticated_client.post('/api/schedules', json={ + 'name': 'Budgeted Normal Scan', + 'cron_expression': '0 2 * * *', + 'scan_type': 'normal', + 'time_budget_minutes': 120 + }) + assert response.status_code == 400 + assert 'file_changes' in response.get_json()['error'] + + def test_budget_must_be_positive_integer(self, authenticated_client, app, db): + with app.app_context(): + for bad in (0, -5, 'sixty', 2.5, True): + response = authenticated_client.post('/api/schedules', json={ + 'name': f'Bad Budget {bad}', + 'cron_expression': '0 2 * * *', + 'scan_type': 'file_changes', + 'time_budget_minutes': bad + }) + assert response.status_code == 400, f'accepted invalid budget {bad!r}' + + def test_null_budget_means_unlimited(self, authenticated_client, app, db): + with app.app_context(): + response = authenticated_client.post('/api/schedules', json={ + 'name': 'Unlimited Integrity', + 'cron_expression': '0 2 * * *', + 'scan_type': 'file_changes', + 'time_budget_minutes': None + }) + assert response.status_code == 201 + assert response.get_json()['time_budget_minutes'] is None + + def test_update_sets_and_clears_budget(self, authenticated_client, app, db): + with app.app_context(): + schedule = ScanSchedule( + name='Editable Integrity', + cron_expression='0 3 * * *', + scan_type='file_changes', + is_active=True + ) + db.session.add(schedule) + db.session.commit() + schedule_id = schedule.id + + response = authenticated_client.put(f'/api/schedules/{schedule_id}', + json={'time_budget_minutes': 45}) + assert response.status_code == 200 + assert response.get_json()['time_budget_minutes'] == 45 + + response = authenticated_client.put(f'/api/schedules/{schedule_id}', + json={'time_budget_minutes': None}) + assert response.status_code == 200 + assert response.get_json()['time_budget_minutes'] is None + + def test_changing_type_away_from_file_changes_clears_budget(self, authenticated_client, app, db): + with app.app_context(): + schedule = ScanSchedule( + name='Type Change', + cron_expression='0 3 * * *', + scan_type='file_changes', + time_budget_minutes=90, + is_active=True + ) + db.session.add(schedule) + db.session.commit() + schedule_id = schedule.id + + response = authenticated_client.put(f'/api/schedules/{schedule_id}', + json={'scan_type': 'normal'}) + assert response.status_code == 200 + assert response.get_json()['time_budget_minutes'] is None + + +class TestManualRunBudgetValidation: + + def test_invalid_manual_budget_rejected(self, authenticated_client, app, db): + with app.app_context(): + for bad in (0, -10, 'fast', True): + response = authenticated_client.post('/api/file-changes', + json={'time_budget_minutes': bad}) + assert response.status_code == 400, f'accepted invalid budget {bad!r}' + assert 'time_budget_minutes' in response.get_json()['error'] diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 107c07d..c8d402d 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -1,4 +1,5 @@ from datetime import datetime, timedelta, timezone +from unittest.mock import patch import pytest @@ -225,6 +226,63 @@ def test_update_schedules_removes_and_reloads(self, scheduler, app, db): # at least verify the method runs without error +class TestScheduleDbSync: + """The db-sync job is how schedule CRUD reaches the scheduler process. + + The Celery reload task always executes in a prefork pool child where the + scheduler is not running, so it is skipped; without the sync job a created + or edited schedule only registered after a container restart. + """ + + @pytest.fixture + def scheduler(self, app, db): + scheduler = MediaScheduler() + scheduler.init_app(app) + yield scheduler + scheduler.shutdown() + + def test_sync_job_registered(self, scheduler): + job = scheduler.scheduler.get_job('db_schedule_sync') + assert job is not None + # update_schedules removes every job whose id starts with 'schedule_'; + # the sync job must never match that prefix or it would delete itself + assert not job.id.startswith('schedule_') + + def test_sync_reloads_when_definitions_change(self, scheduler, app, db): + with app.app_context(): + schedule = ScanSchedule(name='Sync Test', cron_expression='0 2 * * *', + scan_type='file_changes', time_budget_minutes=10, + is_active=True) + db.session.add(schedule) + db.session.commit() + + with patch.object(scheduler, 'update_schedules') as update: + scheduler._sync_schedules_from_db() + update.assert_called_once() + + def test_sync_skips_when_unchanged(self, scheduler, app, db): + with app.app_context(): + scheduler._schedules_fp = scheduler._schedule_fingerprint() + with patch.object(scheduler, 'update_schedules') as update: + scheduler._sync_schedules_from_db() + update.assert_not_called() + + def test_budget_change_triggers_reload(self, scheduler, app, db): + with app.app_context(): + schedule = ScanSchedule(name='Budget Sync', cron_expression='0 2 * * *', + scan_type='file_changes', time_budget_minutes=10, + is_active=True) + db.session.add(schedule) + db.session.commit() + scheduler._schedules_fp = scheduler._schedule_fingerprint() + schedule.time_budget_minutes = 30 + db.session.commit() + + with patch.object(scheduler, 'update_schedules') as update: + scheduler._sync_schedules_from_db() + update.assert_called_once() + + class TestQueueConflictRetry: """Test that skipped scheduled scans get a one-shot retry queued.""" diff --git a/tests/unit/test_bitrot_classification.py b/tests/unit/test_bitrot_classification.py new file mode 100644 index 0000000..df3af1d --- /dev/null +++ b/tests/unit/test_bitrot_classification.py @@ -0,0 +1,414 @@ +"""Unit tests for bitrot classification. + +classify_file_change is the pure classification matrix; MaintenanceService +_apply_change_classification is the Phase 3 state machine (flagging, +auto-expire, self-heal); dispatch_event is the notification rule evaluator. +""" + +import json +from datetime import datetime, timedelta, timezone +from unittest.mock import MagicMock, patch + +from pixelprobe.utils.integrity import ( + classify_file_change, + apply_scan_baseline, + adopt_bitrot_baseline, + MTIME_EPSILON_SECONDS, +) +from pixelprobe.models import ScanResult, NotificationProvider, NotificationRule +from pixelprobe.services import notification_service as ns +from pixelprobe.services import maintenance_service as ms +from pixelprobe.services.maintenance_service import ( + MaintenanceService, + BITROT_STABLE_CHECKS_TO_EXPIRE, +) + +NOW = datetime(2026, 7, 1, 12, 0, 0, tzinfo=timezone.utc) +HASH_A = 'a' * 64 +HASH_B = 'b' * 64 +HASH_C = 'c' * 64 + + +class TestClassifyFileChange: + """Every row of the classification matrix.""" + + def test_hash_match_is_unchanged(self): + assert classify_file_change(HASH_A, HASH_A, NOW.isoformat(), NOW) == ('unchanged', False) + + def test_missing_stored_hash_is_no_hash(self): + assert classify_file_change(None, HASH_A, None, NOW) == ('no_hash', True) + + def test_mismatch_with_changed_mtime_is_modified(self): + stored = (NOW - timedelta(hours=5)).isoformat() + assert classify_file_change(HASH_A, HASH_B, stored, NOW, + mtime_trusted=True) == ('modified', True) + + def test_mismatch_with_unchanged_mtime_is_bitrot(self): + assert classify_file_change(HASH_A, HASH_B, NOW.isoformat(), NOW, + mtime_trusted=True) == ('bitrot_suspected', True) + + def test_untrusted_baseline_never_classifies_bitrot(self): + # Pre-upgrade naive-local baselines fall back to 'modified', which + # re-baselines in UTC. Identical mtimes must NOT flag. + assert classify_file_change(HASH_A, HASH_B, NOW.isoformat(), NOW, + mtime_trusted=False) == ('modified', True) + + def test_unparseable_stored_mtime_is_modified(self): + assert classify_file_change(HASH_A, HASH_B, 'not-a-date', NOW, + mtime_trusted=True) == ('modified', True) + + def test_epsilon_boundary_inside_is_bitrot(self): + stored = (NOW - timedelta(seconds=MTIME_EPSILON_SECONDS)).isoformat() + assert classify_file_change(HASH_A, HASH_B, stored, NOW, + mtime_trusted=True) == ('bitrot_suspected', True) + + def test_epsilon_boundary_outside_is_modified(self): + stored = (NOW - timedelta(seconds=MTIME_EPSILON_SECONDS + 1)).isoformat() + assert classify_file_change(HASH_A, HASH_B, stored, NOW, + mtime_trusted=True) == ('modified', True) + + def test_naive_stored_iso_treated_as_utc(self): + naive_iso = NOW.replace(tzinfo=None).isoformat() + assert classify_file_change(HASH_A, HASH_B, naive_iso, NOW, + mtime_trusted=True) == ('bitrot_suspected', True) + + +class TestClassifyAlreadyFlagged: + """The auto-expire state machine inputs.""" + + def test_current_matches_candidate_is_stable(self): + result = classify_file_change(HASH_A, HASH_B, NOW.isoformat(), NOW, + bitrot_suspected=True, bitrot_candidate_hash=HASH_B) + assert result == ('bitrot_stable', True) + + def test_current_matches_original_baseline_is_self_healed(self): + result = classify_file_change(HASH_A, HASH_A, NOW.isoformat(), NOW, + bitrot_suspected=True, bitrot_candidate_hash=HASH_B) + assert result == ('bitrot_self_healed', True) + + def test_third_hash_is_active_rot(self): + result = classify_file_change(HASH_A, HASH_C, NOW.isoformat(), NOW, + bitrot_suspected=True, bitrot_candidate_hash=HASH_B) + assert result == ('bitrot_active', True) + + +def seed_row(db, path='/m/f.mkv', **kwargs): + defaults = dict( + file_path=path, + file_size=1000, + file_type='video/mp4', + scan_status='completed', + file_hash=HASH_A, + is_corrupted=False, + ) + defaults.update(kwargs) + row = ScanResult(**defaults) + db.session.add(row) + db.session.commit() + return row + + +def change_info(change_type, **kwargs): + info = { + 'file_id': 1, + 'file_path': '/m/f.mkv', + 'change_type': change_type, + 'stored_hash': HASH_A, + 'current_hash': HASH_B, + 'stored_modified': NOW.isoformat(), + 'current_modified': NOW.isoformat(), + 'stored_size': 1000, + 'current_size': 1000, + } + info.update(kwargs) + return info + + +class TestApplyChangeClassification: + + def _service(self): + service = MaintenanceService(':memory:') + return service + + def test_new_detection_flags_file(self, app, db): + row = seed_row(db) + service = self._service() + category = service._apply_change_classification(row, change_info('bitrot_suspected')) + + assert category == 'bitrot_new' + assert row.bitrot_suspected is True + assert row.bitrot_detected_date is not None + assert row.bitrot_candidate_hash == HASH_B + assert row.bitrot_stable_checks == 0 + assert row.scan_status == 'pending' + details = json.loads(row.bitrot_details) + assert details['stored_hash'] == HASH_A + assert details['current_hash'] == HASH_B + + def test_first_detection_date_is_permanent(self, app, db): + detected = NOW.replace(tzinfo=None) + row = seed_row(db, bitrot_detected_date=detected) + service = self._service() + service._apply_change_classification(row, change_info('bitrot_suspected')) + + assert row.bitrot_detected_date == detected # re-detection keeps first date + + def test_rescan_does_not_clear_flag_or_adopt_hash(self, app, db): + # The anti-laundering guard lives in apply_scan_baseline; here we assert + # the Phase 3 detection leaves the stored baseline untouched. + row = seed_row(db) + service = self._service() + service._apply_change_classification(row, change_info('bitrot_suspected')) + + assert row.file_hash == HASH_A # baseline NOT overwritten + assert row.bitrot_suspected is True + + def test_active_rot_resets_counter(self, app, db): + row = seed_row(db, bitrot_suspected=True, bitrot_candidate_hash=HASH_B, + bitrot_stable_checks=1) + service = self._service() + category = service._apply_change_classification( + row, change_info('bitrot_active', current_hash=HASH_C)) + + assert category == 'bitrot_active' + assert row.bitrot_candidate_hash == HASH_C + assert row.bitrot_stable_checks == 0 + + def test_stale_bitrot_result_on_unflagged_row_is_ignored(self, app, db): + # The flag can be cleared (manual accept) between task dispatch and + # Phase 3 apply; the queued bitrot result must not resurrect state. + row = seed_row(db, bitrot_suspected=False, bitrot_candidate_hash=None, + bitrot_stable_checks=0) + service = self._service() + for stale_type in ('bitrot_stable', 'bitrot_self_healed', 'bitrot_active'): + category = service._apply_change_classification( + row, change_info(stale_type, current_hash=HASH_C)) + assert category == 'other' + assert row.bitrot_suspected is False + assert row.bitrot_candidate_hash is None + assert row.file_hash == HASH_A + + def test_stable_check_increments_without_expiry_below_threshold(self, app, db): + row = seed_row(db, bitrot_suspected=True, bitrot_candidate_hash=HASH_B, + bitrot_stable_checks=0) + service = self._service() + category = service._apply_change_classification( + row, change_info('bitrot_stable', current_hash=HASH_B)) + + assert category == 'bitrot_stable' + assert row.bitrot_stable_checks == 1 + assert row.bitrot_suspected is True + assert row.file_hash == HASH_A + + def test_auto_expire_adopts_candidate_and_keeps_record(self, app, db): + detected = NOW.replace(tzinfo=None) + row = seed_row(db, bitrot_suspected=True, bitrot_candidate_hash=HASH_B, + bitrot_stable_checks=BITROT_STABLE_CHECKS_TO_EXPIRE - 1, + bitrot_detected_date=detected, bitrot_details='{"x": 1}', + scan_status='completed', is_corrupted=False) + service = self._service() + category = service._apply_change_classification( + row, change_info('bitrot_stable', current_hash=HASH_B)) + + assert category == 'bitrot_expired' + assert row.bitrot_suspected is False + assert row.file_hash == HASH_B # candidate adopted as baseline + assert row.mtime_baseline_utc is True + assert row.bitrot_candidate_hash is None + assert row.bitrot_stable_checks == 0 + # Detection record is permanent + assert row.bitrot_detected_date == detected + assert row.bitrot_details == '{"x": 1}' + + def test_no_expiry_while_rescan_dirty_or_pending(self, app, db): + for status, corrupted in (('pending', False), ('completed', True)): + row = seed_row(db, path=f'/m/{status}_{corrupted}.mkv', + bitrot_suspected=True, bitrot_candidate_hash=HASH_B, + bitrot_stable_checks=BITROT_STABLE_CHECKS_TO_EXPIRE - 1, + scan_status=status, is_corrupted=corrupted) + service = self._service() + category = service._apply_change_classification( + row, change_info('bitrot_stable', current_hash=HASH_B)) + + assert category == 'bitrot_stable' + assert row.bitrot_suspected is True + assert row.file_hash == HASH_A + + def test_self_healed_clears_flag_keeps_record(self, app, db): + detected = NOW.replace(tzinfo=None) + row = seed_row(db, bitrot_suspected=True, bitrot_candidate_hash=HASH_B, + bitrot_stable_checks=1, bitrot_detected_date=detected) + service = self._service() + category = service._apply_change_classification( + row, change_info('bitrot_self_healed', current_hash=HASH_A)) + + assert category == 'bitrot_self_healed' + assert row.bitrot_suspected is False + assert row.bitrot_candidate_hash is None + assert row.file_hash == HASH_A + assert row.bitrot_detected_date == detected + + def test_deleted_records_absence_without_row_deletion(self, app, db): + row = seed_row(db) + service = self._service() + category = service._apply_change_classification( + row, change_info('deleted', current_hash=None)) + + assert category == 'deleted' + assert row.file_exists is False + # Row deletion is exclusively orphan cleanup's job + assert db.session.get(ScanResult, row.id) is not None + # A gone file is not queued for rescan + assert row.scan_status == 'completed' + + def test_modified_marks_pending(self, app, db): + row = seed_row(db) + service = self._service() + category = service._apply_change_classification(row, change_info('modified')) + + assert category == 'modified' + assert row.scan_status == 'pending' + assert row.bitrot_suspected is False + + +class TestDispatchEvent: + + def _provider_and_rule(self, db, provider_active=True, rule_active=True, + event_type='bitrot_suspected'): + provider = NotificationProvider( + name='test-webhook', + provider_type='webhook', + is_active=provider_active, + configuration={'webhook_url': 'https://example.com/hook'}, + ) + db.session.add(provider) + db.session.commit() + rule = NotificationRule( + provider_id=provider.id, + event_type=event_type, + is_active=rule_active, + priority='high', + ) + db.session.add(rule) + db.session.commit() + return provider, rule + + def test_dispatches_to_matching_active_rule(self, app, db): + provider, _ = self._provider_and_rule(db) + + with patch.object(ns.NotificationService, 'send_notification', + return_value=(True, None)) as send: + sent = ns.dispatch_event('bitrot_suspected', 'title', 'message') + + assert sent == 1 + send.assert_called_once() + assert send.call_args.kwargs['priority'] == 'high' + assert provider.last_notification_status == 'success' + + def test_default_rule_priority_defers_to_event_priority(self, app, db): + # NotificationRule.priority defaults to 'normal' (NOT NULL); a rule + # left at the default must not downgrade a high-priority event. + self._provider_and_rule(db) + rule = NotificationRule.query.first() + rule.priority = 'normal' + db.session.commit() + + with patch.object(ns.NotificationService, 'send_notification', + return_value=(True, None)) as send: + ns.dispatch_event('bitrot_suspected', 't', 'm', priority='high') + + assert send.call_args.kwargs['priority'] == 'high' + + def test_skips_inactive_rule_and_provider(self, app, db): + self._provider_and_rule(db, rule_active=False) + + with patch.object(ns.NotificationService, 'send_notification') as send: + assert ns.dispatch_event('bitrot_suspected', 't', 'm') == 0 + send.assert_not_called() + + def test_skips_non_matching_event_type(self, app, db): + self._provider_and_rule(db, event_type='scan_complete') + + with patch.object(ns.NotificationService, 'send_notification') as send: + assert ns.dispatch_event('bitrot_suspected', 't', 'm') == 0 + send.assert_not_called() + + def test_provider_failure_recorded_not_raised(self, app, db): + provider, _ = self._provider_and_rule(db) + + with patch.object(ns.NotificationService, 'send_notification', + return_value=(False, 'boom')): + assert ns.dispatch_event('bitrot_suspected', 't', 'm') == 0 + assert provider.last_notification_status == 'failure' + + +class TestBaselineHelpers: + + def test_apply_scan_baseline_writes_and_trusts(self, app, db): + row = seed_row(db) + mtime = NOW.replace(tzinfo=None) + + assert apply_scan_baseline(row, HASH_B, mtime) is True + assert row.file_hash == HASH_B + assert row.last_modified == mtime + assert row.mtime_baseline_utc is True + + def test_apply_scan_baseline_refuses_flagged_rows(self, app, db): + row = seed_row(db, bitrot_suspected=True, last_modified=NOW.replace(tzinfo=None)) + + assert apply_scan_baseline(row, HASH_B, NOW.replace(tzinfo=None) + timedelta(days=1)) is False + assert row.file_hash == HASH_A + assert row.last_modified == NOW.replace(tzinfo=None) + assert row.mtime_baseline_utc is False + + def test_apply_scan_baseline_without_mtime_keeps_trust_state(self, app, db): + # Error-path scans carry last_modified=None; the stored mtime and its + # trust flag must survive so classification is not poisoned. + mtime = NOW.replace(tzinfo=None) + row = seed_row(db, last_modified=mtime, mtime_baseline_utc=True) + + assert apply_scan_baseline(row, HASH_B, None) is True + assert row.file_hash == HASH_B + assert row.last_modified == mtime + assert row.mtime_baseline_utc is True + + def test_adopt_bitrot_baseline_clears_flag_keeps_record(self, app, db): + detected = NOW.replace(tzinfo=None) + mtime = detected + timedelta(days=1) + row = seed_row(db, bitrot_suspected=True, bitrot_candidate_hash=HASH_B, + bitrot_stable_checks=2, bitrot_detected_date=detected, + bitrot_details='{"x": 1}') + + adopt_bitrot_baseline(row, mtime) + + assert row.file_hash == HASH_B + assert row.last_modified == mtime + assert row.mtime_baseline_utc is True + assert row.bitrot_suspected is False + assert row.bitrot_candidate_hash is None + assert row.bitrot_stable_checks == 0 + assert row.bitrot_detected_date == detected + assert row.bitrot_details == '{"x": 1}' + + +class TestBitrotSummaryNotification: + + def test_summary_sends_single_aggregated_event(self, app, db): + service = MaintenanceService(':memory:') + paths = [f'/m/f{i}.mkv' for i in range(15)] + + with patch.object(ms, 'dispatch_event') as dispatch: + service._notify_bitrot_summary(paths) + + dispatch.assert_called_once() + args, kwargs = dispatch.call_args + assert args[0] == 'bitrot_suspected' + assert '15' in args[1] + assert kwargs['priority'] == 'high' + assert kwargs['additional_data']['count'] == 15 + + def test_summary_with_no_flagged_files_sends_nothing(self, app, db): + service = MaintenanceService(':memory:') + with patch.object(ms, 'dispatch_event') as dispatch: + service._notify_bitrot_summary([]) + dispatch.assert_not_called() diff --git a/tests/unit/test_maintenance_service.py b/tests/unit/test_maintenance_service.py index bc23c1a..dfdf59b 100644 --- a/tests/unit/test_maintenance_service.py +++ b/tests/unit/test_maintenance_service.py @@ -41,10 +41,12 @@ class TestActiveTaskShape: remain in the active set forever. """ - REQUIRED_FIELDS = {'task', 'size', 'path', 'submitted_at'} + REQUIRED_FIELDS = {'task', 'id', 'size', 'path', 'submitted_at'} def test_required_fields_contract(self): - assert self.REQUIRED_FIELDS == {'task', 'size', 'path', 'submitted_at'} + # 'id' feeds stamp_integrity_check on error/abandon terminal outcomes + # (rolling queue); 'submitted_at' feeds the abandon-on-timeout branch. + assert self.REQUIRED_FIELDS == {'task', 'id', 'size', 'path', 'submitted_at'} class TestAgeComparison: diff --git a/tests/unit/test_rolling_integrity_queue.py b/tests/unit/test_rolling_integrity_queue.py new file mode 100644 index 0000000..af7cba8 --- /dev/null +++ b/tests/unit/test_rolling_integrity_queue.py @@ -0,0 +1,138 @@ +"""Unit tests for the rolling integrity queue. + +fetch_integrity_batch is the ordered, watermark-bounded queue fetch that +replaced the load-everything query in _run_file_changes_check: stalest files +first, rows stamped during the current run excluded, in-flight rows excluded +explicitly. +""" + +from datetime import datetime, timedelta + +from pixelprobe.models import ScanResult +from pixelprobe.services.maintenance_service import fetch_integrity_batch + +# Naive UTC datetimes, matching what the DateTime column stores. +NOW = datetime(2026, 7, 1, 12, 0, 0) +WATERMARK = NOW + timedelta(hours=1) + + +def seed(db, path, checked=None, size=1000): + row = ScanResult( + file_path=path, + file_size=size, + file_type='video/mp4', + scan_status='completed', + file_hash='cafe' * 16, + last_integrity_check_date=checked, + ) + db.session.add(row) + db.session.commit() + return row.id + + +class TestQueueOrdering: + + def test_never_checked_files_come_first(self, db): + checked = seed(db, '/m/checked.mkv', checked=NOW - timedelta(days=2)) + never = seed(db, '/m/never.mkv', checked=None) + + batch = fetch_integrity_batch(None, WATERMARK, set(), 10) + + assert [b['id'] for b in batch] == [never, checked] + + def test_stalest_first_with_id_tiebreak(self, db): + old = seed(db, '/m/old.mkv', checked=NOW - timedelta(days=30)) + older = seed(db, '/m/older.mkv', checked=NOW - timedelta(days=60)) + tie_a = seed(db, '/m/tie_a.mkv', checked=NOW - timedelta(days=10)) + tie_b = seed(db, '/m/tie_b.mkv', checked=NOW - timedelta(days=10)) + + batch = fetch_integrity_batch(None, WATERMARK, set(), 10) + + assert [b['id'] for b in batch] == [older, old, tie_a, tie_b] + + def test_smallest_files_first_within_batch(self, db): + # Staleness decides WHICH rows are in the batch; size decides + # dispatch order within it (small files backfill the wide slots). + big = seed(db, '/m/big.mkv', checked=NOW - timedelta(days=9), size=5000) + small = seed(db, '/m/small.mkv', checked=NOW - timedelta(days=1), size=10) + + batch = fetch_integrity_batch(None, WATERMARK, set(), 10) + + assert [b['id'] for b in batch] == [small, big] + + def test_budgeted_runs_dispatch_largest_first(self, db): + # Under a time budget a huge file must start hashing early in the + # window, not at the deadline. + big = seed(db, '/m/big.mkv', checked=None, size=5000) + small = seed(db, '/m/small.mkv', checked=None, size=10) + + batch = fetch_integrity_batch(None, WATERMARK, set(), 10, largest_first=True) + + assert [b['id'] for b in batch] == [big, small] + + +class TestQueueBounds: + + def test_watermark_excludes_rows_stamped_this_run(self, db): + done = seed(db, '/m/done.mkv', checked=WATERMARK + timedelta(seconds=5)) + todo = seed(db, '/m/todo.mkv', checked=NOW) + + ids = [b['id'] for b in fetch_integrity_batch(None, WATERMARK, set(), 10)] + + assert todo in ids + assert done not in ids + + def test_excluded_ids_are_skipped(self, db): + in_flight = seed(db, '/m/inflight.mkv', checked=None) + fresh = seed(db, '/m/fresh.mkv', checked=None) + + ids = [b['id'] for b in fetch_integrity_batch(None, WATERMARK, {in_flight}, 10)] + + assert ids == [fresh] + + def test_batch_size_limits_fetch(self, db): + for i in range(5): + seed(db, f'/m/f{i}.mkv', checked=NOW - timedelta(days=5 - i)) + + batch = fetch_integrity_batch(None, WATERMARK, set(), 2) + + assert len(batch) == 2 + + def test_file_paths_scopes_the_queue(self, db): + seed(db, '/m/other.mkv', checked=None) + wanted = seed(db, '/m/wanted.mkv', checked=None) + + ids = [b['id'] for b in fetch_integrity_batch(['/m/wanted.mkv'], WATERMARK, set(), 10)] + + assert ids == [wanted] + + +class TestRollingSweep: + + def test_stamped_batches_are_disjoint_and_cover_everything(self, db): + # Two budgeted runs must process disjoint slices; a full sweep is the + # union of successive batches as processed files get stamped. + all_ids = {seed(db, f'/m/s{i}.mkv', checked=NOW - timedelta(days=i)) for i in range(10)} + + processed = [] + for _ in range(4): + batch = fetch_integrity_batch(None, WATERMARK, set(), 3) + for entry in batch: + row = db.session.get(ScanResult, entry['id']) + row.last_integrity_check_date = WATERMARK + timedelta(seconds=1) + db.session.commit() + processed.append({entry['id'] for entry in batch}) + + for i in range(len(processed)): + for j in range(i + 1, len(processed)): + assert not (processed[i] & processed[j]), 'batches overlap' + assert set().union(*processed) == all_ids + assert fetch_integrity_batch(None, WATERMARK, set(), 3) == [] + + def test_dict_shape_matches_dispatch_contract(self, db): + seed(db, '/m/shape.mkv', checked=None, size=123) + + batch = fetch_integrity_batch(None, WATERMARK, set(), 1) + + assert set(batch[0]) == {'id', 'file_path', 'file_hash', 'file_size', 'last_modified', + 'mtime_baseline_utc', 'bitrot_suspected', 'bitrot_candidate_hash'}