fix: stabilize celery reconnect handling

.env.example

@@ -45,6 +45,18 @@ HF_HUB_DISABLE_SYMLINKS_WARNING=1
 HF_TOKEN=****
 
 REDIS_URL=redis://localhost:6379/0
+# Set false if you start Celery manually in a second terminal.
+AUTO_START_CELERY=true
+CELERY_TASK_ACKS_LATE=false
+CELERY_TASK_REJECT_ON_WORKER_LOST=false
+CELERY_WORKER_CANCEL_ON_CONNECTION_LOSS=false
+CELERY_WORKER_PREFETCH_MULTIPLIER=1
+CELERY_VISIBILITY_TIMEOUT_S=7200
+CELERY_BROKER_HEARTBEAT_S=30
+CELERY_BROKER_POOL_LIMIT=1
+CELERY_REDIS_SOCKET_TIMEOUT_S=30
+CELERY_REDIS_HEALTH_CHECK_INTERVAL_S=30
+CELERY_RESULT_EXPIRES_S=86400
 
 MASTER_ADMIN_KEY=****
 

ARCHITECTURE.md

@@ -637,6 +637,9 @@ Login
 | `GEMINI_API_KEY` | Google Gemini access (ingestion + vision) | Yes |
 | `COHERE_API_KEY` | Cohere reranking | Yes |
 | `REDIS_URL` | Redis connection string | Yes |
+| `CELERY_WORKER_CANCEL_ON_CONNECTION_LOSS` | Whether broker disconnect cancels active ingestion tasks; keep `false` for long uploads | No |
+| `CELERY_TASK_ACKS_LATE` | Whether ingestion tasks acknowledge only after completion; keep `false` to avoid redelivery loops on flaky Redis | No |
+| `CELERY_TASK_REJECT_ON_WORKER_LOST` | Whether lost workers requeue tasks; keep `false` unless duplicate-safe retries are required | No |
 | `MASTER_ADMIN_KEY` | Admin endpoint access | Yes |
 | `ALLOWED_ORIGINS` | CORS allowed origins (use `*` for dev only) | Yes |
 | `DOCS_ENABLED` | Enable /docs and /redoc (set `false` in prod) | No |

backend/api/ingest.py

@@ -8,6 +8,7 @@ from backend.core.auth_utils import is_guest_token, require_auth_token
 from backend.core.rate_limit import limiter
 from backend.core.tasks import process_document_task
 from backend.core.tasks import celery_app
+from backend.core.tasks import TRANSIENT_CELERY_CONNECTION_ERRORS
 
 log = logging.getLogger("morpheus.api.ingest")
 router = APIRouter()
@@ -64,6 +65,25 @@ def _ensure_ingest_worker_available() -> None:
         )
 
 
