Add startup cleanup for dangling documents

detection_service/main.py

@@ -88,7 +88,12 @@ def publish_new_files(channel, queue_name, grouped_files):
 
 
 def main():
+    # Imported lazily so that importing this module (e.g. in tests) does not
+    # trigger the database initialisation performed by scansynclib.sqlite_wrapper.
+    from scansynclib.cleanup import cleanup_dangling_documents
+
     ensure_scan_directory_exists(SCAN_DIR)
+    cleanup_dangling_documents()
     connection, channel = setup_rabbitmq_connection(RABBITQUEUE)
 
     logger.info(f"Scanning {SCAN_DIR} for new files...")

scansynclib/scansynclib/cleanup.py

@@ -0,0 +1,99 @@
+import os
+from scansynclib.logging import logger
+from scansynclib.config import config
+from scansynclib.ProcessItem import ProcessStatus, StatusProgressBar
+from scansynclib.sqlite_wrapper import execute_query
+
+
+def _move_leftover_to_failed(file_name: str, local_dir: str):
+    """Move a leftover source file into the failed directory and drop its OCR file.
+
+    The original (non OCR) scan is moved into the failed directory so it stays
+    recoverable from the web UI, while any leftover ``*_OCR.pdf`` working file is
+    removed. Missing files are tolerated and only logged.
+
+    Args:
+        file_name (str): The base file name of the scan (as stored in the DB).
+        local_dir (str): The SMB share sub directory the scan lives in.
+    """
+    if not file_name:
+        logger.warning("Cannot move leftover file to failed directory: missing file name.")
+        return
+
+    smb_path = config.get("smb.path")
+    failed_dir = os.path.join(smb_path, config.get("failedDir"))
+
+    if not os.path.exists(failed_dir):
+        try:
+            os.makedirs(failed_dir)
+        except Exception:
+            logger.exception(f"Failed to create failed directory {failed_dir}")
+            return
+
+    if local_dir:
+        source_path = os.path.join(smb_path, local_dir, file_name)
+        if os.path.exists(source_path):
+            try:
+                dest_path = os.path.join(failed_dir, file_name)
+                if os.path.exists(dest_path):
+                    base, ext = os.path.splitext(file_name)
+                    dest_path = os.path.join(failed_dir, f"{base}_{local_dir}{ext}")
+                os.rename(source_path, dest_path)
+                logger.info(f"Moved leftover file {source_path} to failed directory {dest_path}")
+            except Exception:
+                logger.exception(f"Failed to move leftover file {source_path} to failed directory {failed_dir}")
+        else:
+            logger.debug(f"Leftover file {source_path} not found, nothing to move.")
+
+        # Remove leftover OCR working file if present
+        ocr_file = os.path.join(smb_path, local_dir, os.path.splitext(file_name)[0] + "_OCR.pdf")
+        if os.path.exists(ocr_file):
+            try:
+                os.remove(ocr_file)
+                logger.info(f"Removed leftover OCR file {ocr_file}")
+            except Exception:
+                logger.exception(f"Failed to remove leftover OCR file {ocr_file}")
+    else:
+        logger.warning(f"No source directory recorded for {file_name}, cannot move it to failed directory.")
+
+
+def cleanup_dangling_documents():
+    """Mark documents left in an in-progress state as failed on startup.
+
+    When the services (re)start after a crash or redeploy, any document that was
+    still being processed is stuck in a pending/processing state and will never
+    finish on its own. To make sure a scan is never silently lost, such documents
+    are marked as failed and their leftover source files are moved to the failed
+    directory so they remain recoverable from the web UI.
+    """
+    pending = execute_query(
+        "SELECT id, file_name, local_filepath FROM scanneddata WHERE status_code BETWEEN 0 AND 4",
+        fetchall=True,
+    )
+    if pending is None:
+        logger.error("Startup cleanup: failed querying dangling documents (database query returned None).")
+        return
+    pending = pending or []
+
+    if not pending:
+        logger.info("Startup cleanup: no dangling documents found.")
+        return
+
+    logger.warning(f"Startup cleanup: found {len(pending)} dangling document(s), marking as failed.")
+    for row in pending:
+        doc_id = row.get("id")
+        file_name = row.get("file_name")
+        local_dir = row.get("local_filepath")
+        try:
+            _move_leftover_to_failed(file_name, local_dir)
+        except Exception:
+            logger.exception(f"Failed moving leftover file for document id {doc_id} to failed directory.")
+
+        try:
+            execute_query(
+                "UPDATE scanneddata SET file_status = ?, status_code = ?, modified = DATETIME('now', 'localtime') WHERE id = ?",
+                (ProcessStatus.FAILED.value, StatusProgressBar.get_progress(ProcessStatus.FAILED), doc_id),
+            )
+            logger.info(f"Startup cleanup: marked document id {doc_id} ({file_name}) as failed.")
+        except Exception:
+            logger.exception(f"Failed marking document id {doc_id} as failed during startup cleanup.")

