From c8d66482138dbd8337002458372dbf083ca29dfd Mon Sep 17 00:00:00 2001 From: Adam Hernandez Date: Fri, 18 Sep 2026 13:12:13 -0700 Subject: [PATCH 1/4] fix(import): reduce SQLite contention during recovery --- docs/development/IMPORT_PERFORMANCE.md | 17 +- .../import_deferred_recovery_execution.py | 6 +- src/pullbox/services/import_file_execution.py | 73 ++++- .../services/import_recovery_checkpoint.py | 22 ++ src/pullbox/services/import_workflow_state.py | 4 + src/pullbox/tasks/import_task.py | 24 +- ...test_import_deferred_recovery_execution.py | 15 +- tests/unit/test_import_recovery_contention.py | 277 ++++++++++++++++++ tests/unit/test_import_task.py | 23 +- 9 files changed, 436 insertions(+), 25 deletions(-) create mode 100644 src/pullbox/services/import_recovery_checkpoint.py create mode 100644 tests/unit/test_import_recovery_contention.py diff --git a/docs/development/IMPORT_PERFORMANCE.md b/docs/development/IMPORT_PERFORMANCE.md index 7de632bb..82f56538 100644 --- a/docs/development/IMPORT_PERFORMANCE.md +++ b/docs/development/IMPORT_PERFORMANCE.md @@ -18,7 +18,11 @@ therefore do not hold SQLite's writer lock, and results do not accumulate in an unbounded in-memory collection. Deferred catalog recovery publishes live provider progress without rewriting the full recovery snapshot, then writes one JSON-safe checkpoint after each -completed catalog. +completed catalog. Once preparation has durably stored its decisions on file +rows, consumed catalog and title-lookup caches are removed from the progress +snapshot. Recovery scope, summary counts, and restart state remain. Older +prepared/completed snapshots are compacted on their next progress checkpoint; +unfinished catalog preparation retains its caches for resume. `PULLBOX_IMPORT_SCAN_WORKER_COUNT=0` selects automatic inspection concurrency, up to four workers. CPU affinity, cgroup v2 CPU quotas and parent limits, @@ -34,13 +38,22 @@ Automatic mode deliberately does not saturate high-core machines: local ZIP header parsing did not improve with more than 2-4 workers. Docker Desktop's VM resources are the relevant limits, not the Mac's advertised RAM. -Step 4 keeps `PULLBOX_IMPORT_FILE_WORKER_COUNT=2` and its existing temporary-space +Managed Step 4 keeps `PULLBOX_IMPORT_FILE_WORKER_COUNT=2` and its existing temporary-space preflight, target-collision serialization, per-worker sessions, and rollback journal. It now bounds submitted tasks as well as active workers, rather than creating one waiting task per file. Exiting or canceling either worker pool drains active work before the job can transition; no orphan filesystem work may continue after cancellation is reported complete. +SQLite in-place registration uses one file worker, with an isolated transaction +per file. It retries only transient SQLite lock failures, rolling back and +rechecking control requests and source identity in a fresh session. Exhausted +retries leave the import resumable as database-busy, not a failed match. Source +archives, skips, safety blocks, and confirmed targets are not loosened. Progress +inside the file transaction is live-only; the terminal file checkpoint is +persisted after commit, avoiding a competing progress writer. Archive scanning +and managed copy/conversion keep their existing bounded parallelism. + ## Progress and Evidence Unknown inventory totals are indeterminate. Completed series report 100% for diff --git a/src/pullbox/services/import_deferred_recovery_execution.py b/src/pullbox/services/import_deferred_recovery_execution.py index e14761c6..d338cc30 100644 --- a/src/pullbox/services/import_deferred_recovery_execution.py +++ b/src/pullbox/services/import_deferred_recovery_execution.py @@ -36,6 +36,7 @@ provider_ids, refresh_recovered_groups, ) +from pullbox.services.import_recovery_checkpoint import compact_recovery_state from pullbox.services.import_reference_recovery import ( reference_candidates, repair_catalog_references, @@ -61,7 +62,10 @@ def recovery_state(job: ImportJob) -> dict[str, Any]: def save_recovery_state(job: ImportJob, state: dict[str, Any]) -> None: - job.progress_snapshot = {**dict(job.progress_snapshot or {}), "deferred_recovery": state} + job.progress_snapshot = { + **dict(job.progress_snapshot or {}), + "deferred_recovery": compact_recovery_state(state), + } def _catalog_summary_payload(summary: IssueSummary) -> dict[str, Any]: diff --git a/src/pullbox/services/import_file_execution.py b/src/pullbox/services/import_file_execution.py index 64bb966c..0148ad16 100644 --- a/src/pullbox/services/import_file_execution.py +++ b/src/pullbox/services/import_file_execution.py @@ -14,6 +14,7 @@ import structlog from sqlalchemy import func as sa_func from sqlalchemy import select as sa_select +from sqlalchemy.exc import OperationalError from sqlalchemy.orm import joinedload from pullbox.core.exceptions import ( @@ -35,6 +36,11 @@ build_managed_placement_signature, validate_file_identity_signature, ) +from pullbox.core.sqlite_lock import ( + SQLITE_LOCK_RETRY_ATTEMPTS, + is_sqlite_locked_error, + sqlite_lock_retry_delay, +) from pullbox.models.import_job import ( ImportedFile, ImportedFileStatus, @@ -817,6 +823,9 @@ async def process_import_series_files( item_id = item.id job_id = job.id in_place = job.file_handling_mode == ImportFileHandlingMode.IN_PLACE + sqlite_reference_processing = ( + in_place and session.bind is not None and session.bind.dialect.name == "sqlite" + ) move_to_library = not in_place transfer_method = ( "leave_in_place" if in_place else job.effective_transfer_method or job.transfer_method @@ -847,17 +856,26 @@ async def process_import_series_files( ) effective_worker_count = max(int(file_worker_count or 1), 1) + if sqlite_reference_processing: + # Referencing files is write-heavy, not transfer-heavy. Competing SQLite + # writers add lock waits rather than throughput; isolate each file retry. + effective_worker_count = 1 parallel_processing_available = ( _file_ids_override is None and session_factory is not None - and effective_worker_count > 1 - and len(importable_file_ids) > 1 + and ( + sqlite_reference_processing + or (effective_worker_count > 1 and len(importable_file_ids) > 1) + ) ) - if parallel_processing_available and not await _requires_serial_file_processing( - session, - job, - resolved_series_id=resolved_series_id, - importable_files=importable_files, + if parallel_processing_available and ( + sqlite_reference_processing + or not await _requires_serial_file_processing( + session, + job, + resolved_series_id=resolved_series_id, + importable_files=importable_files, + ) ): assert session_factory is not None await session.commit() @@ -902,7 +920,9 @@ async def locked_record_action( await session.flush() return action - async def process_one_file(imp_file_id: int) -> tuple[int, int]: + await session.commit() + + async def process_one_file_attempt(imp_file_id: int) -> tuple[int, int]: async with session_factory() as worker_session: worker_job = await worker_session.get(type(job), job_id) worker_item = await worker_session.get(type(item), item_id) @@ -939,6 +959,30 @@ async def process_one_file(imp_file_id: int) -> tuple[int, int]: _setup_placeholder_targets=False, ) + async def process_one_file(imp_file_id: int) -> tuple[int, int]: + for attempt in range(1, SQLITE_LOCK_RETRY_ATTEMPTS + 1): + try: + return await process_one_file_attempt(imp_file_id) + except OperationalError as exc: + if not sqlite_reference_processing or not is_sqlite_locked_error(exc): + raise + if attempt == SQLITE_LOCK_RETRY_ATTEMPTS: + raise JobPausedError( + "Import paused because the database is busy. Confirmed matches " + "are preserved; resume the import when database activity settles." + ) from exc + logger.warning( + "import_reference_retrying_after_sqlite_lock", + job_id=job_id, + imported_file_id=imp_file_id, + attempt=attempt, + max_attempts=SQLITE_LOCK_RETRY_ATTEMPTS, + ) + # The attempt session is closed before waiting. A fresh + # transaction rechecks control requests and source identity. + await asyncio.sleep(sqlite_lock_retry_delay(attempt)) + raise RuntimeError("Import reference retry loop ended unexpectedly") + files_imported = 0 files_failed = 0 async with bounded_async_map( @@ -1112,7 +1156,7 @@ async def report_current_file( return report_current_file - force_live_file_progress = placeholder_progress_live_only + force_live_file_progress = placeholder_progress_live_only or sqlite_reference_processing _report_current_file = _build_current_file_reporter( imp_file, file_index, @@ -1547,6 +1591,17 @@ async def report_current_file( await asyncio.to_thread(cleanup_prepared_file, prepared) await session.rollback() placeholder_progress_live_only = False + if ( + sqlite_reference_processing + and isinstance(exc, OperationalError) + and is_sqlite_locked_error(exc) + ): + if _file_ids_override is not None: + raise + raise JobPausedError( + "Import paused because the database is busy. Confirmed matches " + "are preserved; resume the import when database activity settles." + ) from exc destination_preserved_for_review = False try: destination_preserved_for_review = await asyncio.to_thread( diff --git a/src/pullbox/services/import_recovery_checkpoint.py b/src/pullbox/services/import_recovery_checkpoint.py new file mode 100644 index 00000000..50658550 --- /dev/null +++ b/src/pullbox/services/import_recovery_checkpoint.py @@ -0,0 +1,22 @@ +"""Keep resumable recovery context without retaining consumed provider catalogs.""" + +from typing import Any + +_PREPARATION_CACHE_KEYS = frozenset( + { + "candidates", + "completed", + "matches", + "title_candidates", + "title_completed", + "title_matches", + "reference_candidates", + } +) + + +def compact_recovery_state(state: dict[str, Any]) -> dict[str, Any]: + """Discard caches only after their decisions are stored on import-file rows.""" + if state.get("state") not in {"prepared", "completed"}: + return dict(state) + return {key: value for key, value in state.items() if key not in _PREPARATION_CACHE_KEYS} diff --git a/src/pullbox/services/import_workflow_state.py b/src/pullbox/services/import_workflow_state.py index cada4113..c3919b65 100644 --- a/src/pullbox/services/import_workflow_state.py +++ b/src/pullbox/services/import_workflow_state.py @@ -664,6 +664,10 @@ async def persist_progress_snapshot( for key in _PERSISTENT_IMPORT_CONTEXT_KEYS: if key in existing_snapshot: payload[key] = existing_snapshot[key] + if isinstance(payload.get("deferred_recovery"), dict): + from pullbox.services.import_recovery_checkpoint import compact_recovery_state + + payload["deferred_recovery"] = compact_recovery_state(payload["deferred_recovery"]) if int(payload.get("progress_revision") or 0) <= 0: payload["progress_revision"] = next_progress_revision(job) else: diff --git a/src/pullbox/tasks/import_task.py b/src/pullbox/tasks/import_task.py index 2705ab22..bc52d11c 100644 --- a/src/pullbox/tasks/import_task.py +++ b/src/pullbox/tasks/import_task.py @@ -8,6 +8,7 @@ import structlog from sqlalchemy import select as sa_select +from sqlalchemy.exc import OperationalError from pullbox.core.exceptions import JobCancelledError, JobPausedError from pullbox.core.sqlite_lock import ( @@ -778,9 +779,14 @@ async def progress_callback(event: ImportProgressEvent) -> None: else: logger.info("import_runner_noop", job_id=job_id, status=job.status.value) return - except JobPausedError: + except JobPausedError as exc: await session.rollback() - await self._mark_paused(session, job_id) + if isinstance(exc.__cause__, OperationalError) and is_sqlite_locked_error( + exc.__cause__ + ): + await self._mark_stalled(job_id) + else: + await self._mark_paused(session, job_id) except JobCancelledError: await session.rollback() await self._finalize_cancel(session, job_id) @@ -898,11 +904,16 @@ async def progress_callback(event: ImportProgressEvent) -> None: service.schedule_story_arc_sync() if _should_schedule_comicinfo_enrichment(run_import_result): service.schedule_comicinfo_enrichment(session_factory, job_id=job_id) - except JobPausedError: + except JobPausedError as exc: await session.rollback() job = await session.get(ImportJob, job_id) if job is not None and job.status != ImportJobStatus.CANCELLED: - sync_paused_job_state(job) + if isinstance(exc.__cause__, OperationalError) and is_sqlite_locked_error( + exc.__cause__ + ): + sync_stalled_job_state(job) + else: + sync_paused_job_state(job) await session.commit() except JobCancelledError: await session.rollback() @@ -957,7 +968,10 @@ async def progress_callback(event: ImportProgressEvent) -> None: terminal_status = terminal_event_override.status else: job = await session.get(ImportJob, job_id) - if job is not None and job.status == ImportJobStatus.PAUSED: + if job is not None and job.status in { + ImportJobStatus.PAUSED, + ImportJobStatus.STALLED, + }: terminal_status = await _publish_current_snapshot_event_for_job( session, job_id, diff --git a/tests/unit/test_import_deferred_recovery_execution.py b/tests/unit/test_import_deferred_recovery_execution.py index 718072ac..e969e2a8 100644 --- a/tests/unit/test_import_deferred_recovery_execution.py +++ b/tests/unit/test_import_deferred_recovery_execution.py @@ -602,10 +602,19 @@ async def test_catalog_checkpoint_serializes_local_catalog_cutoff(db_session): ) ] - assert await prepare_deferred_recovery(db_session, job.id, metadata_service=provider) + cutoffs = [] + + async def checkpoint(_event): + matches = job.progress_snapshot.get("deferred_recovery", {}).get("matches", {}) + if "7001" in matches: + cutoffs.append(matches["7001"][0]["summary"]["source_cutoff_at"]) + + assert await prepare_deferred_recovery( + db_session, job.id, metadata_service=provider, progress_callback=checkpoint + ) - stored = job.progress_snapshot["deferred_recovery"]["matches"]["7001"][0]["summary"] - assert stored["source_cutoff_at"] == "2026-09-13T05:00:00+00:00" + assert cutoffs and set(cutoffs) == {"2026-09-13T05:00:00+00:00"} + assert "matches" not in job.progress_snapshot["deferred_recovery"] await db_session.commit() diff --git a/tests/unit/test_import_recovery_contention.py b/tests/unit/test_import_recovery_contention.py new file mode 100644 index 00000000..02c22b9a --- /dev/null +++ b/tests/unit/test_import_recovery_contention.py @@ -0,0 +1,277 @@ +"""File-backed SQLite regressions for completed-import recovery.""" + +import asyncio +import json +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from sqlalchemy import event, func, select, update +from sqlalchemy.exc import OperationalError +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from pullbox.core.exceptions import JobCancelledError, JobPausedError +from pullbox.models import Base +from pullbox.models.import_job import ( + ImportedFile, + ImportedFileStatus, + ImportedSeries, + ImportFileHandlingMode, + ImportJob, + ImportJobAction, + ImportJobStatus, + ImportSourceType, +) +from pullbox.models.library import LibraryFile, LibraryRoot +from pullbox.schemas.import_job import ImportProgressEvent +from pullbox.services.import_deferred_recovery_execution import save_recovery_state +from pullbox.services.import_file_execution import process_import_series_files +from pullbox.services.import_job_actions import record_action +from pullbox.services.import_referenced_sources import MYLAR_REFERENCE_ROOT_ID_SIGNATURE_KEY +from pullbox.services.import_workflow_state import persist_progress_snapshot +from tests.unit.test_import_file_execution import ( + _mock_register_library_file, + _setup_full_scenario, +) + + +@pytest.fixture +async def recovery_factory(tmp_path): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'recovery.db'}") + + @event.listens_for(engine.sync_engine, "connect") + def configure(connection, _record): + connection.execute("PRAGMA journal_mode=WAL") + connection.execute("PRAGMA busy_timeout=25") + connection.execute("PRAGMA foreign_keys=ON") + + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + try: + yield async_sessionmaker(engine, expire_on_commit=False) + finally: + await engine.dispose() + + +@pytest.mark.parametrize("source_type", list(ImportSourceType)) +@pytest.mark.parametrize( + "failure", + [None, "once", "actual_lock", "journal_once", "always", "cancel", "second", "not_locked"], +) +async def test_in_place_workers_serialize_retry_and_preserve_decisions( + recovery_factory, monkeypatch, tmp_path, source_type, failure +): + async with recovery_factory() as session: + job, item, files, _, _ = await _setup_full_scenario(session, num_issues=2) + job.source_type = source_type + job.file_handling_mode = ImportFileHandlingMode.IN_PLACE + job.effective_transfer_method = "leave_in_place" + job.update_embedded_comicinfo_from_match = False + job.convert_to_preferred_format = False + root = await session.get(LibraryRoot, job.target_library_root_id) + root.path = str(Path(files[0].file_path).parent) + for file in files: + file.source_signature = { + **file.source_signature, + MYLAR_REFERENCE_ROOT_ID_SIGNATURE_KEY: root.id, + } + paths = [Path(file.file_path) for file in files] + originals = [(path.read_bytes(), path.stat().st_mtime_ns) for path in paths] + job_id, item_id = job.id, item.id + await session.commit() + + active = maximum = attempts = inspections = 0 + progress = [] + register = _mock_register_library_file() + + async def refresh(session, _job, _item, file): + nonlocal inspections + inspections += 1 + # Revalidation refreshes metadata before progress, just like a remount. + file.diagnostics = {**file.diagnostics, "source_rechecked": True} + if failure == "actual_lock" and inspections == 1: + async with recovery_factory() as blocker: + await blocker.execute(update(ImportJob).values(error_message="other writer")) + try: + await session.flush() + finally: + await blocker.rollback() + await session.flush() + + monkeypatch.setattr("pullbox.services.import_file_execution.refresh_recovery_source", refresh) + + async def prepare(_session, _job, file, **_kwargs): + nonlocal active, maximum + active += 1 + maximum = max(maximum, active) + try: + await asyncio.sleep(0.01) + return SimpleNamespace( + registration_source=Path(file.file_path), + original_source=Path(file.file_path), + converted=False, + skip_embedded_comicinfo=True, + ) + finally: + active -= 1 + + async def register_with_contention(*args, **kwargs): + nonlocal attempts + attempts += 1 + if failure == "not_locked": + raise OperationalError("INSERT", {}, Exception("disk I/O error")) + if ( + failure == "always" + or (failure in {"once", "cancel"} and attempts == 1) + or (failure == "second" and attempts > 1) + ): + raise OperationalError("INSERT", {}, Exception("database is locked")) + return await register(*args, **kwargs) + + async def check_control(*_args): + if failure == "cancel" and attempts: + raise JobCancelledError("Cancelled while retrying") + + async def journal(*args, **kwargs): + action = await record_action(*args, **kwargs) + if failure == "journal_once" and attempts == 1: + raise OperationalError("INSERT", {}, Exception("database is locked")) + return action + + async def report(**kwargs): + progress.append(kwargs) + if kwargs["stage"] == "preparing": + assert kwargs.get("live_only"), "Never await a second writer inside a file transaction" + if not kwargs.get("live_only"): + async with recovery_factory() as progress_session: + await progress_session.execute( + update(ImportJob).where(ImportJob.id == job_id).values(error_message=None) + ) + await progress_session.commit() + + async def execute(): + async with recovery_factory() as session: + job = await session.get(ImportJob, job_id) + item = await session.get(ImportedSeries, item_id) + result = await process_import_series_files( + session, + job, + item, + load_media_settings=AsyncMock(return_value={"skip_existing_files": "false"}), + load_trash_dir=AsyncMock(return_value=tmp_path / "trash"), + load_ingest_policy=AsyncMock(return_value=object()), + load_permission_policy=AsyncMock(return_value=object()), + raise_if_cancelled=check_control, + prepare_file=prepare, + build_comicinfo_payload=AsyncMock(), + apply_comicinfo=lambda *_args: None, + cleanup_prepared_file=lambda *_args: None, + record_action=journal, + log_event=AsyncMock(), + register_file=register_with_contention, + move_to_trash=lambda *_args: None, + report_file_progress=report, + session_factory=recovery_factory, + file_worker_count=4, + ) + await session.commit() + return result + + if failure in {"always", "cancel", "second"}: + with pytest.raises(JobCancelledError if failure == "cancel" else JobPausedError): + await execute() + elif failure == "not_locked": + assert await execute() == (0, 2) + assert attempts == 2 + else: + result = await execute() + async with recovery_factory() as session: + errors = list((await session.scalars(select(ImportedFile.error_message))).all()) + assert result == (2, 0), errors + assert maximum == 1 + assert attempts == (3 if failure in {"once", "journal_once"} else 2) + assert inspections == (3 if failure in {"once", "actual_lock", "journal_once"} else 2) + assert any( + p["stage"] == "finalizing" and p["current"] == p["total"] and not p.get("live_only") + for p in progress + ) + + async with recovery_factory() as session: + statuses = list( + (await session.scalars(select(ImportedFile.status).order_by(ImportedFile.id))).all() + ) + expected = ( + ImportedFileStatus.CONFIRMED + if failure in {"always", "cancel"} + else ImportedFileStatus.IMPORTED + ) + if failure == "second": + assert statuses == [ImportedFileStatus.IMPORTED, ImportedFileStatus.CONFIRMED] + elif failure == "not_locked": + assert statuses == [ImportedFileStatus.FAILED, ImportedFileStatus.FAILED] + else: + assert statuses == [expected, expected] + expected_count = ( + 0 if failure in {"always", "cancel", "not_locked"} else 1 if failure == "second" else 2 + ) + assert await session.scalar(select(func.count()).select_from(LibraryFile)) == ( + expected_count + ) + actions = list((await session.scalars(select(ImportJobAction))).all()) + assert len(actions) == expected_count + assert len({action.sequence_no for action in actions}) == len(actions) + if failure == "second": + failure = None + assert await execute() == (1, 0) + async with recovery_factory() as session: + assert await session.scalar(select(func.count()).select_from(LibraryFile)) == 2 + assert await session.scalar(select(func.count()).select_from(ImportJobAction)) == 2 + assert set(await session.scalars(select(ImportedFile.status))) == { + ImportedFileStatus.IMPORTED + } + assert [(p.read_bytes(), p.stat().st_mtime_ns) for p in paths] == originals + + +@pytest.mark.parametrize("state", ["catalogs", "prepared", "completed"]) +@pytest.mark.parametrize("legacy", [False, True]) +async def test_recovery_checkpoints_discard_only_consumed_catalog_caches(db_session, state, legacy): + job = ImportJob( + source_path="/imports", + source_type=ImportSourceType.FILESYSTEM, + status=ImportJobStatus.IMPORTING, + ) + db_session.add(job) + await db_session.flush() + scope = {"state": state, "run_id": "run", "series_ids": [1, 2], "actor_id": 3} + caches = { + key: {"huge": "x" * 200_000} + for key in ( + "candidates", + "completed", + "matches", + "title_candidates", + "title_completed", + "title_matches", + "reference_candidates", + ) + } + original = {**scope, **caches, "catalog_files_prepared": 2} + if legacy: + job.progress_snapshot = {"deferred_recovery": original} + await persist_progress_snapshot( + db_session, + job, + ImportProgressEvent(job_id=job.id, status=job.status, phase="importing", progress=0), + ) + else: + save_recovery_state(job, original) + saved = job.progress_snapshot["deferred_recovery"] + assert all(saved[key] == value for key, value in scope.items()) + assert saved["catalog_files_prepared"] == 2 + if state == "catalogs": + assert saved == original # Still needed to resume preparation without refetching. + else: + assert not set(caches).intersection(saved) + assert len(json.dumps(saved)) < 1024 + assert set(caches).issubset(original) # Do not mutate shared recovery state. diff --git a/tests/unit/test_import_task.py b/tests/unit/test_import_task.py index db50ad1b..8a3d6d03 100644 --- a/tests/unit/test_import_task.py +++ b/tests/unit/test_import_task.py @@ -758,8 +758,9 @@ async def mock_session_ctx(): assert job.progress_snapshot["message"] == "Import cancelled by user." @pytest.mark.asyncio + @pytest.mark.parametrize("database_busy", [False, True]) async def test_execute_task_keeps_truthful_paused_snapshot( - self, db_session: AsyncSession + self, db_session: AsyncSession, database_busy: bool ) -> None: """Paused imports keep their exact import checkpoint instead of being terminalized.""" job = await _create_job(db_session, status=ImportJobStatus.IMPORTING) @@ -778,6 +779,8 @@ async def test_execute_task_keeps_truthful_paused_snapshot( mock_service = AsyncMock() mock_service.run_import.side_effect = JobPausedError("paused") + if database_busy: + mock_service.run_import.side_effect.__cause__ = _sqlite_database_locked_error() @asynccontextmanager async def mock_session_ctx(): @@ -794,11 +797,16 @@ async def mock_session_ctx(): await run_import_execute_task(job.id) await db_session.refresh(job) - assert job.status == ImportJobStatus.PAUSED - assert job.progress_snapshot["status"] == ImportJobStatus.PAUSED.value + expected = ImportJobStatus.STALLED if database_busy else ImportJobStatus.PAUSED + assert job.status == expected + assert job.progress_snapshot["status"] == expected.value assert job.progress_snapshot["phase"] == "importing" assert job.progress_snapshot["progress"] == 64 - assert job.progress_snapshot["message"] == "Import is paused." + assert job.progress_snapshot["message"] == ( + "Import stalled because the database was busy. Resume when ready." + if database_busy + else "Import is paused." + ) assert job.progress_snapshot["current_file_name"] == "Fearscape Vol 02.pdf" assert job.progress_snapshot["current_file_stage"] == "rendering" assert job.progress_snapshot["control_state"]["can_resume"] is True @@ -833,8 +841,9 @@ async def mock_session_ctx(): assert job.error_message is not None @pytest.mark.asyncio + @pytest.mark.parametrize("wrapped", [False, True]) async def test_runner_marks_database_lock_as_stalled( - self, async_engine: object, db_session: AsyncSession + self, async_engine: object, db_session: AsyncSession, wrapped: bool ) -> None: """Transient SQLite lock failures leave imports resumable instead of failed.""" job = await _create_job(db_session, status=ImportJobStatus.IMPORTING) @@ -853,6 +862,10 @@ async def test_runner_marks_database_lock_as_stalled( runner = ImportRunner(factory) mock_service = AsyncMock() mock_service.run_import.side_effect = _sqlite_database_locked_error() + if wrapped: + busy = JobPausedError("Database busy; confirmed matches are preserved.") + busy.__cause__ = _sqlite_database_locked_error() + mock_service.run_import.side_effect = busy with ( patch("pullbox.tasks.import_task._build_import_service", return_value=mock_service), From 961c9d2dddbb90cc157312a32adc6572e0d2f90e Mon Sep 17 00:00:00 2001 From: Adam Hernandez Date: Fri, 18 Sep 2026 13:34:01 -0700 Subject: [PATCH 2/4] fix(import): publish managed files without requiring hard links --- docs/development/IMPORT_REVIEW_RECOVERY.md | 20 ++ src/pullbox/core/file_publication.py | 99 +++++++++ .../import_file_registration_adapters.py | 16 +- .../services/import_managed_copy_preflight.py | 12 +- tests/unit/test_file_publication.py | 192 ++++++++++++++++++ .../test_import_file_registration_adapters.py | 104 +++++++++- .../test_import_managed_copy_preflight.py | 60 ++++++ 7 files changed, 496 insertions(+), 7 deletions(-) create mode 100644 src/pullbox/core/file_publication.py create mode 100644 tests/unit/test_file_publication.py diff --git a/docs/development/IMPORT_REVIEW_RECOVERY.md b/docs/development/IMPORT_REVIEW_RECOVERY.md index 1d6fa998..ff809759 100644 --- a/docs/development/IMPORT_REVIEW_RECOVERY.md +++ b/docs/development/IMPORT_REVIEW_RECOVERY.md @@ -34,6 +34,26 @@ recovery is available in Import Follow-up and does not require an offline command. It works for Mylar and folder imports. This is not a full rescan, a database restore, or an import. +## Managed File Publication + +Mylar and folder imports publish completed sibling staging files with an atomic +no-overwrite rename where available (Linux `RENAME_NOREPLACE`, macOS +`RENAME_EXCL`, or Windows exclusive rename). Copy mode therefore does not +require hard-link support. If exclusive rename is unsupported, the existing +link/unlink publication remains a fallback. Pullbox never falls back to an +overwriting rename or exposes a partially copied final file. Existing targets, +including dangling symlinks, remain protected; journaled stages and source +restoration retain their existing recovery rules. + +Confirmation and execution preflight exercise publication and collision +protection with disposable files under each selected managed destination. +Failure returns an actionable storage message before bulk work starts. The +probe runs off the event loop, does not change import source files, and does not +run for reference-only in-place imports. A root probe cannot guarantee that +every existing subdirectory has the same permissions or that a mount will stay +available; execution failures retain the destination and underlying filesystem +error for retry. No database migration is required. + ## Mylar Inventory And Missing References Copy and keep-in-place scans retain the same Mylar file inventory, including diff --git a/src/pullbox/core/file_publication.py b/src/pullbox/core/file_publication.py new file mode 100644 index 00000000..9ec56132 --- /dev/null +++ b/src/pullbox/core/file_publication.py @@ -0,0 +1,99 @@ +"""Publish completed staging files without replacing an existing destination.""" + +from __future__ import annotations + +import ctypes +import errno +import os +import stat +import sys +import tempfile +from pathlib import Path + +_UNSUPPORTED_RENAME = {errno.ENOSYS, errno.EINVAL, errno.ENOTSUP, errno.EOPNOTSUPP} +_AT_FDCWD = -100 +_RENAME_NOREPLACE = 1 +_RENAME_EXCL = 4 + + +def publish_file_without_overwrite(stage: Path, target: Path) -> None: + """Consume a stage using an atomic destination claim, never plain POSIX rename. + + Native exclusive rename avoids requiring hard-link support on managed-copy + destinations. Older filesystems retain the link/unlink fallback. Neither + path copies bytes into a visible, partially written final file. + """ + mode = stage.lstat().st_mode + if not (stat.S_ISREG(mode) or stat.S_ISLNK(mode)): + raise OSError(errno.EINVAL, "Publication requires a regular file or symlink", str(stage)) + if _native_rename_without_overwrite(stage, target): + return + if stat.S_ISLNK(mode): + os.symlink(os.readlink(stage), target) + else: + os.link(stage, target, follow_symlinks=False) + stage.unlink() + + +def _native_rename_without_overwrite(stage: Path, target: Path) -> bool: + if sys.platform == "win32": + # Windows rename fails if the destination exists; POSIX rename does not. + os.rename(stage, target) + return True + if sys.platform not in {"linux", "darwin"}: + return False + + source_bytes, target_bytes = os.fsencode(stage), os.fsencode(target) + if b"\0" in source_bytes or b"\0" in target_bytes: + raise ValueError("embedded null byte") + libc = ctypes.CDLL(None, use_errno=True) + symbol = "renameat2" if sys.platform == "linux" else "renamex_np" + rename = getattr(libc, symbol, None) + if rename is None: + return False + rename.restype = ctypes.c_int + if sys.platform == "linux": + rename.argtypes = [ + ctypes.c_int, + ctypes.c_char_p, + ctypes.c_int, + ctypes.c_char_p, + ctypes.c_uint, + ] + result = rename(_AT_FDCWD, source_bytes, _AT_FDCWD, target_bytes, _RENAME_NOREPLACE) + else: + rename.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_uint] + result = rename(source_bytes, target_bytes, _RENAME_EXCL) + if result == 0: + return True + error_number = ctypes.get_errno() + if error_number in _UNSUPPORTED_RENAME: + return False + raise OSError(error_number, os.strerror(error_number), str(stage), None, str(target)) + + +def probe_file_publication(root: Path) -> None: + """Exercise publication and collision protection using only disposable files.""" + with tempfile.TemporaryDirectory(prefix=".pullbox-publication-probe-", dir=root) as directory: + stage = Path(directory) / "stage" + target = Path(directory) / "target" + stage.write_bytes(b"original") + publish_file_without_overwrite(stage, target) + stage.write_bytes(b"replacement") + try: + publish_file_without_overwrite(stage, target) + except FileExistsError: + if target.read_bytes() == b"original" and stage.read_bytes() == b"replacement": + return + raise OSError(errno.ENOTSUP, "Storage did not preserve an existing destination", str(root)) + + +def publication_failure_message(directory: Path, error: OSError) -> str: + """Actionable destination guidance shared by preflight and execution failures.""" + return ( + f"Cannot safely publish imported files in {directory}: {error.strerror or str(error)}. " + "Check the destination mount, free space, and the container user's file permissions. " + "The destination must support exclusive rename or hard links without replacing " + "existing files. Correct the storage settings or choose another managed library root, " + "then retry." + ) diff --git a/src/pullbox/services/import_file_registration_adapters.py b/src/pullbox/services/import_file_registration_adapters.py index 36ca12b0..37007f5d 100644 --- a/src/pullbox/services/import_file_registration_adapters.py +++ b/src/pullbox/services/import_file_registration_adapters.py @@ -11,6 +11,10 @@ from typing import TYPE_CHECKING, Any, cast from pullbox.core.exceptions import ImportDestinationValidationError +from pullbox.core.file_publication import ( + publication_failure_message, + publish_file_without_overwrite, +) if TYPE_CHECKING: from collections.abc import Awaitable, Callable @@ -246,17 +250,19 @@ def _publish_stage_without_overwrite(stage_path: Path, target_path: Path) -> Non f"Managed import destination appeared during import and was preserved: {collision}", ) try: - if stage_path.is_symlink(): - os.symlink(os.readlink(stage_path), target_path) - else: - os.link(stage_path, target_path, follow_symlinks=False) + publish_file_without_overwrite(stage_path, target_path) except FileExistsError as exc: raise ImportDestinationValidationError( "destination_appeared", "Managed import destination appeared during import and was preserved: " f"{target_path}", ) from exc - stage_path.unlink() + except OSError as exc: + raise OSError( + exc.errno, + publication_failure_message(target_path.parent, exc), + str(target_path), + ) from exc def _casefold_destination_collision(target_path: Path) -> Path | None: diff --git a/src/pullbox/services/import_managed_copy_preflight.py b/src/pullbox/services/import_managed_copy_preflight.py index d7de457f..551ac1eb 100644 --- a/src/pullbox/services/import_managed_copy_preflight.py +++ b/src/pullbox/services/import_managed_copy_preflight.py @@ -17,6 +17,7 @@ from pullbox.config import get_settings from pullbox.core.exceptions import ConfigurationError, ValidationError +from pullbox.core.file_publication import probe_file_publication, publication_failure_message from pullbox.core.library_root_resolution import resolve_library_root from pullbox.models.import_job import ( ImportControlRequest, @@ -51,6 +52,7 @@ class ManagedCopyPreflightFailure(enum.StrEnum): TARGET_DISABLED = "target_disabled" TARGET_REFERENCE_ONLY = "target_reference_only" TARGET_UNAVAILABLE = "target_unavailable" + TARGET_PUBLICATION_UNAVAILABLE = "target_publication_unavailable" CAPACITY_UNKNOWN = "capacity_unknown" CAPACITY_INSUFFICIENT = "capacity_insufficient" @@ -630,12 +632,20 @@ def _require_managed_root_roles(root: LibraryRoot) -> LibraryRoot: async def _validate_live_managed_root(root: LibraryRoot) -> dict[str, object]: try: - return await validate_managed_library_root(root) + capabilities = await validate_managed_library_root(root) except ValidationError as exc: raise ManagedCopyPreflightError( ManagedCopyPreflightFailure.TARGET_UNAVAILABLE, exc.message, ) from exc + try: + await asyncio.to_thread(probe_file_publication, Path(root.path)) + except OSError as exc: + raise ManagedCopyPreflightError( + ManagedCopyPreflightFailure.TARGET_PUBLICATION_UNAVAILABLE, + publication_failure_message(Path(root.path), exc), + ) from exc + return capabilities def _free_bytes_from_capabilities(capabilities: dict[str, object]) -> int | None: diff --git a/tests/unit/test_file_publication.py b/tests/unit/test_file_publication.py new file mode 100644 index 00000000..9f04099d --- /dev/null +++ b/tests/unit/test_file_publication.py @@ -0,0 +1,192 @@ +"""Atomic publication must never trade compatibility for no-overwrite safety.""" + +from __future__ import annotations + +import ctypes +import errno +import os +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from pullbox.core import file_publication + + +@pytest.mark.parametrize("fallback", [False, True]) +@pytest.mark.parametrize("kind", ["file", "symlink"]) +def test_publication_consumes_stage_without_changing_its_identity( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, fallback: bool, kind: str +) -> None: + source = tmp_path / "source.cbz" + source.write_bytes(b"comic") + stage = tmp_path / "stage" + if kind == "file": + stage.hardlink_to(source) + else: + stage.symlink_to(source.name) + if fallback: + monkeypatch.setattr( + file_publication, "_native_rename_without_overwrite", lambda *args: False + ) + target = tmp_path / "target.cbz" + + file_publication.publish_file_without_overwrite(stage, target) + + assert not os.path.lexists(stage) + assert target.read_bytes() == b"comic" + assert target.is_symlink() == (kind == "symlink") + assert target.samefile(source) + if kind == "symlink": + assert os.readlink(target) == source.name + + +@pytest.mark.parametrize("fallback", [False, True]) +@pytest.mark.parametrize("kind", ["file", "directory", "dangling_symlink"]) +def test_publication_preserves_every_existing_destination( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, fallback: bool, kind: str +) -> None: + stage = tmp_path / "stage" + stage.write_bytes(b"new comic") + target = tmp_path / "target" + if kind == "file": + target.write_bytes(b"existing comic") + elif kind == "directory": + target.mkdir() + else: + target.symlink_to("missing") + original = target.lstat() + if fallback: + monkeypatch.setattr( + file_publication, "_native_rename_without_overwrite", lambda *args: False + ) + + with pytest.raises(FileExistsError): + file_publication.publish_file_without_overwrite(stage, target) + + assert stage.read_bytes() == b"new comic" + assert target.lstat() == original + if kind == "file": + assert target.read_bytes() == b"existing comic" + + +@pytest.mark.parametrize( + "platform,symbol,flags", + [ + ("linux", "renameat2", 1), + ("darwin", "renamex_np", 4), + ], +) +@pytest.mark.parametrize( + "error_number", + [ + 0, + errno.ENOSYS, + errno.EINVAL, + errno.ENOTSUP, + errno.EPERM, + errno.EEXIST, + errno.EIO, + errno.ENOSPC, + ], +) +def test_native_rename_abi_and_errors( + monkeypatch: pytest.MonkeyPatch, platform: str, symbol: str, flags: int, error_number: int +) -> None: + native = Mock(return_value=-1 if error_number else 0) + library = Mock(return_value=SimpleNamespace(**{symbol: native})) + monkeypatch.setattr(file_publication, "sys", SimpleNamespace(platform=platform)) + monkeypatch.setattr(ctypes, "CDLL", library) + monkeypatch.setattr(ctypes, "get_errno", lambda: error_number) + source, target = Path("/library/stage"), Path("/library/target") + + if error_number and error_number not in file_publication._UNSUPPORTED_RENAME: + with pytest.raises(OSError) as error: + file_publication._native_rename_without_overwrite(source, target) + assert error.value.errno == error_number + assert error.value.filename == str(source) + assert error.value.filename2 == str(target) + else: + assert file_publication._native_rename_without_overwrite(source, target) is ( + error_number == 0 + ) + + library.assert_called_once_with(None, use_errno=True) + if platform == "linux": + native.assert_called_once_with(-100, b"/library/stage", -100, b"/library/target", flags) + assert native.argtypes == [ + ctypes.c_int, + ctypes.c_char_p, + ctypes.c_int, + ctypes.c_char_p, + ctypes.c_uint, + ] + else: + native.assert_called_once_with(b"/library/stage", b"/library/target", flags) + assert native.argtypes == [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_uint] + assert native.restype == ctypes.c_int + + +@pytest.mark.parametrize("error_number", [errno.EPERM, errno.EIO, errno.ENOSPC]) +def test_real_errors_do_not_try_another_publication_method( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, error_number: int +) -> None: + stage, target = tmp_path / "stage", tmp_path / "target" + stage.write_bytes(b"comic") + native = Mock(side_effect=OSError(error_number, os.strerror(error_number))) + link = Mock() + monkeypatch.setattr(file_publication, "_native_rename_without_overwrite", native) + monkeypatch.setattr(os, "link", link) + + with pytest.raises(OSError) as error: + file_publication.publish_file_without_overwrite(stage, target) + + assert error.value.errno == error_number + link.assert_not_called() + assert stage.read_bytes() == b"comic" + assert not target.exists() + + +def test_publication_rejects_a_directory_stage(tmp_path: Path) -> None: + stage, target = tmp_path / "stage", tmp_path / "target" + stage.mkdir() + with pytest.raises(OSError, match="regular file or symlink"): + file_publication.publish_file_without_overwrite(stage, target) + assert stage.is_dir() + assert not target.exists() + + +def test_native_missing_symbol_uses_link_fallback( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(ctypes, "CDLL", lambda *args, **kwargs: SimpleNamespace()) + file_publication.probe_file_publication(tmp_path) + assert not list(tmp_path.iterdir()) + + +def test_native_rejects_null_bytes_in_target(tmp_path: Path) -> None: + stage = tmp_path / "stage" + stage.write_bytes(b"comic") + with pytest.raises(ValueError, match="null byte"): + file_publication.publish_file_without_overwrite(stage, tmp_path / "target\0suffix") + assert stage.read_bytes() == b"comic" + assert not (tmp_path / "target").exists() + + +def test_probe_checks_no_overwrite_contract_and_cleans_up( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(file_publication, "publish_file_without_overwrite", os.replace) + with pytest.raises(OSError, match="preserve an existing destination"): + file_publication.probe_file_publication(tmp_path) + assert not list(tmp_path.iterdir()) + + +def test_windows_uses_exclusive_os_rename(monkeypatch: pytest.MonkeyPatch) -> None: + rename = Mock() + monkeypatch.setattr(file_publication, "sys", SimpleNamespace(platform="win32")) + monkeypatch.setattr(os, "rename", rename) + source, target = Path("stage"), Path("target") + assert file_publication._native_rename_without_overwrite(source, target) + rename.assert_called_once_with(source, target) diff --git a/tests/unit/test_import_file_registration_adapters.py b/tests/unit/test_import_file_registration_adapters.py index 9e231292..0028b537 100644 --- a/tests/unit/test_import_file_registration_adapters.py +++ b/tests/unit/test_import_file_registration_adapters.py @@ -2,15 +2,17 @@ from __future__ import annotations +import errno import os from pathlib import Path from types import SimpleNamespace from typing import Any -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, Mock import pytest from pullbox.core.exceptions import ImportDestinationValidationError +from pullbox.services import import_file_registration_adapters from pullbox.services.import_file_registration_adapters import ( build_import_library_file_adapters, ) @@ -31,6 +33,106 @@ def _build_adapters( ) +@pytest.mark.asyncio +@pytest.mark.parametrize("materialize", [False, True]) +async def test_copy_publication_does_not_require_hardlink_support( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, materialize: bool +) -> None: + source = tmp_path / "incoming.cbz" + source.write_bytes(b"source comic") + target = tmp_path / "library" / "issue.cbz" + target.parent.mkdir() + + def refuse_hardlink(*args: object, **kwargs: object) -> None: + raise PermissionError(errno.EPERM, "Operation not permitted") + + monkeypatch.setattr(os, "link", refuse_hardlink) + + async def write_stage( + _session: object, + _job: object, + source_path: Path, + stage_path: Path, + *_args: object, + **_kwargs: object, + ) -> bool: + stage_path.write_bytes(source_path.read_bytes()) + return True + + adapters = _build_adapters(transfer=write_stage, materializer=write_stage) + if materialize: + await adapters.comicinfo_materializer(source, target, {}, transfer_method="copy") + else: + await adapters.artifact_transfer(source, target, "copy") + + assert target.read_bytes() == b"source comic" + assert source.read_bytes() == b"source comic" + assert list(target.parent.iterdir()) == [target] + + +def test_atomic_publication_catches_destination_appearing_after_collision_check( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + stage, target = tmp_path / "stage", tmp_path / "target" + stage.write_bytes(b"new comic") + publish = import_file_registration_adapters.publish_file_without_overwrite + + def racing_publish(source_path: Path, target_path: Path) -> None: + target_path.write_bytes(b"late arrival") + publish(source_path, target_path) + + monkeypatch.setattr( + import_file_registration_adapters, "publish_file_without_overwrite", racing_publish + ) + + with pytest.raises(ImportDestinationValidationError, match="appeared during import"): + import_file_registration_adapters._publish_stage_without_overwrite(stage, target) + + assert target.read_bytes() == b"late arrival" + assert stage.read_bytes() == b"new comic" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("materialize", [False, True]) +async def test_publication_failure_is_actionable_and_preserves_copy_source( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, materialize: bool +) -> None: + source, target = tmp_path / "source", tmp_path / "target" + source.write_bytes(b"comic") + + async def write_stage( + _session: object, + _job: object, + source_path: Path, + stage_path: Path, + *_args: object, + **_kwargs: object, + ) -> bool: + stage_path.write_bytes(source_path.read_bytes()) + return True + + monkeypatch.setattr( + import_file_registration_adapters, + "publish_file_without_overwrite", + Mock(side_effect=PermissionError(errno.EPERM, "Operation not permitted")), + ) + adapters = _build_adapters(transfer=write_stage, materializer=write_stage) + + with pytest.raises(PermissionError, match="Cannot safely publish") as error: + if materialize: + await adapters.comicinfo_materializer(source, target, {}, transfer_method="copy") + else: + await adapters.artifact_transfer(source, target, "copy") + + assert error.value.errno == errno.EPERM + assert "Operation not permitted" in str(error.value) + assert str(tmp_path) in str(error.value) + assert "container user's file permissions" in str(error.value) + assert source.read_bytes() == b"comic" + assert not target.exists() + assert not list(tmp_path.glob(".pullbox-import-*")) + + @pytest.mark.asyncio async def test_import_registration_adapters_forward_callbacks_and_record_timings( tmp_path: Path, diff --git a/tests/unit/test_import_managed_copy_preflight.py b/tests/unit/test_import_managed_copy_preflight.py index 62bc1495..4d3d7ce1 100644 --- a/tests/unit/test_import_managed_copy_preflight.py +++ b/tests/unit/test_import_managed_copy_preflight.py @@ -2,6 +2,9 @@ from __future__ import annotations +import ctypes +import errno +import os from datetime import UTC, datetime from pathlib import Path from types import SimpleNamespace @@ -50,6 +53,63 @@ def test_managed_copy_capacity_reserve_uses_fixed_minimum_then_ten_percent() -> assert managed_copy_capacity_reserve(20 * 1024**3) == 2 * 1024**3 +@pytest.mark.parametrize("source_type", [ImportSourceType.MYLAR3, ImportSourceType.FILESYSTEM]) +@pytest.mark.parametrize("stage", ["confirmation", "execution"]) +async def test_copy_preflight_rejects_storage_without_safe_publication( + db_session: AsyncSession, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + source_type: ImportSourceType, + stage: import_managed_copy_preflight.ManagedCopyPreflightStage, +) -> None: + root = LibraryRoot(name="Managed", path=str(tmp_path), enabled=True) + db_session.add(root) + await db_session.flush() + job = ImportJob( + source_path="/imports", + source_type=source_type, + target_library_root_id=root.id, + file_handling_mode=ImportFileHandlingMode.MANAGED_COPY, + ) + db_session.add(job) + await db_session.flush() + + def refuse_hardlink(*args: object, **kwargs: object) -> None: + raise PermissionError(errno.EPERM, "Operation not permitted") + + monkeypatch.setattr(ctypes, "CDLL", lambda *args, **kwargs: SimpleNamespace()) + monkeypatch.setattr(os, "link", refuse_hardlink) + + with pytest.raises(ManagedCopyPreflightError, match="safely publish") as error: + await validate_managed_copy_preflight(db_session, job, stage=stage) + + assert error.value.reason.value == "target_publication_unavailable" + assert str(tmp_path) in error.value.message + assert "container" in error.value.message + assert not list(tmp_path.glob(".pullbox-*")) + + +@pytest.mark.parametrize("source_type", [ImportSourceType.MYLAR3, ImportSourceType.FILESYSTEM]) +async def test_in_place_preflight_never_probes_source_for_writes( + db_session: AsyncSession, + monkeypatch: pytest.MonkeyPatch, + source_type: ImportSourceType, +) -> None: + job = ImportJob( + source_path="/read-only-library", + source_type=source_type, + file_handling_mode=ImportFileHandlingMode.IN_PLACE, + ) + db_session.add(job) + await db_session.flush() + + def unexpected_probe(*args: object, **kwargs: object) -> None: + pytest.fail("In-place import must not write to the source") + + monkeypatch.setattr(import_managed_copy_preflight, "probe_file_publication", unexpected_probe) + assert await validate_managed_copy_preflight(db_session, job, stage="execution") is None + + async def test_selected_source_bytes_counts_new_and_selected_duplicate_files( db_session: AsyncSession, ) -> None: From 2a12e2807498309c73af3aa13d5e1718303b80c8 Mon Sep 17 00:00:00 2001 From: Adam Hernandez Date: Fri, 18 Sep 2026 15:21:34 -0700 Subject: [PATCH 3/4] fix(import): recover mixed-folder file identities safely --- docs/development/IMPORT_REVIEW_RECOVERY.md | 17 ++ .../import_deferred_recovery_execution.py | 66 +++--- .../services/import_reconcile_helpers.py | 9 + .../services/import_recovery_identity.py | 160 +++++++++++++ .../services/import_reference_recovery.py | 116 ++++++---- .../test_import_mixed_identity_recovery.py | 211 ++++++++++++++++++ 6 files changed, 503 insertions(+), 76 deletions(-) create mode 100644 src/pullbox/services/import_recovery_identity.py create mode 100644 tests/unit/test_import_mixed_identity_recovery.py diff --git a/docs/development/IMPORT_REVIEW_RECOVERY.md b/docs/development/IMPORT_REVIEW_RECOVERY.md index ff809759..e7dd14c8 100644 --- a/docs/development/IMPORT_REVIEW_RECOVERY.md +++ b/docs/development/IMPORT_REVIEW_RECOVERY.md @@ -241,6 +241,23 @@ checkpoint. Resume preserves completed repairs, and repeated runs do not create another file registration. This is logical library repair, not authorization to reorganize the user's filesystem. +Recovery reads the original per-file ComicInfo title and issue designation even +when older imports saved the parent folder's name as the parsed title. Explicit +reading-order prefixes and `(converted)` annotations are removed for candidate +lookups only; filenames remain unchanged. Deferred and referenced files share +the same per-file catalog checks, including publication dates and issue types. +A date accepted for one file never authorizes another file in the same batch. +Conflicting embedded identities, manual decisions, unsafe sources and duplicate +copies remain protected. Unresolved catalog candidates retain their proposed +series, issue and reason in `mixed_folder_recovery` diagnostics. + +An issue mistakenly registered under the parent series can itself be reassigned +only when its embedded ComicVine issue ID agrees with the unique catalog target, +it is the current referenced issue, and no other file or imported registration +owns it. The issue ID and reader history are preserved. The old series identity +is retained in the cleanup audit. A provisional issue is not offered under a +folder series when independent file evidence identifies an unrelated series. + The deferred pass uses complete local catalogs first. Exact issue identity may correct stale Mylar ownership only when the file's title, issue number, type, and embedded identity agree with the target. Conflicting embedded IDs remain diff --git a/src/pullbox/services/import_deferred_recovery_execution.py b/src/pullbox/services/import_deferred_recovery_execution.py index d338cc30..f0d5f24c 100644 --- a/src/pullbox/services/import_deferred_recovery_execution.py +++ b/src/pullbox/services/import_deferred_recovery_execution.py @@ -37,6 +37,11 @@ refresh_recovered_groups, ) from pullbox.services.import_recovery_checkpoint import compact_recovery_state +from pullbox.services.import_recovery_identity import ( + catalog_file_identity, + catalog_target_agrees, + record_catalog_review, +) from pullbox.services.import_reference_recovery import ( reference_candidates, repair_catalog_references, @@ -80,45 +85,31 @@ def _catalog_summary_payload(summary: IssueSummary) -> dict[str, Any]: def _filename_catalog_identity( file: ImportedFile, item: ImportedSeries, -) -> dict[str, str | int] | None: +) -> dict[str, Any] | None: """Return bounded filename identity eligible for local-catalog recovery.""" - if protected_file(file, item) or provider_ids(file): + if protected_file(file, item): return None diagnostics = dict(file.diagnostics or {}) - source = source_metadata_for_import_file(item, file).diagnostics - if source.get("identity_conflicts"): - return None - parsed = source.get("filename_parse") - parsed = parsed if isinstance(parsed, dict) else {} - source_title = str(parsed.get("series_name") or "").strip() - raw_number = parsed.get("issue_number_text") or parsed.get("issue_number") - if not source_title or raw_number is None: + identity = catalog_file_identity(file) + if identity is None: return None - try: - issue_number = normalize_issue_number_text(str(raw_number)) - except ValueError: + ids = provider_ids(file) + if ids and ids != {identity.get("issue_cv_id")}: return None - issue_type = str(parsed.get("issue_type") or diagnostics.get("source_issue_type") or "issue") - if issue_type in {"annual", "special"}: - label = issue_type.title() - if not NameMatcher.normalize(source_title).endswith(f" {NameMatcher.normalize(label)}"): - source_title = f"{source_title} {label}" - normalized_title = NameMatcher.normalize(source_title) + source_title = identity["query"] + normalized_title = identity["key"] parent_title = NameMatcher.normalize(item.cv_title or item.raw_series_name) if not normalized_title or normalized_title == parent_title: return None title_match = NameMatcher().match(source_title, item.cv_title or item.raw_series_name) corroborated = diagnostics.get("conflict_type") == "corroborated_file_series_mismatch" - if title_match.is_match and not corroborated and issue_type not in {"annual", "special"}: + if ( + title_match.is_match + and not corroborated + and identity["issue_type"] not in {"annual", "special"} + ): return None - year = parsed.get("year") or file.parsed_year - return { - "key": normalized_title, - "query": source_title, - "issue_number": issue_number, - "year": int(year) if isinstance(year, int | float) else 0, - "issue_type": issue_type, - } + return identity async def _catalog_title_candidates( @@ -435,7 +426,17 @@ async def _prepare_title_catalog_targets(session: AsyncSession, job: ImportJob) continue options_by_number = title_matches.get(str(identity["key"]), {}) options = options_by_number.get(str(identity["issue_number"]), []) + options = [option for option in options if catalog_target_agrees(identity, option)] if len(options) != 1: + record_catalog_review( + file, + identity, + ( + "Multiple catalog issues agree; choose the correct series and issue." + if options + else "No catalog issue agrees with this file's title, number, type and year." + ), + ) continue eligible.append((file, item, options[0])) @@ -469,6 +470,15 @@ async def _prepare_title_catalog_targets(session: AsyncSession, job: ImportJob) or counts[str(issue_cv_id)] != 1 or issue_cv_id in existing_issue_ids ): + identity = _filename_catalog_identity(file, original) + if identity is not None: + record_catalog_review( + file, + identity, + "Multiple files claim this issue; choose which copy to keep." + if counts[str(issue_cv_id)] != 1 + else "This issue already has a catalog entry; review its existing ownership.", + ) continue target_item = targets.get(target_cv_id) if target_item is None: diff --git a/src/pullbox/services/import_reconcile_helpers.py b/src/pullbox/services/import_reconcile_helpers.py index 342b4d8e..59ba177e 100644 --- a/src/pullbox/services/import_reconcile_helpers.py +++ b/src/pullbox/services/import_reconcile_helpers.py @@ -18,6 +18,7 @@ PROVIDER_MISSING_ISSUE_PLACEHOLDER_METHOD, ) from pullbox.services.import_file_preparation import format_comicinfo_issue_number +from pullbox.services.import_recovery_identity import catalog_file_identity if TYPE_CHECKING: from collections.abc import Callable @@ -143,6 +144,14 @@ def provisional_issue_number_for_file( diagnostics = dict(imp_file.diagnostics or {}) if diagnostics.get("kind") == "metadata_conflict" or diagnostics.get("conflict_type"): return None + identity = catalog_file_identity(imp_file) + if ( + identity is not None + and not NameMatcher() + .match(identity["query"], item.cv_title or item.raw_series_name) + .is_match + ): + return None issue_number = archive_entry_issue_number(imp_file) if issue_number is None: diff --git a/src/pullbox/services/import_recovery_identity.py b/src/pullbox/services/import_recovery_identity.py new file mode 100644 index 00000000..42888190 --- /dev/null +++ b/src/pullbox/services/import_recovery_identity.py @@ -0,0 +1,160 @@ +"""File-level catalog evidence shared by deferred and referenced-file recovery.""" + +from __future__ import annotations + +import re +from typing import TYPE_CHECKING, Any + +from pullbox.core.issue_numbers import normalize_issue_number_text +from pullbox.core.name_matcher import NameMatcher +from pullbox.core.release_parser import parse_release_title +from pullbox.core.story_arc_ordering import extract_story_arc_order_prefix + +if TYPE_CHECKING: + from pullbox.models.import_job import ImportedFile + + +def catalog_file_identity(file: ImportedFile) -> dict[str, Any] | None: + """Extract a candidate without inheriting the parent folder's series or year.""" + diagnostics = dict(file.diagnostics or {}) + source = diagnostics.get("source_metadata") or {} + signals = diagnostics.get("metadata_signals") or {} + if not isinstance(source, dict) or not isinstance(signals, dict): + return None + if source.get("identity_conflicts") or diagnostics.get("identity_conflicts"): + return None + comicinfo = source.get("comicinfo") or {} + if not isinstance(comicinfo, dict): + return None + embedded_title = str(comicinfo.get("series") or "").strip() + embedded_number = comicinfo.get("number") + use_embedded = bool( + embedded_title + and embedded_number is not None + and signals.get("series_name") == "comicinfo" + and signals.get("issue_number") == "comicinfo" + ) + use_sidecar = signals.get("series_name") == signals.get("issue_number") == "sidecar" and bool( + file.parsed_series + ) + name = file.file_name + prefix = extract_story_arc_order_prefix(name) + if prefix is not None: + name = prefix.residual_file_name + name = re.sub(r"\s*\(converted\)\s*", " ", name, flags=re.IGNORECASE) + parsed = parse_release_title(name, expected_series=(embedded_title,) if use_embedded else ()) + if parsed is not None and parsed.is_pack: + return None + if use_embedded: + title = embedded_title + raw_number = embedded_number + evidence = "comicinfo" + elif use_sidecar: + title = file.parsed_series or "" + raw_number = file.issue_number_raw or file.parsed_issue_number + evidence = "sidecar" + else: + if parsed is None or parsed.volume is not None or parsed.issue_number is None: + return None + title = parsed.series_name or "" + raw_number = parsed.issue_number_text or parsed.issue_number + evidence = "filename_parse" + if not isinstance(raw_number, str | float | int): + return None + try: + number = normalize_issue_number_text(raw_number) + except (TypeError, ValueError): + return None + # Embedded metadata can be generated or stale; don't erase a contradictory + # issue designation in an otherwise recognizable filename. + if ( + use_embedded + and parsed is not None + and parsed.issue_number is not None + and normalize_issue_number_text(parsed.issue_number_text or parsed.issue_number) != number + ): + return None + if ( + use_embedded + and parsed is not None + and parsed.series_name + and not NameMatcher().match(parsed.series_name, title).is_match + ): + return None + issue_type = str(diagnostics.get("source_issue_type") or "issue") + if parsed is not None and issue_type == "issue": + issue_type = parsed.issue_type.value + if issue_type in {"annual", "special"} and not NameMatcher.normalize(title).endswith( + f" {issue_type}" + ): + title = f"{title} {issue_type.title()}" + year = parsed.year if parsed is not None else None + if not year and use_embedded: + raw_year = comicinfo.get("year") + if isinstance(raw_year, int) or (isinstance(raw_year, str) and raw_year.isdigit()): + year = int(raw_year) + if not title or (not year and evidence == "filename_parse"): + return None + return { + "key": NameMatcher.normalize(title), + "query": title, + "issue_number": number, + "year": year or 0, + "issue_type": issue_type, + "series_cv_id": diagnostics.get("comicvine_series_id") + if signals.get("comicvine_series_id") in {"comicinfo", "sidecar"} + else None, + "issue_cv_id": file.comicvine_issue_id + if signals.get("comicvine_issue_id") in {"comicinfo", "sidecar"} + else None, + "evidence": evidence, + } + + +def catalog_target_agrees(identity: dict[str, Any], target: dict[str, Any]) -> bool: + """Validate each file, even when lookup results were grouped across dates.""" + summary = target["summary"] + target_type = str(summary.get("issue_type") or "issue") + source_type = identity["issue_type"] + # Dedicated Annual/Special volumes commonly catalog their numbered entries + # as ordinary issues. Their exact qualified title still distinguishes them. + type_agrees = target_type == source_type or ( + source_type in {"annual", "special"} + and target_type == "issue" + and identity["key"].endswith(f" {source_type}") + ) + if ( + NameMatcher.normalize(str(target["title"])) != identity["key"] + or not type_agrees + or (identity.get("series_cv_id") and str(identity["series_cv_id"]) != str(target["cv_id"])) + or ( + identity.get("issue_cv_id") + and str(identity["issue_cv_id"]) != str(summary["provider_id"]) + ) + ): + return False + try: + if ( + normalize_issue_number_text(summary.get("issue_number_text") or summary["issue_number"]) + != identity["issue_number"] + ): + return False + release_date = summary.get("release_date") + if identity["year"] and release_date: + return abs(int(str(release_date)[:4]) - int(identity["year"])) <= 1 + except (TypeError, ValueError): + return False + return bool(identity["evidence"] != "filename_parse") + + +def record_catalog_review(file: ImportedFile, identity: dict[str, Any], reason: str) -> None: + """Keep unresolved candidate evidence without changing any import decision.""" + file.diagnostics = { + **file.diagnostics, + "mixed_folder_recovery": { + "candidate_series": identity["query"], + "candidate_issue": identity["issue_number"], + "evidence": identity["evidence"], + "reason": reason, + }, + } diff --git a/src/pullbox/services/import_reference_recovery.py b/src/pullbox/services/import_reference_recovery.py index ba4474de..e60edb9d 100644 --- a/src/pullbox/services/import_reference_recovery.py +++ b/src/pullbox/services/import_reference_recovery.py @@ -13,10 +13,8 @@ from sqlalchemy import and_, or_, select from pullbox.core.exceptions import ConfigurationError -from pullbox.core.issue_numbers import normalize_issue_number_text from pullbox.core.library_file_ownership import build_file_identity_signature from pullbox.core.name_matcher import NameMatcher -from pullbox.core.release_parser import parse_release_title from pullbox.models.import_job import ( ImportedFile, ImportedFileStatus, @@ -30,11 +28,15 @@ from pullbox.services.catalog.reader import CatalogIssueSummary, CatalogSeriesMetadata from pullbox.services.import_deferred_recovery import ( apply_proven_identity, - positive_id, provider_ids, refresh_recovered_groups, same_source, ) +from pullbox.services.import_recovery_identity import ( + catalog_file_identity, + catalog_target_agrees, + record_catalog_review, +) if TYPE_CHECKING: from collections.abc import Awaitable, Callable @@ -69,6 +71,9 @@ def reference_candidate_ids(job_id: int) -> Select[tuple[int]]: ), or_( ImportedFile.parsed_series != ImportedSeries.raw_series_name, + ImportedFile.diagnostics["source_metadata"]["comicinfo"]["series"].as_string() + != ImportedSeries.raw_series_name, + ImportedFile.match_method == "import_reconcile_provisional_issue", and_( ImportedSeries.files_no_match > 0, ImportedFile.diagnostics["metadata_signals"]["issue_number"].as_string() @@ -112,8 +117,6 @@ def _stamp(file: ImportedFile, item: ImportedSeries, library: LibraryFile) -> st async def reference_candidates(session: AsyncSession, job_id: int) -> dict[str, dict[str, Any]]: """Freeze exact identity evidence without inspecting or modifying source files.""" - from pullbox.services.import_completed_cleanup import _mixed_folder_source_identity - result: dict[str, dict[str, Any]] = {} rows = await session.stream( select(ImportedFile, ImportedSeries, LibraryFile, Issue) @@ -145,10 +148,11 @@ async def reference_candidates(session: AsyncSession, job_id: int) -> dict[str, or source.get("identity_conflicts") ): continue - identity = _mixed_folder_source_identity(file) + identity = catalog_file_identity(file) if identity is None: continue - title, raw_number, series_cv_id, issue_cv_id, evidence = identity + issue_cv_id = identity["issue_cv_id"] + evidence = identity["evidence"] # A derived old match is not source evidence. Any other saved ID must # agree with the trusted embedded identity, or remain for review. allowed_ids = {value for value in (old_issue.comicvine_id, issue_cv_id) if value} @@ -158,27 +162,11 @@ async def reference_candidates(session: AsyncSession, job_id: int) -> dict[str, or (source.get("comicinfo") and evidence == "filename_parse" and provider_ids(file)) ): continue - key = NameMatcher.normalize(title) + key = identity["key"] if not key or key == NameMatcher.normalize(item.cv_title or item.raw_series_name): continue - try: - number = normalize_issue_number_text(raw_number) - except ValueError: - continue - parsed = parse_release_title(file.file_name) - year = parsed.year if parsed is not None else file.parsed_year - issue_type = str(diagnostics.get("source_issue_type") or "issue") - if evidence == "filename_parse" and parsed is not None: - issue_type = parsed.issue_type.value result[str(file.id)] = { - "key": key, - "query": title, - "issue_number": number, - "year": year or 0, - "issue_type": issue_type, - "series_cv_id": series_cv_id, - "issue_cv_id": issue_cv_id, - "evidence": evidence, + **identity, "stamp": _stamp(file, item, library), } finally: @@ -187,25 +175,7 @@ async def reference_candidates(session: AsyncSession, job_id: int) -> dict[str, def _target_agrees(identity: dict[str, Any], target: dict[str, Any]) -> bool: - summary = target["summary"] - if ( - NameMatcher.normalize(str(target["title"])) != identity["key"] - or str(summary.get("issue_type") or "issue") != identity["issue_type"] - or (identity["series_cv_id"] and identity["series_cv_id"] != target["cv_id"]) - or ( - identity["issue_cv_id"] - and identity["issue_cv_id"] != positive_id(summary["provider_id"]) - ) - ): - return False - release_date = summary.get("release_date") - if identity["year"] and release_date: - try: - return abs(int(str(release_date)[:4]) - int(identity["year"])) <= 1 - except ValueError: - return False - # Filename-only evidence cannot distinguish same-name reboots without a date. - return bool(identity["evidence"] != "filename_parse") + return catalog_target_agrees(identity, target) def _unchanged_source(path: str, signature: dict[str, Any], root_path: str) -> bool: @@ -251,6 +221,16 @@ async def repair_catalog_references( options = [option for option in options if _target_agrees(identity, option)] if len(options) == 1: plans.append((int(file_id_text), identity, options[0])) + else: + file = await session.get(ImportedFile, int(file_id_text)) + if file is not None: + record_catalog_review( + file, + identity, + "Multiple catalog issues agree; choose the correct series and issue." + if options + else "No catalog issue agrees with this file's title, number, type and year.", + ) counts = Counter(str(target["summary"]["provider_id"]) for _, _, target in plans) state = dict(job.progress_snapshot.get("deferred_recovery") or {}) repaired = int(state.get("reference_files_repaired", 0)) @@ -258,6 +238,11 @@ async def repair_catalog_references( if progress is not None: await progress(position, len(plans)) if counts[str(target["summary"]["provider_id"])] != 1: + file = await session.get(ImportedFile, file_id) + if file is not None: + record_catalog_review( + file, identity, "Multiple files claim this issue; choose which copy to keep." + ) continue await raise_if_job_cancelled(session, job.id) file = await session.get(ImportedFile, file_id) @@ -296,15 +281,36 @@ async def repair_catalog_references( series = await session.scalar(select(Series).where(Series.comicvine_id == cv_id)) created_series = series is None issue = await session.scalar(select(Issue).where(Issue.comicvine_id == issue_cv_id)) - if issue is not None and (series is None or issue.series_id != series.id): + reparent_issue = issue is not None and (series is None or issue.series_id != series.id) + if reparent_issue and ( + issue is None + or issue.id != library.issue_id + or issue.series_id != item.series_id + or identity.get("issue_cv_id") != issue.comicvine_id + or identity["evidence"] != "comicinfo" + ): continue if issue is not None and ( issue.effective_issue_number_text != identity["issue_number"] - or issue.issue_type.value != identity["issue_type"] + or issue.issue_type.value != str(target["summary"].get("issue_type") or "issue") ): continue if issue is not None and await session.scalar( - select(LibraryFile.id).where(LibraryFile.issue_id == issue.id) + select(LibraryFile.id).where( + LibraryFile.issue_id == issue.id, LibraryFile.id != library.id + ) + ): + continue + if ( + reparent_issue + and issue is not None + and await session.scalar( + select(ImportedFile.id).where( + ImportedFile.matched_issue_id == issue.id, + ImportedFile.id != file.id, + ImportedFile.status == ImportedFileStatus.IMPORTED, + ) + ) ): continue if await session.scalar( @@ -315,7 +321,7 @@ async def repair_catalog_references( ) ): continue - if series is not None and issue is None: + if series is not None and (issue is None or reparent_issue): existing = list( await session.scalars(select(Issue).where(Issue.series_id == series.id)) ) @@ -341,6 +347,11 @@ async def repair_catalog_references( ) series.monitored = False series.issue_catalog_state = IssueCatalogState.PARTIAL + previous_series_id = issue.series_id if reparent_issue and issue is not None else None + if reparent_issue and issue is not None: + # The exact embedded identity and catalog agree, and this is the + # only registration. Retain the issue ID (including reader state). + issue.series_id = series.id if issue is None: payload = target["summary"] cutoff = payload.get("source_cutoff_at") @@ -373,6 +384,15 @@ async def repair_catalog_references( source_updated_at=file.updated_at.isoformat(), ) affected, _ = await _apply_mixed_folder_resolutions(session, job, resolutions=(resolution,)) + if previous_series_id is not None: + file.diagnostics = { + **file.diagnostics, + "completed_import_cleanup": { + **file.diagnostics["completed_import_cleanup"], + "source_catalog_series_id": previous_series_id, + "issue_identity_preserved": True, + }, + } previous_issue = await session.get(Issue, resolution.source_issue_id) if ( previous_issue is not None diff --git a/tests/unit/test_import_mixed_identity_recovery.py b/tests/unit/test_import_mixed_identity_recovery.py new file mode 100644 index 00000000..80b24fea --- /dev/null +++ b/tests/unit/test_import_mixed_identity_recovery.py @@ -0,0 +1,211 @@ +"""Mixed folders use per-file identities rather than inherited folder labels.""" + +from dataclasses import replace + +import pytest +from sqlalchemy import select + +from pullbox.models.import_job import ImportedFileStatus, ImportedSeries, ImportSourceType +from pullbox.models.issue import Issue +from pullbox.models.series import Series +from pullbox.services.import_deferred_recovery_execution import prepare_deferred_recovery +from tests.unit.test_import_deferred_recovery import add_file +from tests.unit.test_import_reference_recovery import reference_case + + +def embedded_identity(*, number="104", issue_id=7001): + return { + "source_issue_type": "issue", + "metadata_signals": { + "series_name": "comicinfo", + "issue_number": "comicinfo", + "comicvine_issue_id": "comicinfo", + "comicvine_series_id": "mylar3", + }, + "comicvine_series_id": 100, + "source_metadata": { + "comicinfo": { + "series": "Thunderbolts", + "number": number, + "year": 2021, + "web": f"https://comicvine.gamespot.com/issue/4000-{issue_id}/", + } + }, + } + + +@pytest.mark.parametrize("source_type", list(ImportSourceType)) +@pytest.mark.parametrize("existing_wrong_issue", [False, True]) +async def test_embedded_identity_repairs_stale_folder_label( + db_session, tmp_path, source_type, existing_wrong_issue +): + job, file, library, wrong, metadata, path = await reference_case( + db_session, tmp_path, source_type + ) + item = await db_session.get(ImportedSeries, file.import_series_id) + item.files_no_match = 0 + file.parsed_series = item.raw_series_name + file.comicvine_issue_id = 7001 + file.diagnostics = embedded_identity() + if existing_wrong_issue: + wrong.comicvine_id = file.matched_issue_cv_id = 7001 + original_id = library.id + original_bytes = path.read_bytes() + await db_session.commit() + + await prepare_deferred_recovery(db_session, job.id, metadata_service=metadata) + + target = await db_session.scalar(select(Issue).where(Issue.comicvine_id == 7001)) + assert target is not None, "Embedded titles must not disappear behind saved folder titles" + target_series = await db_session.get(Series, target.series_id) + assert target_series.title == "Thunderbolts" + assert library.issue_id == file.matched_issue_id == target.id + assert library.id == file.library_file_id == original_id + assert path.read_bytes() == original_bytes + assert library.file_path == str(path) + assert target_series.path is None + if existing_wrong_issue: + assert target.id == wrong.id, "Preserve the issue row and its reading history" + + +@pytest.mark.parametrize("source_type", list(ImportSourceType)) +@pytest.mark.parametrize("evidence", ["embedded", "embedded_id", "reading_order"]) +async def test_deferred_mixed_file_uses_independent_title( + db_session, tmp_path, source_type, evidence +): + job, file, library, _wrong, metadata, _ = await reference_case( + db_session, tmp_path, source_type + ) + await db_session.delete(library) + file.library_file_id = file.matched_issue_id = file.matched_issue_cv_id = None + file.comicvine_issue_id = None + file.status = ImportedFileStatus.NO_MATCH + file.parsed_series = "Fritzi Ritz" + file.file_name = "042 - Thunderbolts 104 (2021) (converted).cbz" + file.diagnostics = {"metadata_signals": {"issue_number": "release_title"}} + if evidence.startswith("embedded"): + file.file_name = "Thunderbolts 104 (2021) (DR & Quinch-Empire).cbz" + file.diagnostics = embedded_identity() + if evidence == "embedded_id": + file.comicvine_issue_id = 7001 + else: + file.diagnostics["metadata_signals"].pop("comicvine_issue_id") + file.diagnostics["source_metadata"]["comicinfo"].pop("web") + await db_session.commit() + + await prepare_deferred_recovery(db_session, job.id, metadata_service=metadata) + + assert file.status is ImportedFileStatus.CONFIRMED + assert file.matched_issue_cv_id == 7001 + target = await db_session.get(ImportedSeries, file.import_series_id) + assert target.cv_title == "Thunderbolts" + metadata.search_catalog_series.assert_awaited_once_with("Thunderbolts", limit=1000) + + +async def test_catalog_years_are_checked_for_each_file_not_just_the_group(db_session, tmp_path): + job, file, library, _, metadata, _ = await reference_case(db_session, tmp_path) + await db_session.delete(library) + file.library_file_id = file.matched_issue_id = file.matched_issue_cv_id = None + file.status = ImportedFileStatus.NO_MATCH + item = await db_session.get(ImportedSeries, file.import_series_id) + other = await add_file( + db_session, + job, + item, + file_name="Thunderbolts 104 (1990).cbz", + file_path="/comics/mixed/Thunderbolts 104 (1990).cbz", + comicvine_issue_id=None, + parsed_series="Fritzi Ritz", + parsed_year=1990, + diagnostics={"metadata_signals": {"issue_number": "release_title"}}, + ) + await db_session.commit() + + await prepare_deferred_recovery(db_session, job.id, metadata_service=metadata) + + assert file.status is ImportedFileStatus.CONFIRMED + assert other.status is ImportedFileStatus.NO_MATCH + assert "year" in other.diagnostics["mixed_folder_recovery"]["reason"].lower() + + +@pytest.mark.parametrize("conflict", ["number", "embedded_id", "date", "type", "title"]) +async def test_embedded_recovery_retains_conflicting_evidence(db_session, tmp_path, conflict): + job, file, library, wrong, metadata, _ = await reference_case(db_session, tmp_path) + file.parsed_series = "Fritzi Ritz" + file.comicvine_issue_id = 7001 + file.diagnostics = embedded_identity() + if conflict == "number": + file.diagnostics = embedded_identity(number="105") + elif conflict == "embedded_id": + file.diagnostics = embedded_identity(issue_id=7002) + elif conflict == "date": + metadata.get_catalog_issue_summaries_for_series.return_value = [ + replace( + metadata.get_catalog_issue_summaries_for_series.return_value[0], + release_date="1990-01-01", + ) + ] + elif conflict == "title": + file.file_name = "Action Comics 104 (2021).cbz" + else: + metadata.get_catalog_issue_summaries_for_series.return_value = [ + replace( + metadata.get_catalog_issue_summaries_for_series.return_value[0], issue_type="tpb" + ) + ] + await db_session.commit() + + await prepare_deferred_recovery(db_session, job.id, metadata_service=metadata) + + assert library.issue_id == wrong.id + assert await db_session.scalar(select(Series).where(Series.comicvine_id == 700)) is None + + +@pytest.mark.parametrize("shared", ["file", "import", "manual"]) +async def test_wrong_catalog_issue_is_not_reparented_with_other_ownership( + db_session, tmp_path, shared +): + from tests.unit.test_import_deferred_recovery import register + + job, file, library, wrong, metadata, _ = await reference_case(db_session, tmp_path) + item = await db_session.get(ImportedSeries, file.import_series_id) + file.parsed_series = "Fritzi Ritz" + file.comicvine_issue_id = wrong.comicvine_id = file.matched_issue_cv_id = 7001 + file.diagnostics = embedded_identity() + if shared == "manual": + item.user_selected_cv_id = 100 + else: + twin = await add_file( + db_session, + job, + item, + matched_issue_id=wrong.id, + status=ImportedFileStatus.IMPORTED + if shared == "import" + else ImportedFileStatus.SKIPPED, + ) + if shared == "file": + from pullbox.models.library import LibraryRoot + + root = await db_session.get(LibraryRoot, library.library_root_id) + await register(db_session, twin, wrong, root) + await db_session.commit() + + await prepare_deferred_recovery(db_session, job.id, metadata_service=metadata) + + assert wrong.series_id == item.series_id + assert await db_session.scalar(select(Series).where(Series.comicvine_id == 700)) is None + + +async def test_wrong_folder_cannot_offer_a_provisional_issue_for_a_foreign_comic( + db_session, tmp_path +): + from pullbox.services.import_reconcile_helpers import provisional_issue_number_for_file + + _, file, _, _, _, _ = await reference_case(db_session, tmp_path) + item = await db_session.get(ImportedSeries, file.import_series_id) + file.status = ImportedFileStatus.NO_MATCH + file.parsed_series = "Fritzi Ritz" + file.diagnostics = embedded_identity() + + assert provisional_issue_number_for_file(item, file, []) is None From 6c08ffad24d2f26f0cdac4deb4a132a56660bebf Mon Sep 17 00:00:00 2001 From: Adam Hernandez Date: Fri, 18 Sep 2026 18:56:12 -0700 Subject: [PATCH 4/4] fix(utilities): cancel conversion workers and recover interrupted jobs --- docs/development/ARCHITECTURE_OVERVIEW.md | 11 + .../templates/partials/utilities_queue.html | 1 + src/pullbox/utilities/base_executor.py | 1 + src/pullbox/utilities/cancellation.py | 47 ++ .../utilities/executors/archive_subprocess.py | 4 +- .../utilities/executors/file_converter.py | 25 +- .../executors/mass_convert_pipeline.py | 47 +- .../executors/utility_archive_work.py | 82 +++ src/pullbox/utilities/job_queue.py | 151 ++++- .../utilities/job_queue_batch_processing.py | 36 +- .../utilities/job_queue_batch_state.py | 20 +- .../utilities/job_queue_cancellation.py | 64 +++ .../utilities/job_queue_dispatch_start.py | 4 +- .../utilities/job_queue_finalization.py | 8 + .../utilities/job_queue_item_persistence.py | 20 +- src/pullbox/utilities/job_queue_items.py | 1 + src/pullbox/utilities/job_queue_state.py | 3 +- src/pullbox/utilities/router.py | 3 +- src/pullbox/utilities/worker_pool.py | 26 +- tests/utilities/test_base_executor.py | 3 +- tests/utilities/test_cancellation_recovery.py | 522 ++++++++++++++++++ tests/utilities/test_job_queue.py | 5 +- tests/utilities/test_router_branches.py | 2 +- 23 files changed, 1017 insertions(+), 69 deletions(-) create mode 100644 src/pullbox/utilities/cancellation.py create mode 100644 src/pullbox/utilities/executors/utility_archive_work.py create mode 100644 src/pullbox/utilities/job_queue_cancellation.py create mode 100644 tests/utilities/test_cancellation_recovery.py diff --git a/docs/development/ARCHITECTURE_OVERVIEW.md b/docs/development/ARCHITECTURE_OVERVIEW.md index a8c13d9b..e2e9fc06 100644 --- a/docs/development/ARCHITECTURE_OVERVIEW.md +++ b/docs/development/ARCHITECTURE_OVERVIEW.md @@ -583,6 +583,17 @@ effects such as: - File matched handling. - Series added handling. +Utility cancellation is shared by the serial utility queue. Repeated requests +are idempotent, and cancellation takes precedence over a pending pause. No new +batch can be leased after cancellation commits. Busy workers observe a per-pool +signal; conversion and metadata stages on disposable outputs use interruptible +archive subprocesses. Source replacement and its result journal finish before +the queue releases its execution slot. Other non-interruptible item operations +finish at their safe item boundary rather than being killed mid-mutation. +Terminal job state and activity progress are committed together. Startup recovery +finishes abandoned `CANCELLING` jobs as `CANCELLED`, preserving completed results +and rollback records without replaying unfinished files. + Composition helpers make event-bus intent explicit: - `build_domain_event_bus()` returns the shared application event bus. diff --git a/src/pullbox/ui/templates/partials/utilities_queue.html b/src/pullbox/ui/templates/partials/utilities_queue.html index 00ae58b7..f2febe2b 100644 --- a/src/pullbox/ui/templates/partials/utilities_queue.html +++ b/src/pullbox/ui/templates/partials/utilities_queue.html @@ -100,6 +100,7 @@ x-cloak type="button" data-testid="utilities-queue-active-job-cancel" + :disabled="job.state === 'CANCELLING'" @click="controlJob(job.id, 'cancel')" class="utilities-queue-act-btn utilities-queue-act-btn-danger" data-tip="Cancel job" diff --git a/src/pullbox/utilities/base_executor.py b/src/pullbox/utilities/base_executor.py index 16c49d64..7a71a156 100644 --- a/src/pullbox/utilities/base_executor.py +++ b/src/pullbox/utilities/base_executor.py @@ -30,6 +30,7 @@ class ItemResult(enum.StrEnum): COMPLETED = "completed" FAILED = "failed" SKIPPED = "skipped" + CANCELLED = "cancelled" @dataclass diff --git a/src/pullbox/utilities/cancellation.py b/src/pullbox/utilities/cancellation.py new file mode 100644 index 00000000..5ba1007e --- /dev/null +++ b/src/pullbox/utilities/cancellation.py @@ -0,0 +1,47 @@ +"""Cooperative cancellation at safe utility worker boundaries.""" + +from __future__ import annotations + +from contextlib import contextmanager +from contextvars import ContextVar +from pathlib import Path +from typing import TYPE_CHECKING + +from pullbox.core.exceptions import JobCancelledError + +if TYPE_CHECKING: + from collections.abc import Iterator + +_marker: ContextVar[str | None] = ContextVar("utility_cancel_marker", default=None) + + +@contextmanager +def worker_cancellation(marker: str | None) -> Iterator[None]: + """Bind a per-pool signal without sharing mutable context between threads.""" + token = _marker.set(marker) + try: + yield + finally: + _marker.reset(token) + + +def check_cancelled() -> None: + """Raise only at boundaries where no source mutation needs to finish.""" + marker = _marker.get() + if marker is not None and Path(marker).exists(): + raise JobCancelledError("Utility job cancelled") + + +def cancellation_enabled() -> bool: + """Whether this call belongs to a cancellable queue worker.""" + return _marker.get() is not None + + +async def check_cancelled_async() -> None: + """Adapt the worker signal to interruptible archive operations.""" + check_cancelled() + + +def check_archive_progress(_stage: str, _current: int, _total: int, _unit: str) -> None: + """Use archive progress boundaries to stop expensive disposable work.""" + check_cancelled() diff --git a/src/pullbox/utilities/executors/archive_subprocess.py b/src/pullbox/utilities/executors/archive_subprocess.py index 577e8c18..9e137ddb 100644 --- a/src/pullbox/utilities/executors/archive_subprocess.py +++ b/src/pullbox/utilities/executors/archive_subprocess.py @@ -56,10 +56,12 @@ async def convert_file_interruptible( pdf_quality: str = "medium", progress_callback: ProgressCallback | None = None, allow_resource_safety_exception: bool = False, + output_path: Path | None = None, ) -> Path: """Convert an archive in a child process that can be terminated on cancel.""" dest_dir = destination or source.parent - target_path = dest_dir / f"{source.stem}.{target_format}" + target_path = output_path or dest_dir / f"{source.stem}.{target_format}" + dest_dir = target_path.parent progress_state_path = _create_progress_state_path(dest_dir) payload = { "source": str(source), diff --git a/src/pullbox/utilities/executors/file_converter.py b/src/pullbox/utilities/executors/file_converter.py index a40acfdb..be60023a 100644 --- a/src/pullbox/utilities/executors/file_converter.py +++ b/src/pullbox/utilities/executors/file_converter.py @@ -687,8 +687,13 @@ def process_item( item_id = item_data.get("id", "unknown") file_path = item_data.get("file_path", "") source = Path(file_path) + disposable_output: Path | None = None try: + from pullbox.utilities.cancellation import check_cancelled + from pullbox.utilities.executors.utility_archive_work import convert_utility_file + + check_cancelled() referenced_paths = (job_context or {}).get("referenced_paths", []) resolved_source = source.expanduser().resolve(strict=False) if isinstance(referenced_paths, list) and any( @@ -722,17 +727,22 @@ def process_item( is_repack = source_format == target_format or target_path == source if is_repack: temp_target = source.with_name(f"{source.stem}._repack_.{target_format}") - result_path = _convert_sync( + if temp_target.exists(): + raise FileExistsError(f"Temporary target already exists: {temp_target}") + disposable_output = temp_target + result_path = convert_utility_file( source, target_format, temp_target, pdf_quality=pdf_quality ) elif target_path.exists(): raise FileExistsError(f"Target already exists: {target_path}") else: - result_path = _convert_sync( + disposable_output = target_path + result_path = convert_utility_file( source, target_format, target_path, pdf_quality=pdf_quality ) # Move original to trash + check_cancelled() original_trash_path: str | None = None trash_dir = resolve_trash_directory(job_config.get("trash_folder")) if trash_dir: @@ -776,6 +786,17 @@ def process_item( ) except Exception as exc: duration_ms = int((time.monotonic() - start) * 1000) + from pullbox.core.exceptions import JobCancelledError + + if isinstance(exc, JobCancelledError): + if disposable_output is not None: + disposable_output.unlink(missing_ok=True) + return ProcessedItem( + item_id=item_id, + result=ItemResult.CANCELLED, + duration_ms=duration_ms, + log_entries=[("INFO", "Conversion cancelled; original retained.", {})], + ) return ProcessedItem( item_id=item_id, result=ItemResult.FAILED, diff --git a/src/pullbox/utilities/executors/mass_convert_pipeline.py b/src/pullbox/utilities/executors/mass_convert_pipeline.py index 1f788b26..71ebee35 100644 --- a/src/pullbox/utilities/executors/mass_convert_pipeline.py +++ b/src/pullbox/utilities/executors/mass_convert_pipeline.py @@ -16,7 +16,6 @@ import os import tempfile import time -import zipfile from contextlib import asynccontextmanager from datetime import date from pathlib import Path @@ -26,6 +25,7 @@ from sqlalchemy import select from sqlalchemy.orm import joinedload +from pullbox.core.exceptions import JobCancelledError from pullbox.core.file_safety import classify_resource_safety_exception from pullbox.core.filesystem_scan import iter_supported_files from pullbox.core.issue_numbers import format_issue_number @@ -42,6 +42,12 @@ ProcessedItem, RuntimeLogEntry, ) +from pullbox.utilities.cancellation import check_cancelled +from pullbox.utilities.executors.utility_archive_work import ( + convert_utility_file, + embed_utility_metadata, + verify_utility_archive, +) from pullbox.utilities.settings import ( move_file_to_utility_trash, resolve_utility_directory, @@ -384,6 +390,7 @@ def process_item( created_paths: list[Path] = [] try: + check_cancelled() if item_data.get( "storage_mode" ) == LibraryFileStorageMode.REFERENCED.value or _path_is_referenced( @@ -414,10 +421,6 @@ def process_item( # ── Step 1: Convert to CBZ ───────────────────────── if 1 in steps: - from pullbox.utilities.executors.file_converter import ( - _convert_sync, - ) - log_entries.append( ( "DEBUG", @@ -434,9 +437,11 @@ def process_item( if source.suffix.lower() == ".cbz": temp_target = source.with_name(f"{source.stem}._mass_convert_.cbz") - _convert_sync(source, "cbz", temp_target) - current_path = temp_target + if temp_target.exists(): + raise FileExistsError(f"Temporary target already exists: {temp_target}") created_paths.append(temp_target) + convert_utility_file(source, "cbz", temp_target) + current_path = temp_target log_entries.append( ( "INFO", @@ -449,9 +454,9 @@ def process_item( ) ) else: - _convert_sync(source, "cbz", target_path) - current_path = target_path created_paths.append(target_path) + convert_utility_file(source, "cbz", target_path) + current_path = target_path log_entries.append( ( "INFO", @@ -465,6 +470,7 @@ def process_item( ) # ── Step 2: Embed ComicInfo.xml ──────────────────── + check_cancelled() if 2 in steps and current_path.suffix.lower() == ".cbz": metadata = item_data.get("metadata", {}) metadata_source = str( @@ -500,9 +506,7 @@ def process_item( }, ) ) - from pullbox.utilities.comicinfo import embed_comicinfo_in_cbz - - embed_comicinfo_in_cbz(current_path, metadata) + embed_utility_metadata(current_path, metadata) log_entries.append( ( "INFO", @@ -530,6 +534,7 @@ def process_item( ) # ── Step 4: Verify integrity ─────────────────────── + check_cancelled() if 4 in steps and current_path.suffix.lower() == ".cbz": log_entries.append( ( @@ -542,11 +547,7 @@ def process_item( }, ) ) - with zipfile.ZipFile(current_path, "r") as zf: - bad = zf.testzip() - if bad is not None: - raise ValueError(f"Integrity check failed: corrupt entry '{bad}'") - file_count = len(zf.namelist()) + file_count = verify_utility_archive(current_path) log_entries.append( ( "INFO", @@ -560,6 +561,8 @@ def process_item( ) # ── Move original to trash ───────────────────────── + # Once source replacement starts, finish and return its rollback journal. + check_cancelled() original_trash_path: str | None = None trash_dir = _resolve_effective_trash_directory(job_config.get("trash_folder")) trash_dest = move_file_to_utility_trash( @@ -615,6 +618,16 @@ def process_item( ) ) duration_ms = int((time.monotonic() - start) * 1000) + if isinstance(exc, JobCancelledError): + return ProcessedItem( + item_id=item_id, + result=ItemResult.CANCELLED, + duration_ms=duration_ms, + log_entries=[ + *log_entries, + ("INFO", "Conversion cancelled; original retained.", {}), + ], + ) if resource_block is not None: safety_payload = resource_block.to_diagnostics() log_entries.append( diff --git a/src/pullbox/utilities/executors/utility_archive_work.py b/src/pullbox/utilities/executors/utility_archive_work.py new file mode 100644 index 00000000..c96c2116 --- /dev/null +++ b/src/pullbox/utilities/executors/utility_archive_work.py @@ -0,0 +1,82 @@ +"""Interruptible disposable archive stages; never interrupt source publication.""" + +from __future__ import annotations + +import asyncio +import zipfile +from typing import TYPE_CHECKING, Any + +from pullbox.utilities.cancellation import ( + cancellation_enabled, + check_archive_progress, + check_cancelled, + check_cancelled_async, +) + +if TYPE_CHECKING: + from pathlib import Path + + +def convert_utility_file( + source: Path, + target_format: str, + target: Path, + *, + pdf_quality: str = "medium", +) -> Path: + """Convert only into disposable output, using a killable child when queued.""" + from pullbox.utilities.executors.archive_subprocess import convert_file_interruptible + from pullbox.utilities.executors.file_converter import _convert_sync + + check_cancelled() + if cancellation_enabled(): + return asyncio.run( + convert_file_interruptible( + source, + target_format, + output_path=target, + pdf_quality=pdf_quality, + cancellation_check=check_cancelled_async, + ) + ) + return _convert_sync( + source, + target_format, + target, + pdf_quality=pdf_quality, + progress_callback=check_archive_progress, + ) + + +def embed_utility_metadata(target: Path, metadata: dict[str, Any]) -> None: + """Rewrite a converted disposable output, not a live source archive.""" + from pullbox.utilities.comicinfo import embed_comicinfo_in_cbz + from pullbox.utilities.executors.archive_subprocess import embed_comicinfo_in_cbz_interruptible + + check_cancelled() + if cancellation_enabled(): + asyncio.run( + embed_comicinfo_in_cbz_interruptible( + target, + metadata, + cancellation_check=check_cancelled_async, + ) + ) + else: + embed_comicinfo_in_cbz(target, metadata, progress_callback=check_archive_progress) + + +def verify_utility_archive(target: Path) -> int: + """Validate CRCs with bounded reads so cancellation need not wait for a whole archive.""" + with zipfile.ZipFile(target, "r") as archive: + if cancellation_enabled(): + for entry in archive.infolist(): + check_cancelled() + with archive.open(entry) as stream: + while stream.read(1024 * 1024): + check_cancelled() + else: + bad = archive.testzip() + if bad is not None: + raise ValueError(f"Integrity check failed: corrupt entry '{bad}'") + return len(archive.namelist()) diff --git a/src/pullbox/utilities/job_queue.py b/src/pullbox/utilities/job_queue.py index 5f8a5906..47ce47d0 100644 --- a/src/pullbox/utilities/job_queue.py +++ b/src/pullbox/utilities/job_queue.py @@ -21,6 +21,7 @@ from pullbox.services.utility_operation_progress import project_utility_operation_progress from pullbox.utilities.job_queue_batch_processing import process_dispatch_batches +from pullbox.utilities.job_queue_cancellation import drain_task from pullbox.utilities.job_queue_config import ( get_utility_log_level, get_utility_worker_count, @@ -338,6 +339,8 @@ async def dispatch_next(self) -> None: ) batch_size = worker_runtime.batch_size worker_pool = worker_runtime.worker_pool + interrupted = False + dispatch_error: Exception | None = None try: await process_dispatch_batches( session_factory=self._session_factory, @@ -356,28 +359,41 @@ async def dispatch_next(self) -> None: logger=logger, timestamp_factory=lambda: datetime.now(UTC).isoformat(), ) + except asyncio.CancelledError: + interrupted = True + except Exception as exc: + dispatch_error = exc finally: - worker_pool.shutdown() + await drain_task(asyncio.create_task(asyncio.to_thread(worker_pool.shutdown))) async with self._session_factory() as session: - finalization = await finalize_dispatch_job( - session, - job_id=job_id, - job_type=job_type, - executor=executor, - summary=summary, - config=config, - job_context=job_context, - get_utility_log_level=get_utility_log_level, - persist_log=self._persist_utility_log, - transition_job=self.transition, - ) + if dispatch_error is not None: + await self._record_dispatch_failure(session, job_id, dispatch_error) + continue + try: + finalization = await finalize_dispatch_job( + session, + job_id=job_id, + job_type=job_type, + executor=executor, + summary=summary, + config=config, + job_context=job_context, + get_utility_log_level=get_utility_log_level, + persist_log=self._persist_utility_log, + transition_job=self.transition, + project_progress=project_utility_operation_progress, + ) + except Exception as exc: + await session.rollback() + await self._record_dispatch_failure(session, job_id, exc) + if interrupted: + raise asyncio.CancelledError from exc + continue if finalization.log_event is not None: logger.info(finalization.log_event, **finalization.log_context) - finalized_job = await session.get(UtilityJob, job_id) - if finalized_job is not None: - await project_utility_operation_progress(session, finalized_job) - await session.commit() + if interrupted: + raise asyncio.CancelledError # ── Startup Recovery ─────────────────────────────────────── @@ -385,26 +401,72 @@ async def recover_interrupted_jobs(self, session: AsyncSession) -> int: """Recover jobs interrupted by a server crash. - RUNNING/PAUSING jobs → PAUSED - - IN_PROGRESS items → PENDING + - CANCELLING jobs -> CANCELLED, without replaying unfinished items + - IN_PROGRESS items of interrupted jobs -> PENDING Returns: Number of jobs recovered. """ result = await session.execute( - select(UtilityJob).where(UtilityJob.state.in_([JobState.RUNNING, JobState.PAUSING])) + select(UtilityJob).where( + UtilityJob.state.in_([JobState.RUNNING, JobState.PAUSING, JobState.CANCELLING]) + ) ) interrupted = list(result.scalars().all()) for job in interrupted: - job.state = JobState.PAUSED - job.paused_at = datetime.now(UTC).isoformat() + old_state = job.state + if old_state == JobState.CANCELLING: + uncertain_items = ( + await session.scalars( + select(UtilityJobItem).where( + UtilityJobItem.job_id == job.id, + UtilityJobItem.state == ItemState.IN_PROGRESS, + ) + ) + ).all() + for item in uncertain_items: + self._persist_utility_log( + session, + configured_level="INFO", + job_id=job.id, + item_id=item.id, + file_path=item.file_path, + level="WARNING", + message=( + "Work on this file was interrupted before its result was saved. " + "Inspect the source, output, and trash before retrying; " + "this cancellation will not replay the file." + ), + extra={"previous_state": str(item.state), "started_at": item.started_at}, + ) + job.warning_count = (job.warning_count or 0) + len(uncertain_items) + self.transition(job, JobState.CANCELLED) + self._persist_utility_log( + session, + configured_level="INFO", + job_id=job.id, + level="WARNING", + message=( + "Finished an interrupted cancellation after restart. " + "Completed work and rollback records were retained; " + "unfinished items will not be restarted. Check any file " + "that was in progress when the server stopped." + ), + ) + else: + job.state = JobState.PAUSED + job.paused_at = datetime.now(UTC).isoformat() await project_utility_operation_progress(session, job) - logger.info("job_recovered", job_id=job.id, old_state="RUNNING/PAUSING") + logger.info("job_recovered", job_id=job.id, old_state=old_state, state=job.state) # Reset any IN_PROGRESS items to PENDING await session.execute( update(UtilityJobItem) - .where(UtilityJobItem.state == ItemState.IN_PROGRESS) + .where( + UtilityJobItem.state == ItemState.IN_PROGRESS, + UtilityJobItem.job_id.in_([job.id for job in interrupted]), + ) .values( state=ItemState.PENDING, started_at=None, @@ -420,6 +482,32 @@ async def recover_interrupted_jobs(self, session: AsyncSession) -> int: return len(interrupted) + async def _record_dispatch_failure( + self, + session: AsyncSession, + job_id: str, + exc: Exception, + ) -> None: + """Close a stopped dispatcher without losing already committed item outcomes.""" + job = await session.get(UtilityJob, job_id) + if job is None: + return + if job.state == JobState.CANCELLING: + self.transition(job, JobState.CANCELLED) + elif job.state in {JobState.RUNNING, JobState.PAUSING}: + self.transition(job, JobState.FAILED) + job.error_message = f"Job finalization failed: {exc}" + self._persist_utility_log( + session, + configured_level="INFO", + job_id=job_id, + level="ERROR", + message=job.error_message, + ) + await project_utility_operation_progress(session, job) + await session.commit() + logger.error("job_finalization_failed", job_id=job_id, error=str(exc)) + async def recover_and_dispatch(self) -> int: """Recover interrupted jobs, then restart serial dispatch for queued work.""" async with self._session_factory() as session: @@ -488,13 +576,28 @@ async def cancel_job( raise ValueError(f"Job not found: {job_id}") current = JobState(job.state) + if current in {JobState.CANCELLING, JobState.CANCELLED}: + # Retrying the same request must neither restart work nor duplicate rollback. + if rollback: + existing = await session.scalar( + select(UtilityJob.id) + .where( + UtilityJob.parent_job_id == job_id, + UtilityJob.job_type == JobType.ROLLBACK, + ) + .limit(1) + ) + if existing is None: + await self._create_rollback_job(session, job) + await project_utility_operation_progress(session, job) + return if current not in _CANCELLABLE_STATES: raise ValueError( f"Cannot cancel job in state {current.value}. " f"Cancellable states: {', '.join(s.value for s in _CANCELLABLE_STATES)}" ) - if current == JobState.RUNNING: + if current in {JobState.RUNNING, JobState.PAUSING}: # Running jobs go through CANCELLING first self.transition(job, JobState.CANCELLING) else: diff --git a/src/pullbox/utilities/job_queue_batch_processing.py b/src/pullbox/utilities/job_queue_batch_processing.py index 72d52dde..590641a4 100644 --- a/src/pullbox/utilities/job_queue_batch_processing.py +++ b/src/pullbox/utilities/job_queue_batch_processing.py @@ -2,8 +2,10 @@ from __future__ import annotations +import asyncio from typing import TYPE_CHECKING, Any +from pullbox.utilities.base_executor import ItemResult from pullbox.utilities.job_queue_batch_failure import ( build_batch_dispatch_failure_item, persist_batch_dispatch_failure_item, @@ -13,6 +15,7 @@ lease_dispatch_batch, prepare_batch_checkpoint, ) +from pullbox.utilities.job_queue_cancellation import wait_for_dispatch_batch from pullbox.utilities.job_queue_item_persistence import ( persist_post_commit_logs, persist_processed_item_failure, @@ -134,6 +137,9 @@ async def process_dispatch_batch( summary.warnings += item_persistence.warning_delta seen_item_ids.add(processed.item_id) + if processed.result == ItemResult.CANCELLED: + continue + post_commit_logs = await executor.after_item_commit( payload_data, processed, @@ -230,20 +236,28 @@ async def process_dispatch_batches( if project_progress is not None and batch_items: await project_progress(job_id, batch_items[0].id) - await process_dispatch_batch( + batch_task = asyncio.create_task( + process_dispatch_batch( + session_factory=session_factory, + job_id=job_id, + job_type=job_type, + executor=executor, + config=config, + job_context=job_context, + summary=summary, + utility_log_level=utility_log_level, + batch_items=batch_items, + worker_pool=worker_pool, + persist_log=persist_log, + logger=logger, + timestamp_factory=timestamp_factory, + ) + ) + await wait_for_dispatch_batch( + batch_task, session_factory=session_factory, job_id=job_id, - job_type=job_type, - executor=executor, - config=config, - job_context=job_context, - summary=summary, - utility_log_level=utility_log_level, - batch_items=batch_items, worker_pool=worker_pool, - persist_log=persist_log, - logger=logger, - timestamp_factory=timestamp_factory, ) if project_progress is not None: await project_progress(job_id, None) diff --git a/src/pullbox/utilities/job_queue_batch_state.py b/src/pullbox/utilities/job_queue_batch_state.py index 2cbbb881..5816b93b 100644 --- a/src/pullbox/utilities/job_queue_batch_state.py +++ b/src/pullbox/utilities/job_queue_batch_state.py @@ -5,7 +5,7 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Any -from sqlalchemy import update +from sqlalchemy import select, update from pullbox.utilities.job_queue_state import transition_job_state from pullbox.utilities.models import ItemState, JobState, UtilityJob, UtilityJobItem @@ -63,18 +63,26 @@ async def mark_batch_items_in_progress( *, item_ids: Iterable[str], started_at: str, -) -> None: +) -> set[str]: """Mark a batch of pending job items as leased to workers.""" - await session.execute( + result = await session.execute( update(UtilityJobItem) - .where(UtilityJobItem.id.in_(list(item_ids))) + .where( + UtilityJobItem.id.in_(list(item_ids)), + UtilityJobItem.state == ItemState.PENDING, + UtilityJobItem.job_id.in_( + select(UtilityJob.id).where(UtilityJob.state == JobState.RUNNING) + ), + ) .values( state=ItemState.IN_PROGRESS, started_at=started_at, completed_at=None, worker_id=None, ) + .returning(UtilityJobItem.id) ) + return set(result.scalars()) async def lease_dispatch_batch( @@ -87,10 +95,10 @@ async def lease_dispatch_batch( ) -> list[UtilityJobItem]: """Mark and return the next pending item slice leased to workers.""" batch_items = pending_items[batch_start : batch_start + batch_size] - await mark_batch_items_in_progress( + leased_ids = await mark_batch_items_in_progress( session, item_ids=[item.id for item in batch_items], started_at=started_at, ) await session.commit() - return batch_items + return [item for item in batch_items if item.id in leased_ids] diff --git a/src/pullbox/utilities/job_queue_cancellation.py b/src/pullbox/utilities/job_queue_cancellation.py new file mode 100644 index 00000000..754e7863 --- /dev/null +++ b/src/pullbox/utilities/job_queue_cancellation.py @@ -0,0 +1,64 @@ +"""Observe committed cancellation while retaining ownership of active workers.""" + +from __future__ import annotations + +import asyncio +from typing import Any + +from pullbox.services.utility_operation_progress import project_utility_operation_progress +from pullbox.utilities.job_queue_state import transition_job_state +from pullbox.utilities.models import JobState, UtilityJob + + +async def drain_task(task: asyncio.Task[Any]) -> Any: + """Do not abandon a worker/result journal when the dispatcher is cancelled.""" + while not task.done(): + try: + await asyncio.shield(task) + except asyncio.CancelledError: + if task.cancelled(): + raise + return task.result() + + +async def wait_for_dispatch_batch( + task: asyncio.Task[None], + *, + session_factory: Any, + job_id: str, + worker_pool: Any, +) -> None: + """Signal cooperative workers, then wait for all source mutations and journals.""" + request_cancel = getattr(worker_pool, "request_cancel", lambda: None) + try: + while not task.done(): + done, _ = await asyncio.wait({task}, timeout=0.2) + if done: + break + async with session_factory() as session: + job = await session.get(UtilityJob, job_id) + if job is None or job.state in {JobState.CANCELLING, JobState.CANCELLED}: + request_cancel() + await task + except asyncio.CancelledError: + request_cancel() + + # Persist the interrupted dispatch intent before draining results. A crash + # during the drain will be recovered as cancelled on the next startup. + async def record_and_drain() -> None: + try: + async with session_factory() as session: + job = await session.get(UtilityJob, job_id) + if job is not None and job.state in {JobState.RUNNING, JobState.PAUSING}: + transition_job_state(job, JobState.CANCELLING) + await project_utility_operation_progress(session, job) + await session.commit() + finally: + await drain_task(task) + + await drain_task(asyncio.create_task(record_and_drain())) + raise + except Exception: + request_cancel() + await drain_task(task) + raise diff --git a/src/pullbox/utilities/job_queue_dispatch_start.py b/src/pullbox/utilities/job_queue_dispatch_start.py index ff97ab43..6ffed240 100644 --- a/src/pullbox/utilities/job_queue_dispatch_start.py +++ b/src/pullbox/utilities/job_queue_dispatch_start.py @@ -50,7 +50,9 @@ class DispatchStartResult: async def load_next_dispatch_candidate(session: AsyncSession) -> UtilityJob | None: """Return the next queued job only when no job is already running.""" running = await session.execute( - select(UtilityJob).where(UtilityJob.state == JobState.RUNNING).limit(1) + select(UtilityJob) + .where(UtilityJob.state.in_([JobState.RUNNING, JobState.PAUSING, JobState.CANCELLING])) + .limit(1) ) if running.scalar_one_or_none() is not None: return None diff --git a/src/pullbox/utilities/job_queue_finalization.py b/src/pullbox/utilities/job_queue_finalization.py index 4aba2413..0d0b8319 100644 --- a/src/pullbox/utilities/job_queue_finalization.py +++ b/src/pullbox/utilities/job_queue_finalization.py @@ -55,6 +55,7 @@ async def finalize_dispatch_job( get_utility_log_level: Callable[[Any], Awaitable[str]], persist_log: Callable[..., None], transition_job: Callable[[UtilityJob, JobState], Any], + project_progress: Callable[[Any, UtilityJob], Awaitable[None]] | None = None, ) -> DispatchFinalizationResult: """Persist final counters, logs, executor finalization, and terminal state.""" job = await session.get(UtilityJob, job_id) @@ -72,6 +73,9 @@ async def finalize_dispatch_job( ) current = JobState(job.state) + if current == JobState.PAUSING: + transition_job(job, JobState.PAUSED) + current = JobState.PAUSED if current == JobState.PAUSED: paused_finalization = persist_paused_dispatch_log( session, @@ -80,6 +84,8 @@ async def finalize_dispatch_job( configured_level=utility_log_level, persist_log=persist_log, ) + if project_progress is not None: + await project_progress(session, job) await session.commit() return DispatchFinalizationResult( status="paused", @@ -120,6 +126,8 @@ async def finalize_dispatch_job( persist_log=persist_log, ) + if project_progress is not None: + await project_progress(session, job) await session.commit() return DispatchFinalizationResult( status="completed", diff --git a/src/pullbox/utilities/job_queue_item_persistence.py b/src/pullbox/utilities/job_queue_item_persistence.py index c397db36..b4645772 100644 --- a/src/pullbox/utilities/job_queue_item_persistence.py +++ b/src/pullbox/utilities/job_queue_item_persistence.py @@ -5,6 +5,7 @@ from dataclasses import dataclass from typing import TYPE_CHECKING, Any +from pullbox.utilities.base_executor import ApplyResult, ItemResult from pullbox.utilities.job_queue_processed_result import ( apply_processed_item_snapshot, build_processed_item_counter_delta, @@ -18,7 +19,6 @@ from collections.abc import Callable, Iterable from pullbox.utilities.base_executor import ( - ApplyResult, JobExecutor, JobRunSummary, ProcessedItem, @@ -91,6 +91,24 @@ async def persist_processed_item_result( if item is None: return None + if processed.result == ItemResult.CANCELLED: + # An interrupted disposable stage is not a failed or skipped comic. + # Retain original discovery/rollback evidence and do not apply executor side effects. + item.state = ItemState.PENDING + item.started_at = None + item.completed_at = None + item.worker_id = None + persist_processed_item_log_entries( + session, + processed=processed, + persist_log=persist_log, + configured_level=configured_level, + job_id=job_id, + item_id=item_id, + file_path=file_path, + ) + return ProcessedItemPersistenceResult(ApplyResult(), 0, 0, 0, 0) + apply_processed_item_snapshot( item, processed, diff --git a/src/pullbox/utilities/job_queue_items.py b/src/pullbox/utilities/job_queue_items.py index b9d64471..7123e7e0 100644 --- a/src/pullbox/utilities/job_queue_items.py +++ b/src/pullbox/utilities/job_queue_items.py @@ -31,6 +31,7 @@ def item_result_to_state(result: object) -> ItemState: ItemResult.COMPLETED: ItemState.COMPLETED, ItemResult.FAILED: ItemState.FAILED, ItemResult.SKIPPED: ItemState.SKIPPED, + ItemResult.CANCELLED: ItemState.PENDING, } return result_states.get(result, ItemState.FAILED) diff --git a/src/pullbox/utilities/job_queue_state.py b/src/pullbox/utilities/job_queue_state.py index 408e7d77..ad15ab06 100644 --- a/src/pullbox/utilities/job_queue_state.py +++ b/src/pullbox/utilities/job_queue_state.py @@ -15,7 +15,7 @@ JobState.CANCELLING, JobState.FAILED, }, - JobState.PAUSING: {JobState.PAUSED, JobState.FAILED}, + JobState.PAUSING: {JobState.PAUSED, JobState.CANCELLING, JobState.FAILED}, JobState.PAUSED: {JobState.RUNNING, JobState.CANCELLING}, JobState.CANCELLING: {JobState.CANCELLED, JobState.FAILED}, JobState.CANCELLED: {JobState.ROLLING_BACK}, @@ -39,6 +39,7 @@ JobState.QUEUED, JobState.RUNNING, JobState.PAUSED, + JobState.PAUSING, } ) diff --git a/src/pullbox/utilities/router.py b/src/pullbox/utilities/router.py index c65a9cc2..00890b2d 100644 --- a/src/pullbox/utilities/router.py +++ b/src/pullbox/utilities/router.py @@ -449,7 +449,8 @@ async def cancel_job( await mgr.cancel_job(session, job_id, rollback=rollback) except ValueError as exc: raise ValidationError(str(exc)) from None - return {"status": "cancelling" if rollback else "cancelled", "job_id": job_id} + job = await session.get(UtilityJob, job_id) + return {"status": str(job.state).lower() if job is not None else "cancelled", "job_id": job_id} @router.post("/jobs/{job_id}/rollback", status_code=200) diff --git a/src/pullbox/utilities/worker_pool.py b/src/pullbox/utilities/worker_pool.py index 49cd7a6f..343bdd94 100644 --- a/src/pullbox/utilities/worker_pool.py +++ b/src/pullbox/utilities/worker_pool.py @@ -8,17 +8,21 @@ from __future__ import annotations import asyncio +import tempfile from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor +from pathlib import Path from typing import TYPE_CHECKING, Any import structlog +from pullbox.core.exceptions import JobCancelledError from pullbox.utilities.base_executor import ( ExecutionMode, ItemResult, JobExecutor, ProcessedItem, ) +from pullbox.utilities.cancellation import check_cancelled, worker_cancellation if TYPE_CHECKING: from collections.abc import AsyncIterator @@ -32,10 +36,13 @@ def _execute_in_worker( job_config: dict[str, Any], job_context: dict[str, Any] | None, worker_id: int, + cancellation_marker: str | None = None, ) -> ProcessedItem: """Run process_item in a worker process. Must be a module-level function for pickling.""" try: - result = executor.run_process_item(item_data, job_config, job_context) + with worker_cancellation(cancellation_marker): + check_cancelled() + result = executor.run_process_item(item_data, job_config, job_context) if result is None: return ProcessedItem( item_id=item_data.get("id", "unknown"), @@ -45,6 +52,13 @@ def _execute_in_worker( ) result.worker_id = worker_id return result + except JobCancelledError: + return ProcessedItem( + item_id=item_data.get("id", "unknown"), + result=ItemResult.CANCELLED, + worker_id=worker_id, + log_entries=[("INFO", "Stopped before the next safe operation; source retained.", {})], + ) except Exception as exc: return ProcessedItem( item_id=item_data.get("id", "unknown"), @@ -81,6 +95,9 @@ def __init__( else: # pragma: no cover - defensive fallback raise ValueError(f"Unsupported execution mode: {execution_mode}") self._shutdown = False + # A private filesystem signal is picklable on spawn/fork and shared with threads. + self._control_dir = tempfile.TemporaryDirectory(prefix="pullbox-utility-control-") + self._cancellation_marker = Path(self._control_dir.name) / "cancel" def _ensure_active(self) -> None: """Raise if the pool is no longer available.""" @@ -147,6 +164,7 @@ async def iter_batch_results( job_config, job_context, worker_id, + str(self._cancellation_marker), ) pending.append(asyncio.create_task(self._run_batch_future(idx, item_data, future))) @@ -199,3 +217,9 @@ def shutdown(self) -> None: self._pool.shutdown(wait=True) self._shutdown = True self._pool = None + self._control_dir.cleanup() + + def request_cancel(self) -> None: + """Ask workers to stop before the next safe operation.""" + if not self._shutdown: + self._cancellation_marker.touch(exist_ok=True) diff --git a/tests/utilities/test_base_executor.py b/tests/utilities/test_base_executor.py index 538ae5e6..e42e306a 100644 --- a/tests/utilities/test_base_executor.py +++ b/tests/utilities/test_base_executor.py @@ -32,7 +32,8 @@ def test_values(self) -> None: assert ItemResult.COMPLETED == "completed" assert ItemResult.FAILED == "failed" assert ItemResult.SKIPPED == "skipped" - assert len(ItemResult) == 3 + assert ItemResult.CANCELLED == "cancelled" + assert len(ItemResult) == 4 # ── ProcessedItem Pickling ───────────────────────────────────── diff --git a/tests/utilities/test_cancellation_recovery.py b/tests/utilities/test_cancellation_recovery.py new file mode 100644 index 00000000..843dbe76 --- /dev/null +++ b/tests/utilities/test_cancellation_recovery.py @@ -0,0 +1,522 @@ +"""Regression coverage for abandoned utility cancellations and safe completion.""" + +from __future__ import annotations + +import asyncio +import json +import sys +import time +import zipfile +from contextlib import suppress +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from pullbox.models.base import Base +from pullbox.models.operation_progress import OperationProgress +from pullbox.services.utility_operation_progress import project_utility_operation_progress +from pullbox.utilities.base_executor import ExecutionMode, FinalizeResult +from pullbox.utilities.cancellation import check_cancelled, worker_cancellation +from pullbox.utilities.executors.file_converter import FileConverterExecutor +from pullbox.utilities.executors.mass_convert_pipeline import MassConvertPipelineExecutor +from pullbox.utilities.job_queue import JobQueueManager +from pullbox.utilities.job_queue_batch_state import lease_dispatch_batch +from pullbox.utilities.models import ( + ItemState, + JobState, + JobType, + UtilityJob, + UtilityJobItem, + UtilityJobLog, +) +from pullbox.utilities.worker_pool import WorkerPool +from tests.utilities.conftest import StubExecutor + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + +@pytest.fixture +async def session_factory(tmp_path: Path) -> AsyncIterator[Any]: + # Separate connections/transactions, like production. Cancelling a query can + # discard its connection, which would erase a shared in-memory test database. + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'jobs.db'}") + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + try: + yield async_sessionmaker(engine, expire_on_commit=False) + finally: + await engine.dispose() + + +async def make_job(session: Any, state: JobState = JobState.CANCELLING) -> UtilityJob: + manager = JobQueueManager(None) + job = await manager.create_job(session, JobType.MASS_CONVERT_PIPELINE, "Convert", {}) + job.state = state + await project_utility_operation_progress(session, job) + await session.commit() + return job + + +async def test_restart_finishes_cancel_preserving_completed_work(session_factory: Any) -> None: + async with session_factory() as session: + job = await make_job(session) + job.total_items = 48 + job.completed_items = 36 + for index in range(48): + session.add( + UtilityJobItem( + id=f"item-{index}", + job_id=job.id, + item_index=index, + operation="pipeline", + state=ItemState.COMPLETED if index < 36 else ItemState.PENDING, + before_state=json.dumps({"original": index}), + after_state=json.dumps({"trash": index}) if index < 36 else None, + ) + ) + await session.commit() + before = list((await session.execute(select(UtilityJobItem))).scalars()) + snapshots = [(item.state, item.before_state, item.after_state) for item in before] + manager = JobQueueManager(session_factory) + count = await manager.recover_interrupted_jobs(session) + await session.commit() + assert count == 1 + await session.refresh(job) + assert job.state == JobState.CANCELLED + assert job.completed_at and job.queue_position is None + assert (job.completed_items, job.failed_items, job.skipped_items) == (36, 0, 0) + for item, snapshot in zip(before, snapshots, strict=True): + await session.refresh(item) + assert (item.state, item.before_state, item.after_state) == snapshot + progress = ( + await session.execute( + select(OperationProgress).where(OperationProgress.operation_key == job.id) + ) + ).scalar_one() + assert progress.state == "cancelled" + assert progress.completed_at is not None + assert progress.item_key is None + assert await manager.recover_interrupted_jobs(session) == 0 + + +async def test_cancelled_job_cannot_lease_another_batch(session_factory: Any) -> None: + async with session_factory() as session: + job = await make_job(session) + item = UtilityJobItem( + id="pending", + job_id=job.id, + item_index=0, + state=ItemState.PENDING, + operation="pipeline", + ) + session.add(item) + await session.commit() + leased = await lease_dispatch_batch( + session, + pending_items=[item], + batch_start=0, + batch_size=1, + started_at="now", + ) + assert leased == [] + await session.refresh(item) + assert item.state == ItemState.PENDING + + +async def test_restart_records_uncertain_in_progress_file_before_clearing_lease( + session_factory: Any, +) -> None: + async with session_factory() as session: + job = await make_job(session) + item = UtilityJobItem( + id="interrupted", + job_id=job.id, + item_index=0, + state=ItemState.IN_PROGRESS, + operation="pipeline", + file_path="/comics/example.cbr", + before_state='{"source":"retained"}', + after_state='{"trash":"retained"}', + ) + session.add(item) + await session.commit() + await JobQueueManager(session_factory).recover_interrupted_jobs(session) + await session.commit() + logs = list( + ( + await session.execute(select(UtilityJobLog).where(UtilityJobLog.item_id == item.id)) + ).scalars() + ) + assert len(logs) == 1 and logs[0].level == "WARNING" + assert logs[0].file_path == "/comics/example.cbr" + assert "interrupted" in logs[0].message.lower() + await session.refresh(item) + assert item.state == ItemState.PENDING + assert item.before_state == '{"source":"retained"}' + assert item.after_state == '{"trash":"retained"}' + + +@pytest.mark.parametrize("state", [JobState.CANCELLING, JobState.CANCELLED, JobState.PAUSING]) +async def test_cancel_can_be_repeated_and_overrides_pause( + session_factory: Any, state: JobState +) -> None: + async with session_factory() as session: + job = await make_job(session, state) + manager = JobQueueManager(session_factory) + error = None + try: + await manager.cancel_job(session, job.id) + await manager.cancel_job(session, job.id) + except ValueError as exc: + error = exc + assert error is None, f"Cancel should be idempotent: {error}" + assert job.state == ( + JobState.CANCELLED if state == JobState.CANCELLED else JobState.CANCELLING + ) + + +async def test_repeated_cancel_does_not_duplicate_rollback(session_factory: Any) -> None: + async with session_factory() as session: + job = await make_job(session, JobState.RUNNING) + manager = JobQueueManager(session_factory) + await manager.cancel_job(session, job.id, rollback=True) + error = None + try: + await manager.cancel_job(session, job.id, rollback=True) + except ValueError as exc: + error = exc + assert error is None, f"Repeated cancel must not create or reject another rollback: {error}" + children = list( + ( + await session.execute(select(UtilityJob).where(UtilityJob.parent_job_id == job.id)) + ).scalars() + ) + assert len(children) == 1 + + +class BrokenFinalizer(StubExecutor): + execution_mode = ExecutionMode.THREAD + + async def finalize_job(self, session: Any, job: Any, *args: Any) -> FinalizeResult: + raise RuntimeError("finalizer unavailable") + + +async def test_finalizer_failure_does_not_leave_a_running_job(session_factory: Any) -> None: + manager = JobQueueManager(session_factory) + manager.register_executor(JobType.FILE_CONVERT, BrokenFinalizer) + async with session_factory() as session: + job = await manager.create_job(session, JobType.FILE_CONVERT, "Test", {"count": 1}) + await session.commit() + job_id = job.id + error = None + try: + await manager.dispatch_next() + except RuntimeError as exc: + error = exc + assert error is None, f"Finalization failure must be durably recorded: {error}" + async with session_factory() as session: + job = await session.get(UtilityJob, job_id) + assert job is not None and job.state == JobState.FAILED + assert job.completed_items == 1 + assert "finalizer unavailable" in job.error_message + progress = ( + await session.execute( + select(OperationProgress).where(OperationProgress.operation_key == job_id) + ) + ).scalar_one() + assert progress.state == "failed" + + +async def test_cancelling_job_survives_finalizer_error( + session_factory: Any, + monkeypatch: Any, +) -> None: + manager = JobQueueManager(session_factory) + manager.register_executor(JobType.FILE_CONVERT, BrokenFinalizer) + project = manager._project_dispatch_progress + + async def cancel_after_batch(job_id: str, item_id: str | None = None) -> None: + await project(job_id, item_id) + if item_id is None: + async with session_factory() as session: + await manager.cancel_job(session, job_id) + await session.commit() + + monkeypatch.setattr(manager, "_project_dispatch_progress", cancel_after_batch) + async with session_factory() as session: + job = await manager.create_job(session, JobType.FILE_CONVERT, "Test", {"count": 1}) + await session.commit() + job_id = job.id + await manager.dispatch_next() + async with session_factory() as session: + job = await session.get(UtilityJob, job_id) + assert job.state == JobState.CANCELLED + assert job.completed_items == 1 and job.failed_items == 0 + assert "finalizer unavailable" in job.error_message + progress = ( + await session.execute( + select(OperationProgress).where(OperationProgress.operation_key == job_id) + ) + ).scalar_one() + assert progress.state == "cancelled" + + +async def test_long_archive_child_is_stopped_and_partial_output_removed( + tmp_path: Path, + monkeypatch: Any, +) -> None: + from pullbox.utilities.executors import archive_subprocess + + source = tmp_path / "comic.cbz" + source.write_bytes(b"original comic") + marker = tmp_path / "cancel" + partial = tmp_path / "comic._mass_convert_.cbz" + processes = [] + spawn = asyncio.create_subprocess_exec + + async def slow_archive(*args: Any, **kwargs: Any) -> Any: + process = await spawn( + sys.executable, + "-c", + "import pathlib,time,sys; pathlib.Path(sys.argv[1]).write_bytes(b'partial'); " + "time.sleep(30)", + str(partial), + **kwargs, + ) + processes.append(process) + return process + + monkeypatch.setattr(archive_subprocess.asyncio, "create_subprocess_exec", slow_archive) + with worker_cancellation(str(marker)): + task = asyncio.create_task( + asyncio.to_thread( + MassConvertPipelineExecutor().process_item, + {"id": "item", "file_path": str(source)}, + {"steps": [1], "trash_folder": str(tmp_path / "trash")}, + ) + ) + try: + async with asyncio.timeout(5): + while not partial.exists(): + await asyncio.sleep(0.01) + marker.touch() + result = await asyncio.wait_for(asyncio.shield(task), 5) + finally: + marker.touch() + await task + assert str(result.result) == "cancelled" + assert len(processes) == 1 and processes[0].returncode is not None + assert source.read_bytes() == b"original comic" + assert not partial.exists() + assert not (tmp_path / "trash").exists() + + +@pytest.mark.parametrize("state", [JobState.CANCELLING, JobState.PAUSING]) +async def test_draining_worker_keeps_serial_queue_slot( + session_factory: Any, state: JobState +) -> None: + manager = JobQueueManager(session_factory) + manager.register_executor(JobType.FILE_CONVERT, BrokenFinalizer) + async with session_factory() as session: + await make_job(session, state) + queued = await manager.create_job(session, JobType.FILE_CONVERT, "Next", {"count": 0}) + await session.commit() + job_id = queued.id + with suppress(RuntimeError): + await manager.dispatch_next() + async with session_factory() as session: + queued = await session.get(UtilityJob, job_id) + assert queued is not None and queued.state == JobState.QUEUED + + +class CooperativeExecutor(StubExecutor): + execution_mode = ExecutionMode.THREAD + + def process_item(self, item_data: Any, config: Any) -> Any: + Path(config["started"]).touch() + for _ in range(400): + check_cancelled() + time.sleep(0.01) + return super().process_item(item_data, config) + + +def test_cancel_during_source_replacement_finishes_and_retains_rollback_journal( + tmp_path: Path, + monkeypatch: Any, +) -> None: + from pullbox.utilities.executors import file_converter, mass_convert_pipeline + + source = tmp_path / "comic.cbz" + with zipfile.ZipFile(source, "w") as archive: + archive.writestr("page.jpg", b"page") + original = source.read_bytes() + marker = tmp_path / "cancel" + trash = tmp_path / "trash" + move = mass_convert_pipeline.move_file_to_utility_trash + + def cancel_after_move(*args: Any, **kwargs: Any) -> Path: + path = move(*args, **kwargs) + marker.touch() + return path + + monkeypatch.setattr(mass_convert_pipeline, "convert_utility_file", file_converter._convert_sync) + monkeypatch.setattr( + mass_convert_pipeline, "_resolve_effective_trash_directory", lambda _: trash + ) + monkeypatch.setattr(mass_convert_pipeline, "move_file_to_utility_trash", cancel_after_move) + with worker_cancellation(str(marker)): + result = MassConvertPipelineExecutor().process_item( + {"id": "item", "file_path": str(source)}, + {"steps": [1]}, + ) + assert str(result.result) == "completed" + assert marker.exists() + assert Path(result.after_state["original_path"]).read_bytes() == original + assert Path(result.after_state["path"]) == source and source.is_file() + + +@pytest.mark.parametrize("mode", [ExecutionMode.THREAD, ExecutionMode.PROCESS]) +async def test_worker_cancel_stops_cooperative_work(tmp_path: Path, mode: ExecutionMode) -> None: + pool = WorkerPool(execution_mode=mode, max_workers=1) + started = tmp_path / "started" + task = asyncio.create_task( + pool.process_batch( + [{"id": "first"}, {"id": "second"}], CooperativeExecutor(), {"started": str(started)} + ) + ) + try: + async with asyncio.timeout(10): + while not started.exists(): + await asyncio.sleep(0.01) + pool.request_cancel() + results = await asyncio.wait_for(asyncio.shield(task), 2) + assert [str(result.result) for result in results] == ["cancelled", "cancelled"] + finally: + await task + await asyncio.to_thread(pool.shutdown) + + +@pytest.mark.parametrize("executor_type", [MassConvertPipelineExecutor, FileConverterExecutor]) +def test_conversion_cancel_before_publish_keeps_original( + tmp_path: Path, + monkeypatch: Any, + executor_type: Any, +) -> None: + source = tmp_path / "comic.cbz" + with zipfile.ZipFile(source, "w") as archive: + archive.writestr("page.jpg", b"page") + original = source.read_bytes() + marker = tmp_path / "cancel" + from pullbox.utilities.executors import file_converter, mass_convert_pipeline + + convert = file_converter._convert_sync + + def cancel_after_convert(*args: Any, **kwargs: Any) -> Any: + result = convert(*args, **kwargs) + marker.touch() + return result + + monkeypatch.setattr(mass_convert_pipeline, "convert_utility_file", cancel_after_convert) + from pullbox.utilities.executors import utility_archive_work + + monkeypatch.setattr(utility_archive_work, "convert_utility_file", cancel_after_convert) + with worker_cancellation(str(marker)): + result = executor_type().process_item( + {"id": "item", "file_path": str(source)}, + {"steps": [1], "trash_folder": str(tmp_path / "trash")}, + ) + assert str(result.result) == "cancelled" + assert source.read_bytes() == original + assert not (tmp_path / "comic._mass_convert_.cbz").exists() + assert not (tmp_path / "comic._repack_.cbz").exists() + assert not (tmp_path / "trash").exists() + + +async def test_cancel_interrupts_active_dispatch_without_counting_failures( + session_factory: Any, + tmp_path: Path, +) -> None: + manager = JobQueueManager(session_factory) + manager.register_executor(JobType.FILE_CONVERT, CooperativeExecutor) + started = tmp_path / "started" + async with session_factory() as session: + job = await manager.create_job( + session, + JobType.FILE_CONVERT, + "Test", + { + "count": 2, + "started": str(started), + }, + ) + await session.commit() + job_id = job.id + task = asyncio.create_task(manager.dispatch_next()) + try: + async with asyncio.timeout(10): + while not started.exists(): + await asyncio.sleep(0.01) + async with session_factory() as session: + await manager.cancel_job(session, job_id) + await session.commit() + await asyncio.wait_for(asyncio.shield(task), 2) + finally: + await task + async with session_factory() as session: + job = await session.get(UtilityJob, job_id) + assert job is not None and job.state == JobState.CANCELLED + assert (job.completed_items, job.failed_items, job.skipped_items) == (0, 0, 0) + items = list( + ( + await session.execute(select(UtilityJobItem).where(UtilityJobItem.job_id == job_id)) + ).scalars() + ) + assert len(items) == 2 and all(item.state == ItemState.PENDING for item in items) + + +async def test_dispatch_task_cancellation_drains_workers_and_finalizes( + session_factory: Any, + tmp_path: Path, +) -> None: + manager = JobQueueManager(session_factory) + manager.register_executor(JobType.FILE_CONVERT, CooperativeExecutor) + started = tmp_path / "started" + async with session_factory() as session: + job = await manager.create_job( + session, + JobType.FILE_CONVERT, + "Test", + { + "count": 1, + "started": str(started), + }, + ) + await session.commit() + job_id = job.id + task = asyncio.create_task(manager.dispatch_next()) + async with asyncio.timeout(10): + while not started.exists(): + await asyncio.sleep(0.01) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + async with session_factory() as session: + job = await session.get(UtilityJob, job_id) + assert job is not None and job.state == JobState.CANCELLED + assert job.completed_items == 0 + items = list( + ( + await session.execute(select(UtilityJobItem).where(UtilityJobItem.job_id == job_id)) + ).scalars() + ) + assert all(item.state == ItemState.PENDING for item in items) + progress = ( + await session.execute( + select(OperationProgress).where(OperationProgress.operation_key == job_id) + ) + ).scalar_one() + assert progress.state == "cancelled" diff --git a/tests/utilities/test_job_queue.py b/tests/utilities/test_job_queue.py index 47f860e0..84030420 100644 --- a/tests/utilities/test_job_queue.py +++ b/tests/utilities/test_job_queue.py @@ -1274,6 +1274,7 @@ async def test_dispatch_transitions_pausing_job_to_paused( # Wait for at least one item to start processing await asyncio.sleep(0.3) await mgr.pause_job(db_session, job.id) + await db_session.commit() await dispatch_task await db_session.refresh(job) @@ -1550,6 +1551,7 @@ async def test_dispatch_drains_into_next_queued_job_after_pause( await asyncio.sleep(0.3) await mgr.pause_job(db_session, first.id) + await db_session.commit() await dispatch_task await db_session.refresh(first) @@ -1593,6 +1595,7 @@ async def test_dispatch_drains_into_next_queued_job_after_cancel( await asyncio.sleep(0.3) await mgr.cancel_job(db_session, first.id) + await db_session.commit() await dispatch_task await db_session.refresh(first) @@ -1751,7 +1754,7 @@ async def test_after_item_commit_receives_matching_item_payload( await db_session.refresh(job) assert job.state == JobState.COMPLETED - assert AfterCommitPayloadExecutor.seen_after_commit_paths == [ + assert sorted(AfterCommitPayloadExecutor.seen_after_commit_paths) == [ "/comics/first.cbz", "/comics/second.cbz", ] diff --git a/tests/utilities/test_router_branches.py b/tests/utilities/test_router_branches.py index 5b81868e..b60228e6 100644 --- a/tests/utilities/test_router_branches.py +++ b/tests/utilities/test_router_branches.py @@ -355,7 +355,7 @@ async def test_job_control_routes_success_and_validation( "job_id": "job-1", } assert await utilities_router.cancel_job("job-2", _user(), db_session, rollback=True) == { - "status": "cancelling", + "status": "cancelled", "job_id": "job-2", } assert await utilities_router.rollback_job("job-1", _user(), db_session) == {