🐾 fix(tasks): unsupported file types fail ingest terminally, no retries
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 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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)}"
|
||||
)
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user