tests/test_cleanup.py

@@ -0,0 +1,114 @@
+import sys
+import types
+
+import pytest
+
+
+# Importing scansynclib.cleanup pulls in scansynclib.sqlite_wrapper, which
+# initializes a real SQLite database at import time. That database is not
+# available in the unit test environment, so replace the module with a stub.
+# The query helper is mocked per-test anyway.
+_sqlite_stub = types.ModuleType("scansynclib.sqlite_wrapper")
+_sqlite_stub.execute_query = lambda *args, **kwargs: None
+_sqlite_stub.update_scanneddata_database = lambda *args, **kwargs: None
+
+_original_modules = {
+    "scansynclib.sqlite_wrapper": sys.modules.get("scansynclib.sqlite_wrapper"),
+}
+sys.modules["scansynclib.sqlite_wrapper"] = _sqlite_stub
+
+
+def teardown_module(module):
+    for name, original in _original_modules.items():
+        if original is None:
+            sys.modules.pop(name, None)
+        else:
+            sys.modules[name] = original
+
+
+import scansynclib.cleanup as cleanup  # noqa: E402
+from scansynclib.ProcessItem import ProcessStatus, StatusProgressBar  # noqa: E402
+
+
+@pytest.fixture
+def smb(tmp_path, mocker):
+    """Point the cleanup module at a temporary SMB directory."""
+    smb_path = tmp_path / "scans"
+    (smb_path / "ShareA").mkdir(parents=True)
+
+    def fake_get(key, default=None):
+        return {
+            "smb.path": str(smb_path),
+            "failedDir": "failed-documents",
+        }.get(key, default)
+
+    mocker.patch.object(cleanup.config, "get", side_effect=fake_get)
+    return smb_path
+
+
+def test_cleanup_marks_pending_documents_as_failed(smb, mocker):
+    scan = smb / "ShareA" / "scan.pdf"
+    scan.write_bytes(b"%PDF-1.4 test")
+
+    execute_query = mocker.patch.object(
+        cleanup,
+        "execute_query",
+        return_value=[{"id": 7, "file_name": "scan.pdf", "local_filepath": "ShareA"}],
+    )
+
+    cleanup.cleanup_dangling_documents()
+
+    # The pending document is moved out of the share and into the failed dir.
+    assert not scan.exists()
+    assert (smb / "failed-documents" / "scan.pdf").is_file()
+
+    update_calls = [c for c in execute_query.call_args_list if "UPDATE scanneddata" in c.args[0]]
+    assert len(update_calls) == 1
+    status_value, status_code, doc_id = update_calls[0].args[1]
+    assert status_value == ProcessStatus.FAILED.value
+    assert status_code == StatusProgressBar.get_progress(ProcessStatus.FAILED)
+    assert status_code < 0
+    assert doc_id == 7
+
+
+def test_cleanup_removes_leftover_ocr_file(smb, mocker):
+    scan = smb / "ShareA" / "scan.pdf"
+    scan.write_bytes(b"%PDF-1.4 test")
+    ocr_file = smb / "ShareA" / "scan_OCR.pdf"
+    ocr_file.write_bytes(b"%PDF-1.4 ocr")
+
+    mocker.patch.object(
+        cleanup,
+        "execute_query",
+        return_value=[{"id": 1, "file_name": "scan.pdf", "local_filepath": "ShareA"}],
+    )
+
+    cleanup.cleanup_dangling_documents()
+
+    # The original file is preserved in the failed dir, the OCR file is dropped.
+    assert (smb / "failed-documents" / "scan.pdf").is_file()
+    assert not ocr_file.exists()
+    assert not (smb / "failed-documents" / "scan_OCR.pdf").exists()
+
+
+def test_cleanup_marks_failed_even_when_file_missing(smb, mocker):
+    execute_query = mocker.patch.object(
+        cleanup,
+        "execute_query",
+        return_value=[{"id": 3, "file_name": "gone.pdf", "local_filepath": "ShareA"}],
+    )
+
+    cleanup.cleanup_dangling_documents()
+
+    update_calls = [c for c in execute_query.call_args_list if "UPDATE scanneddata" in c.args[0]]
+    assert len(update_calls) == 1
+    assert update_calls[0].args[1][2] == 3
+
+
+def test_cleanup_noop_when_no_pending_documents(smb, mocker):
+    execute_query = mocker.patch.object(cleanup, "execute_query", return_value=[])
+
+    cleanup.cleanup_dangling_documents()
+
+    update_calls = [c for c in execute_query.call_args_list if "UPDATE scanneddata" in c.args[0]]
+    assert update_calls == []