fix(backend): ensure processing job reconciliation against ScaleODM works well

also fixes tests. Assisted by: Opus 4.6 LLM

Author: spwoodcock

src/backend/app/projects/project_logic.py

@@ -131,7 +131,7 @@ async def reconcile_project_level_odm(
 
     Fetches the single project-level ScaleODM task status and reconciles:
     - Completed but webhook missed -> enqueue finalize job (reproject + state)
-    - Failed/canceled -> mark project FAILED
+    - Failed/canceled -> enqueue finalize job (captures error + cleanup)
     - Not found (age > threshold) -> mark project FAILED
     - Otherwise -> report current status
 
@@ -250,19 +250,25 @@ async def reconcile_project_level_odm(
             }
 
         if status_code in (30, 50):
-            # Failed or canceled - mark project FAILED.
-            try:
-                await update_processing_status(
-                    db, project.id, ImageProcessingStatus.FAILED
-                )
-                log.warning(
-                    "Reconciled project-level ODM failure: project={} odm_uuid={} status={}",
-                    project.id,
-                    project_odm_uuid,
-                    status_label,
-                )
-            except Exception as e:
-                log.error("Failed to reconcile project ODM failure: {}", e)
+            # Failed or canceled - enqueue finalize to capture error message + cleanup.
+            scaleodm_endpoint = (
+                getattr(project, "odm_endpoint_used", None) or settings.ODM_ENDPOINT
+            )
+            await redis_pool.enqueue_job(
+                "finalize_scaleodm_task",
+                scaleodm_url=scaleodm_endpoint,
+                dtm_project_id=str(project.id),
+                odm_task_uuid=project_odm_uuid,
+                odm_status_code=30,  # normalize 50 (canceled) → 30 (failed)
+                _job_id=f"odm-assets:project:{project.id}",
+                _queue_name="default_queue",
+            )
+            log.warning(
+                "Reconciling project-level ODM failure: project={} odm_uuid={} status={}",
+                project.id,
+                project_odm_uuid,
+                status_label,
+            )
 
             display_label = "Failed (canceled)" if status_code == 50 else status_label
             task_entry = project_schemas.OdmQueueTask(

src/backend/app/projects/project_routes.py

@@ -558,7 +558,7 @@ async def odm_webhook_for_processing_whole_project(
     if not odm_task_id or not status:
         raise HTTPException(status_code=400, detail="Invalid webhook payload")
 
-    if status["code"] in {30, 40}:
+    if status["code"] in {30, 40, 50}:
         # Use the endpoint persisted when processing started so finalize jobs
         # target the correct server even if config has changed since.
         scaleodm_url = await _get_project_odm_endpoint(db, dtm_project_id)
@@ -567,7 +567,7 @@ async def odm_webhook_for_processing_whole_project(
             scaleodm_url=scaleodm_url,
             dtm_project_id=str(dtm_project_id),
             odm_task_uuid=odm_task_id,
-            odm_status_code=status["code"],
+            odm_status_code=30 if status["code"] == 50 else status["code"],
             _job_id=f"odm-assets:project:{dtm_project_id}",
             _queue_name="default_queue",
         )
@@ -612,6 +612,24 @@ async def odm_webhook_for_processing_a_single_task(
         status["code"],
     )
 
+    # Reject stale webhooks from a previous ScaleODM submission so they
+    # can't corrupt state after a user re-submits a failed task.
+    async with db.cursor(row_factory=dict_row) as _cur:
+        await _cur.execute(
+            "SELECT odm_task_uuid FROM tasks WHERE id = %(tid)s AND project_id = %(pid)s",
+            {"tid": str(dtm_task_id), "pid": str(dtm_project_id)},
+        )
+        _row = await _cur.fetchone()
+    current_odm_uuid = _row["odm_task_uuid"] if _row else None
+    if current_odm_uuid and odm_task_id != str(current_odm_uuid):
+        log.warning(
+            "ODM webhook for task {}: payload uuid {} does not match current odm_task_uuid {} - ignoring stale webhook",
+            dtm_task_id,
+            odm_task_id,
+            current_odm_uuid,
+        )
+        return {"message": "Webhook ignored (stale)", "odm_task_id": odm_task_id}
+
     if status["code"] == 40:
         await redis_pool.enqueue_job(
             "finalize_scaleodm_task",
@@ -626,7 +644,7 @@ async def odm_webhook_for_processing_a_single_task(
             _queue_name="default_queue",
         )
 
-    elif status["code"] == 30 and state_value != State.IMAGE_PROCESSING_FAILED:
+    elif status["code"] in {30, 50} and state_value != State.IMAGE_PROCESSING_FAILED:
         # Finalize handler records the failure and clears the ScaleODM row.
         # We don't pre-flip state here so the finalize job can pull the real
         # errorMessage from ScaleODM and write a more descriptive comment.
@@ -1384,29 +1402,32 @@ async def fetch_task_info(
                         odm_uuid,
                     )
                 else:
-                    # Failed or canceled - just flip the state
-                    try:
-                        await task_logic.update_task_state_system(
-                            db,
-                            project.id,
-                            dtm_task_id,
-                            f"Reconciled: ScaleODM reports {status_label}",
-                            State.IMAGE_PROCESSING_STARTED,
-                            State.IMAGE_PROCESSING_FAILED,
-                            timestamp(),
-                        )
-                        reconciled.append(
-                            f"Task {task_index} ({dtm_task_id}): STARTED -> FAILED ({status_label})"
-                        )
-                        log.warning(
-                            "Reconciled stuck task: project={} dtm_task={} odm_uuid={} odm_status={}",
-                            project.id,
-                            dtm_task_id,
-                            odm_uuid,
-                            status_label,
-                        )
-                    except Exception as e:
-                        log.error("Failed to reconcile task {}: {}", dtm_task_id, e)
+                    # Failed or canceled - enqueue finalize to capture the real
+                    # error message from ScaleODM and clean up the ScaleODM row.
+                    # Direct update_task_state_system here would skip cleanup and
+                    # cause the subsequent ScaleODM webhook to be silently ignored.
+                    await redis_pool.enqueue_job(
+                        "finalize_scaleodm_task",
+                        scaleodm_url=odm_url,
+                        dtm_project_id=str(project.id),
+                        odm_task_uuid=odm_uuid,
+                        state_name=State.IMAGE_PROCESSING_STARTED.name,
+                        message=f"Reconciled: ScaleODM reports {status_label}",
+                        dtm_task_id=str(dtm_task_id),
+                        odm_status_code=30,  # normalize 50 (canceled) → 30 (failed)
+                        _job_id=f"odm-assets:task:{dtm_task_id}",
+                        _queue_name="default_queue",
+                    )
+                    reconciled.append(
+                        f"Task {task_index} ({dtm_task_id}): STARTED -> FAILED ({status_label}, enqueued finalize)"
+                    )
+                    log.warning(
+                        "Reconciling failed/canceled task: project={} dtm_task={} odm_uuid={} odm_status={}",
+                        project.id,
+                        dtm_task_id,
+                        odm_uuid,
+                        status_label,
+                    )
         elif fetch_failed:
             # Could not reach ScaleODM for this task (timeout, network error).
             # Do NOT reconcile - the task may still be running. Show it as

src/backend/tests/test_project_processing.py

@@ -10,11 +10,32 @@
 from app.arq import tasks as arq_tasks
 
 
+class _FakeCursor:
+    """Minimal async context-manager cursor; fetchone() returns None by default."""
+
+    def __init__(self, fetchone_result=None):
+        self._fetchone_result = fetchone_result
+        self.executed = []
+
+    async def execute(self, query, params=None):
+        self.executed.append({"query": query, "params": params})
+
+    async def fetchone(self):
+        return self._fetchone_result
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return False
+
+
 class _FakeConn:
-    def __init__(self):
+    def __init__(self, cursor_fetchone=None):
         self.commit_calls = 0
         self.rollback_calls = 0
         self.executed = []
+        self._cursor_fetchone = cursor_fetchone
 
     async def execute(self, query, params=None):
         self.executed.append({"query": query, "params": params})
@@ -25,6 +46,9 @@ async def commit(self):
     async def rollback(self):
         self.rollback_calls += 1
 
+    def cursor(self, **kwargs):
+        return _FakeCursor(self._cursor_fetchone)
+
 
 class _FakePoolConnection:
     def __init__(self, conn):

tasks/start

@@ -36,7 +36,7 @@ all:
   # ScaleODM runs inside a local Talos k8s cluster (itself Docker containers).
   # Workflow pods need to reach DroneTM's S3 (RustFS), which is published on
   # the Docker host at port 9000. The reachable address from inside those pods
-  # is the Docker network's gateway IP — not localhost.
+  # is the Docker network's gateway IP - not localhost.
   #
   # We only auto-detect when SCALEODM_S3_ENDPOINT is unset/empty, so an
   # explicit value in .env always wins.