+def _raise_reconnecting(exc: BaseException) -> None:
+    raise HTTPException(
+        status_code=503,
+        detail="Background ingestion queue is reconnecting. Please retry shortly.",
+    ) from exc
+
+
+def _get_task_result(task_id: str):
+    try:
+        return celery_app.AsyncResult(task_id)
+    except TRANSIENT_CELERY_CONNECTION_ERRORS as exc:
+        log.warning(
+            "Celery result backend reconnecting while reading task %s: %s",
+            task_id,
+            exc,
+        )
+        _raise_reconnecting(exc)
+
+
 def _register_ingest_task(task_id: str, *, user_id: str, tmp_path: str, filename: str) -> None:
     with _TASK_LOCK:
         _ACTIVE_INGEST_TASKS[task_id] = {
@@ -172,7 +192,14 @@ async def upload(
             )
         with open(tmp_path, "wb") as f:
             f.write(contents)
-        task = process_document_task.delay(tmp_path, file.filename, x_auth_token)
+        try:
+            task = process_document_task.delay(tmp_path, file.filename, x_auth_token)
+        except TRANSIENT_CELERY_CONNECTION_ERRORS as exc:
+            log.warning(
+                "Failed to queue ingestion while Celery/Redis is reconnecting: %s",
+                exc,
+            )
+            _raise_reconnecting(exc)
         _register_ingest_task(
             task.id,
             user_id=user_id,
@@ -202,7 +229,7 @@ def get_ingest_status(task_id: str):
         _forget_ingest_task(task_id)
         return {"status": "CANCELLED", "message": "Ingestion cancelled."}
 
-    task_result = celery_app.AsyncResult(task_id)
+    task_result = _get_task_result(task_id)
 
     if task_result.state == "PENDING":
         return {"status": "PENDING", "message": "Waiting in queue..."}
@@ -240,7 +267,7 @@ def cancel_ingest(
 ):
     _ensure_ingest_worker_available()
     active = _get_user_ingest_task(task_id, user_id)
-    task_result = celery_app.AsyncResult(task_id)
+    task_result = _get_task_result(task_id)
     if task_result.state in {"SUCCESS", "FAILURE", "REVOKED"}:
         _forget_ingest_task(task_id)
         return {"ok": True, "status": task_result.state}
@@ -251,6 +278,13 @@ def cancel_ingest(
         else:
             celery_app.control.revoke(task_id, terminate=True, signal="SIGTERM")
         _CANCELLED_INGEST_TASKS.add(task_id)
+    except TRANSIENT_CELERY_CONNECTION_ERRORS as exc:
+        log.warning(
+            "Could not cancel ingestion task %s while Celery/Redis reconnects: %s",
+            task_id,
+            exc,
+        )
+        _raise_reconnecting(exc)
     except Exception as exc:
         log.error("Could not cancel ingestion task %s: %s", task_id, exc)
         raise HTTPException(status_code=500, detail="Could not cancel ingestion task.") from exc

backend/core/config.py

@@ -197,6 +197,21 @@ UPLOAD_RETRY_MAX_SLEEP_S = float(os.getenv("UPLOAD_RETRY_MAX_SLEEP_S", "20"))
 CELERY_VISIBILITY_TIMEOUT_S = int(os.getenv("CELERY_VISIBILITY_TIMEOUT_S", "7200"))
 CELERY_BROKER_HEARTBEAT_S = int(os.getenv("CELERY_BROKER_HEARTBEAT_S", "30"))
 CELERY_BROKER_POOL_LIMIT = int(os.getenv("CELERY_BROKER_POOL_LIMIT", "1"))
+CELERY_TASK_ACKS_LATE = os.getenv("CELERY_TASK_ACKS_LATE", "false").strip().lower() in {
+    "1",
+    "true",
+    "yes",
+}
+CELERY_TASK_REJECT_ON_WORKER_LOST = os.getenv(
+    "CELERY_TASK_REJECT_ON_WORKER_LOST", "false"
+).strip().lower() in {"1", "true", "yes"}
+CELERY_WORKER_CANCEL_ON_CONNECTION_LOSS = os.getenv(
+    "CELERY_WORKER_CANCEL_ON_CONNECTION_LOSS", "false"
+).strip().lower() in {"1", "true", "yes"}
+CELERY_WORKER_PREFETCH_MULTIPLIER = int(
+    os.getenv("CELERY_WORKER_PREFETCH_MULTIPLIER", "1")
+)
+CELERY_RESULT_EXPIRES_S = int(os.getenv("CELERY_RESULT_EXPIRES_S", "86400"))
 CELERY_REDIS_SOCKET_TIMEOUT_S = float(
     os.getenv("CELERY_REDIS_SOCKET_TIMEOUT_S", "30")
 )

backend/core/tasks.py

@@ -14,17 +14,47 @@ try:
 except Exception:
     Celery = None
 
+try:
+    from kombu.exceptions import OperationalError as KombuOperationalError
+except Exception:
+    KombuOperationalError = None
+
+try:
+    from redis.exceptions import ConnectionError as RedisConnectionError
+    from redis.exceptions import TimeoutError as RedisTimeoutError
+except Exception:
+    RedisConnectionError = None
+    RedisTimeoutError = None
+
+
+TRANSIENT_CELERY_CONNECTION_ERRORS = tuple(
+    err
+    for err in (
+        KombuOperationalError,
+        RedisConnectionError,
+        RedisTimeoutError,
+        ConnectionError,
+        TimeoutError,
+    )
+    if err is not None
+)
+
 if Celery is not None:
     celery_app = Celery("morpheus_worker", broker=REDIS_URL, backend=REDIS_URL)
     celery_app.conf.update(
         task_track_started=True,
-        task_acks_late=True,
-        task_reject_on_worker_lost=True,
-        worker_cancel_long_running_tasks_on_connection_loss=True,
+        task_acks_late=config.CELERY_TASK_ACKS_LATE,
+        task_reject_on_worker_lost=config.CELERY_TASK_REJECT_ON_WORKER_LOST,
+        worker_cancel_long_running_tasks_on_connection_loss=(
+            config.CELERY_WORKER_CANCEL_ON_CONNECTION_LOSS
+        ),
+        worker_prefetch_multiplier=config.CELERY_WORKER_PREFETCH_MULTIPLIER,
         broker_connection_retry_on_startup=True,
+        broker_connection_retry=True,
         broker_connection_max_retries=None,
         broker_heartbeat=config.CELERY_BROKER_HEARTBEAT_S,
         broker_pool_limit=config.CELERY_BROKER_POOL_LIMIT,
+        result_expires=config.CELERY_RESULT_EXPIRES_S,
         broker_transport_options={
             "visibility_timeout": config.CELERY_VISIBILITY_TIMEOUT_S,
             "socket_keepalive": True,
@@ -69,10 +99,16 @@ def _process_document_task_impl(
     """
     def update_progress(step: int, total_steps: int, msg: str):
         # This updates the Redis cache so FastAPI can read the current status
-        self.update_state(
-            state="PROCESSING", 
-            meta={"step": step, "total": total_steps, "message": msg}
-        )
+        try:
+            self.update_state(
+                state="PROCESSING",
+                meta={"step": step, "total": total_steps, "message": msg},
+            )
+        except TRANSIENT_CELERY_CONNECTION_ERRORS as exc:
+            log.warning(
+                "Skipped ingestion progress update because Celery/Redis is reconnecting: %s",
+                exc,
+            )
 
     try:
         return run_ingestion(

backend/main.py

@@ -52,6 +52,8 @@ async def lifespan(app: FastAPI):
                     "backend.core.tasks",
                     "worker",
                     "--pool=solo",
+                    "--without-gossip",
+                    "--without-mingle",
                     "--loglevel=info",
                 ]
             )

tests/test_ingest_api.py

@@ -138,6 +138,21 @@ def test_get_ingest_status_requires_available_worker(monkeypatch):
     assert exc_info.value.status_code == 503
 
 
+def test_get_ingest_status_returns_503_when_result_backend_reconnects(monkeypatch):
+    class ReconnectingApp:
+        def AsyncResult(self, _task_id):
+            raise ConnectionError("redis reconnecting")
+
+    monkeypatch.setattr(ingest_api, "celery_app", ReconnectingApp())
+    monkeypatch.setattr(ingest_api, "process_document_task", SimpleNamespace(delay=lambda *_args: None))
+
+    with pytest.raises(HTTPException) as exc_info:
+        ingest_api.get_ingest_status("task-1")
+
+    assert exc_info.value.status_code == 503
+    assert "reconnecting" in exc_info.value.detail.lower()
+
+
 def test_cancel_ingest_rejects_unknown_task(monkeypatch):
     monkeypatch.setattr(ingest_api, "celery_app", SimpleNamespace())
     monkeypatch.setattr(ingest_api, "process_document_task", SimpleNamespace(delay=lambda *_args: None))
@@ -181,6 +196,36 @@ def test_cancel_ingest_revokes_owned_task_and_cleans_temp_file(monkeypatch):
     assert not os.path.exists(tmp_path)
 
 
+def test_upload_returns_503_when_broker_reconnects_during_queue(monkeypatch):
+    _install_fake_magic(monkeypatch)
+    monkeypatch.setattr(
+        pipeline,
+        "_build_supabase_client",
+        lambda **_kwargs: FakeCountSupabase(count=0),
+    )
+    monkeypatch.setattr(ingest_api, "celery_app", SimpleNamespace())
+    monkeypatch.setattr(
+        ingest_api,
+        "process_document_task",
+        SimpleNamespace(delay=lambda *_args: (_ for _ in ()).throw(ConnectionError("redis reconnecting"))),
+    )
+
+    file = FakeUploadFile("guide.pdf", b"%PDF-1.4\nsmall")
+
+    with pytest.raises(HTTPException) as exc_info:
+        asyncio.run(
+            ingest_api.upload(
+                request=_fake_request(),
+                file=file,
+                user_id="user-1",
+                x_auth_token="token",
+            )
+        )
+
+    assert exc_info.value.status_code == 503
+    assert "reconnecting" in exc_info.value.detail.lower()
+
+
 def test_process_pdf_task_impl_preserves_original_exception_and_cleans_temp_file(monkeypatch):
     fd, tmp_path = tempfile.mkstemp(suffix="_guide.pdf")
     os.close(fd)
@@ -199,6 +244,29 @@ def test_process_pdf_task_impl_preserves_original_exception_and_cleans_temp_file
     assert not os.path.exists(tmp_path)
 
 
+def test_process_pdf_task_impl_ignores_transient_progress_backend_error(monkeypatch):
+    fd, tmp_path = tempfile.mkstemp(suffix="_guide.pdf")
+    os.close(fd)
+    progress_callback = {}
+
+    def _run_ingestion(**kwargs):
+        progress_callback["callback"] = kwargs["progress_callback"]
+        kwargs["progress_callback"](1, 2, "Parsing")
+        return {"ok": True}
+
+    monkeypatch.setattr(tasks, "run_ingestion", _run_ingestion)
+
+    class ReconnectingTask:
+        def update_state(self, **_kwargs):
+            raise ConnectionError("redis reconnecting")
+
+    result = tasks._process_pdf_task_impl(ReconnectingTask(), tmp_path, "guide.pdf", "token")
+
+    assert result == {"ok": True}
+    assert "callback" in progress_callback
+    assert not os.path.exists(tmp_path)
+
+
 def test_upload_accepts_markdown_and_queues_document_task(monkeypatch):
     _install_fake_magic(monkeypatch, mime_type="text/plain")
     monkeypatch.setattr(

tests/test_url_ingestion.py

@@ -9,10 +9,81 @@ from backend.core import pipeline
 from backend.core import url_ingestion
 
 
+@pytest.fixture(autouse=True)
+def _isolate_url_ingestion_config(monkeypatch):
+    monkeypatch.setattr(url_ingestion.config, "URL_INGEST_EXTRACTOR", "basic")
+    monkeypatch.setattr(url_ingestion.config, "URL_INGEST_CRAWL4AI_ENABLED", False)
+    monkeypatch.setattr(url_ingestion.config, "URL_INGEST_CRAWL4AI_FALLBACK", False)
+    monkeypatch.setattr(url_ingestion.config, "URL_INGEST_DEFAULT_MODE", "single_page")
+    monkeypatch.setattr(url_ingestion.config, "URL_INGEST_ALLOWED_HOSTS", [])
+
+
 class _FakeTask:
     id = "task-123"
 
 
+class _FakeIngestionTable:
+    def __init__(self, supabase, name: str):
+        self.supabase = supabase
+        self.name = name
+        self.action = None
+        self.filters = {}
+        self.payload = None
+
+    def select(self, *_args):
+        self.action = "select"
+        return self
+
+    def delete(self):
+        self.action = "delete"
+        return self
+
+    def upsert(self, payload, on_conflict=None):
+        self.action = "upsert"
+        self.payload = payload
+        self.on_conflict = on_conflict
+        return self
+
+    def insert(self, payload):
+        self.action = "insert"
+        self.payload = payload
+        return self
+
+    def eq(self, key, value):
+        self.filters[key] = value
+        return self
+
+    def contains(self, key, value):
+        self.filters[key] = value
+        return self
+
+    def limit(self, value):
+        self.filters["limit"] = value
+        return self
+
+    def execute(self):
+        self.supabase.ops.append((self.name, self.action, dict(self.filters)))
+        if self.action == "insert":
+            self.supabase.inserts.append((self.name, self.payload))
+        if self.action == "upsert":
+            self.supabase.upserts.append(
+                (self.name, self.payload, getattr(self, "on_conflict", None))
+            )
+        if self.name == "ingested_files" and self.action == "select":
+            return SimpleNamespace(data=[])
+        return SimpleNamespace(data=[])
+
+
+class _FakeIngestionSupabase:
+    def __init__(self):
+        self.ops = []
+        self.inserts = []
+        self.upserts = []
+
+    def table(self, name: str):
+        return _FakeIngestionTable(self, name)
+
+
 class _FakeResponse:
     def __init__(self, *, url: str, content_type: str, body: bytes, headers=None, status_code: int = 200):
         self.url = url
@@ -88,6 +159,7 @@ def test_fetch_url_to_tempfile_extracts_readable_html(monkeypatch):
 
 def test_fetch_url_to_tempfile_uses_crawl4ai_extractor_when_enabled(monkeypatch):
     monkeypatch.setattr(url_ingestion.config, "URL_INGEST_EXTRACTOR", "crawl4ai")
+    monkeypatch.setattr(url_ingestion.config, "URL_INGEST_CRAWL4AI_ENABLED", True)
     monkeypatch.setattr(
         url_ingestion,
         "_run_crawl4ai_single_page",
@@ -162,9 +234,7 @@ def test_fetch_url_to_tempfile_bounded_crawl_requires_allowlist(monkeypatch):
 
 
 def test_run_ingestion_with_url_override_persists_url_metadata(monkeypatch):
-    from tests.test_pipeline_regressions import FakeIngestionSupabase
-
-    fake_supabase = FakeIngestionSupabase()
+    fake_supabase = _FakeIngestionSupabase()
     captured = {}
     root = Path("tests") / "_tmp_graph_hybrid" / "url_ingestion"
     root.mkdir(parents=True, exist_ok=True)
@@ -240,7 +310,11 @@ def test_run_url_ingest_requires_admin_and_queues_task(monkeypatch):
             pages=[{"final_url": url}],
         ),
     )
-    monkeypatch.setattr(admin_api.process_document_task, "delay", lambda *args: _FakeTask())
+    monkeypatch.setattr(
+        admin_api,
+        "process_document_task",
+        SimpleNamespace(delay=lambda *args: _FakeTask()),
+    )
 
     result = admin_api.run_url_ingest(
         admin_api.UrlIngestPayload(