From e1f128659e65cafca9da489d1cdc4e4dd0bd2790 Mon Sep 17 00:00:00 2001 From: Robert Helewka Date: Mon, 3 Aug 2026 12:27:24 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=BE=20fix(tasks):=20unsupported=20file?= =?UTF-8?q?=20types=20fail=20ingest=20terminally,=20no=20retries?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Unsupported file types raised a bare ValueError from the parser, which ingest_from_daedalus treated like any transient fault — ERROR logs and pointless Celery retries for input that can never parse. A new UnsupportedFileTypeError (ValueError subclass, so existing callers still catch it) is classified in the task as a terminal client-data failure: logged at WARNING, job marked failed with reason "unsupported_file_type", never retried. Recovered from uncommitted work predating PR #6 (which deliberately excluded it); test fixes on top: IngestJob's pk is id not job_id, and .run() needs push_request() since the task persists self.request.id into the NOT NULL celery_task_id column. Co-Authored-By: Claude Fable 5 --- docs/deploy.md | 13 +--- mnemosyne/library/services/parsers.py | 15 +++- mnemosyne/library/tasks.py | 19 +++++ mnemosyne/library/tests/test_tasks.py | 108 ++++++++++++++++++++++++++ 4 files changed, 144 insertions(+), 11 deletions(-) diff --git a/docs/deploy.md b/docs/deploy.md index 220fc4c..04cc8d7 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -85,26 +85,21 @@ an explicit `when: mnemosyne_first_deploy` flag. ```bash # Apply Django ORM migrations (PostgreSQL schema) -docker compose -f /srv/mnemosyne/docker-compose.yaml run --rm app migrate +docker compose run --rm app migrate # Create Neo4j vector + full-text indexes and load library-type defaults -docker compose -f /srv/mnemosyne/docker-compose.yaml \ - run --rm app setup +docker compose run --rm app setup # Seed the MCPSigningKey used to sign long-lived Pallas team JWTs. # --retire-other deactivates any previously-active key. The hex # emitted to stdout is persisted in Mnemosyne's database and is # not re-injected from the vault — no operator action required # beyond running this command once per fresh deployment. -docker compose -f /srv/mnemosyne/docker-compose.yaml \ - run --rm app \ - python manage.py seed_signing_key --kid daedalus-1 --retire-other +docker compose run --rm app python manage.py seed_signing_key --kid daedalus-1 --retire-other # Create Django groups for SSO role mapping (View Only / Staff / SME / Admin). # Safe to re-run — idempotent. -docker compose -f /srv/mnemosyne/docker-compose.yaml \ - run --rm app \ - python manage.py create_sso_groups +docker compose run --rm app python manage.py create_sso_groups ``` The `seed_signing_key` command prints the generated secret once to stdout — it diff --git a/mnemosyne/library/services/parsers.py b/mnemosyne/library/services/parsers.py index 6a44fb6..28abeb4 100644 --- a/mnemosyne/library/services/parsers.py +++ b/mnemosyne/library/services/parsers.py @@ -22,6 +22,17 @@ from .text_utils import remove_excessive_whitespace, sanitize_text logger = logging.getLogger(__name__) + +class UnsupportedFileTypeError(ValueError): + """Raised when a file's type cannot be parsed. + + Deterministic and input-driven — re-parsing identical bytes can never + succeed — so ingest must treat it as a terminal failure and never retry. + Subclasses ``ValueError`` so existing ``except ValueError`` callers still + catch it. + """ + + # File extensions supported by PyMuPDF PYMUPDF_EXTENSIONS = { "pdf", "epub", "xps", "mobi", "fb2", "cbz", "svg", @@ -88,7 +99,7 @@ class DocumentParser: :param file_path: Path to the document file. :param file_type: File extension (without dot), e.g. 'pdf', 'epub'. :returns: ParseResult with text blocks, images, and metadata. - :raises ValueError: If the file type is not supported. + :raises UnsupportedFileTypeError: If the file type is not supported. """ file_type = file_type.lower().lstrip(".") @@ -117,7 +128,7 @@ class DocumentParser: if file_type in ("html", "htm"): return self._parse_with_pymupdf(file_path, file_type) - raise ValueError( + raise UnsupportedFileTypeError( f"Unsupported file type '{file_type}'. " f"Supported: {sorted(PYMUPDF_EXTENSIONS | PLAINTEXT_EXTENSIONS | IMAGE_EXTENSIONS)}" ) diff --git a/mnemosyne/library/tasks.py b/mnemosyne/library/tasks.py index bf55e03..b70525a 100644 --- a/mnemosyne/library/tasks.py +++ b/mnemosyne/library/tasks.py @@ -347,6 +347,7 @@ def ingest_from_daedalus(self, job_id: str): from datetime import datetime, timezone from library.models import IngestJob, Item, Library + from library.services.parsers import UnsupportedFileTypeError from library.services.source_s3 import ( copy_into_mnemosyne, fetch_from_source, @@ -465,6 +466,24 @@ def ingest_from_daedalus(self, job_id: str): **result, } + except UnsupportedFileTypeError as exc: + # Deterministic, input-driven — re-parsing identical bytes can never + # succeed. Terminal client-data failure, never retried, and logged at + # WARNING (not ERROR) because an unparseable input is not a server fault. + logger.warning( + "Task ingest_from_daedalus rejected job_id=%s: %s", job_id, exc, + ) + job.status = "failed" + job.error = str(exc) + job.completed_at = datetime.now(timezone.utc) + job.save(update_fields=["status", "error", "completed_at"]) + return { + "success": False, + "job_id": job_id, + "error": str(exc), + "reason": "unsupported_file_type", + } + except Exception as exc: logger.error( "Task ingest_from_daedalus failed job_id=%s: %s", diff --git a/mnemosyne/library/tests/test_tasks.py b/mnemosyne/library/tests/test_tasks.py index 239b6d2..2d5a8f6 100644 --- a/mnemosyne/library/tests/test_tasks.py +++ b/mnemosyne/library/tests/test_tasks.py @@ -67,6 +67,114 @@ class ReembedItemTaskTests(TestCase): mock_pipeline.reprocess_item.assert_called_once() +@override_settings(CELERY_TASK_ALWAYS_EAGER=True) +class IngestFromDaedalusFailureClassificationTests(TestCase): + """A deterministic parse failure is terminal; a transient error retries. + + Neo4j and S3 are mocked at the task's boundaries so the test exercises the + exception-classification branch (parsers.UnsupportedFileTypeError vs. any + other Exception) without a live graph or object store. + """ + + def setUp(self): + from library.tasks import ingest_from_daedalus + + # Calling .run() bypasses Celery's request setup, but the task + # persists self.request.id into the NOT NULL celery_task_id column + # and reads self.request.retries — push a real request context. + ingest_from_daedalus.push_request(id="test-task-id", retries=0) + self.addCleanup(ingest_from_daedalus.pop_request) + + def _make_job(self): + from library.models import IngestJob + + return IngestJob.objects.create( + id="job_test_unsupported", + library_uid="lib-uid-123", + source="daedalus", + s3_key="incoming/bad.drawio", + file_type="vnd.jgraph.mxfile", + title="bad.drawio", + content_hash="abc123", + ) + + def _patched_boundaries(self, pipeline_side_effect): + """Patch every boundary the task hits before the pipeline runs. + + Returns a context-manager list; the pipeline's ``process_item`` is set + to raise ``pipeline_side_effect``. + """ + from library.services.parsers import UnsupportedFileTypeError # noqa: F401 + + patchers = [ + patch("library.tasks.db"), + patch("library.models.Library"), + patch("library.models.Item"), + patch("library.services.source_s3.fetch_from_source", return_value=b"data"), + patch("library.services.source_s3.copy_into_mnemosyne"), + patch("library.tasks._resolve_or_create_default_collection"), + patch("library.services.pipeline.EmbeddingPipeline"), + ] + mocks = [p.start() for p in patchers] + self.addCleanup(lambda: [p.stop() for p in patchers]) + + # db.cypher_query returns (rows, meta); no prior item to supersede. + mocks[0].cypher_query.return_value = ([], None) + # Library.nodes.get returns a stand-in library node. + mocks[1].nodes.get.return_value = MagicMock(uid="lib-uid-123") + # Item() instances carry a uid used for the S3 key. + item_instance = MagicMock(uid="item-uid-abc") + mocks[2].return_value = item_instance + # The pipeline raises the classification-relevant error. + pipeline_instance = MagicMock() + pipeline_instance.process_item.side_effect = pipeline_side_effect + mocks[6].return_value = pipeline_instance + return pipeline_instance + + def test_unsupported_file_type_is_terminal_and_not_retried(self): + from library.models import IngestJob + from library.services.parsers import UnsupportedFileTypeError + from library.tasks import ingest_from_daedalus + + job = self._make_job() + self._patched_boundaries( + UnsupportedFileTypeError("Unsupported file type 'vnd.jgraph.mxfile'.") + ) + + with patch.object(ingest_from_daedalus, "retry") as mock_retry: + result = ingest_from_daedalus.run(job.id) + + # Never retried. + mock_retry.assert_not_called() + # Terminal failure with a machine-readable reason. + self.assertFalse(result["success"]) + self.assertEqual(result["reason"], "unsupported_file_type") + + job.refresh_from_db() + self.assertEqual(job.status, "failed") + self.assertEqual(job.retry_count, 0) + self.assertIsNotNone(job.completed_at) + self.assertIn("Unsupported file type", job.error) + + def test_transient_error_takes_the_retry_path(self): + from library.tasks import ingest_from_daedalus + + job = self._make_job() + self._patched_boundaries(ConnectionError("S3 hiccup")) + + # self.retry raises Retry in real Celery; simulate that so the task + # doesn't fall through to the terminal branch. + from celery.exceptions import Retry + + with patch.object(ingest_from_daedalus, "retry", side_effect=Retry()) as mock_retry: + with self.assertRaises(Retry): + ingest_from_daedalus.run(job.id) + + mock_retry.assert_called_once() + job.refresh_from_db() + self.assertEqual(job.retry_count, 1) + + class ResolveUserTests(TestCase): """Tests for the _resolve_user helper."""