fix(engine): isolate operation completion from best-effort side-effects

hindsight-api-slim/hindsight_api/engine/memory_engine.py

@@ -2413,16 +2413,14 @@ async def _mark_operation_completed(self, operation_id: str):
                             f"""
                             UPDATE {fq_table("async_operations")}
                             SET status = 'failed', error_message = $2, updated_at = NOW(), completed_at = NOW()
-                            WHERE operation_id = $1
+                            WHERE operation_id = $1 AND status NOT IN ('completed', 'failed', 'cancelled')
                             RETURNING operation_id
                             """,
                             uuid.UUID(operation_id),
                             error_message,
                         )
                         if row is None:
-                            logger.info(
-                                f"Operation {operation_id} no longer exists (bank deleted), skipping mark-failed"
-                            )
+                            logger.info(f"Operation {operation_id} already terminal or deleted, skipping mark-failed")
                             return
                         logger.warning(
                             f"Marked async operation as failed due to {extraction_errors_count} "
@@ -2431,20 +2429,21 @@ async def _mark_operation_completed(self, operation_id: str):
                         await self._maybe_update_parent_operation(operation_id, conn)
                         return
 
-                    # Mark this operation as completed
+                    # Mark this operation as completed. Guarded so an already-terminal
+                    # row is never re-terminalized: this keeps the engine idempotent
+                    # with the worker poller's completion backstop (PR #2608) and never
+                    # re-runs parent aggregation on a row that is already done.
                     row = await conn.fetchrow(
                         f"""
                         UPDATE {fq_table("async_operations")}
                         SET status = 'completed', updated_at = NOW(), completed_at = NOW()
-                        WHERE operation_id = $1
+                        WHERE operation_id = $1 AND status NOT IN ('completed', 'failed', 'cancelled')
                         RETURNING operation_id
                         """,
                         uuid.UUID(operation_id),
                     )
                     if row is None:
-                        logger.info(
-                            f"Operation {operation_id} no longer exists (bank deleted), skipping mark-completed"
-                        )
+                        logger.info(f"Operation {operation_id} already terminal or deleted, skipping mark-completed")
                         return
                     logger.info(f"Marked async operation as completed: {operation_id}")
 
@@ -2545,11 +2544,23 @@ async def _mark_operation_completed_and_fire_webhook(
         schema: str | None = None,
         error_message: str | None = None,
     ) -> None:
-        """Mark an operation as completed and queue webhook deliveries in a single transaction.
+        """Mark an operation as completed and queue its consolidation webhook.
+
+        Happy path uses the transactional outbox pattern: the webhook delivery row is
+        inserted in the *same* transaction as the ``status = 'completed'`` update, which
+        guarantees at-least-once delivery even if the process crashes right after commit.
 
-        Uses the transactional outbox pattern: the webhook delivery row is inserted in the
-        same database transaction as the status update. This guarantees at-least-once delivery
-        even if the process crashes immediately after committing.
+        The critical property is that a failure in the best-effort side-effects (webhook
+        outbox insert, parent aggregation) must never roll back the completion with it.
+        The original code wrapped everything in one transaction and swallowed the
+        exception, so any hiccup left the operation stuck in ``processing`` forever while
+        the log already said the work was done (issue #2601). If the combined transaction
+        fails we therefore fall back to committing the completion on its own and fire the
+        webhook best-effort (non-transactional) instead of dropping both.
+
+        The UPDATE only fires on a non-terminal row, so it is idempotent with the worker
+        poller's completion backstop (PR #2608): whichever path runs second sees an
+        already-terminal row, updates nothing, and does not re-run parent aggregation.
         """
         from ..webhooks.models import ConsolidationEventData, WebhookEvent, WebhookEventType
 
@@ -2561,15 +2572,13 @@ async def _mark_operation_completed_and_fire_webhook(
                         f"""
                         UPDATE {fq_table("async_operations")}
                         SET status = 'completed', updated_at = NOW(), completed_at = NOW()
-                        WHERE operation_id = $1
+                        WHERE operation_id = $1 AND status NOT IN ('completed', 'failed', 'cancelled')
                         RETURNING operation_id
                         """,
                         uuid.UUID(operation_id),
                     )
                     if row is None:
-                        logger.info(
-                            f"Operation {operation_id} no longer exists (bank deleted), skipping mark-completed"
-                        )
+                        logger.info(f"Operation {operation_id} already terminal or deleted, skipping mark-completed")
                         return
                     logger.info(f"Marked async operation as completed: {operation_id}")
                     await self._maybe_update_parent_operation(operation_id, conn)
@@ -2591,8 +2600,51 @@ async def _mark_operation_completed_and_fire_webhook(
                             data=data,
                         )
                         await self._webhook_manager.fire_event_with_conn(event, conn, schema=schema)
+            return
+        except Exception as e:
+            logger.error(
+                f"Atomic complete+webhook failed for {operation_id}: {e}. "
+                "Falling back to a completion-only commit so the operation is not left unfinished."
+            )
+
+        # Fallback: the combined transaction above rolled back (atomically), so the row is
+        # still non-terminal. Commit the terminal state on its own, then deliver the webhook
+        # best-effort. Losing at-least-once atomicity for a single notification is far better
+        # than leaving the operation stuck. We only re-fire the webhook when this fallback
+        # actually transitioned the row: if the row is already terminal the happy-path
+        # transaction had already committed (status + outbox together), so re-firing would
+        # duplicate the delivery.
+        completed_in_fallback = False
+        try:
+            backend = await self._get_backend()
+            async with acquire_with_retry(backend) as conn:
+                async with conn.transaction():
+                    row = await conn.fetchrow(
+                        f"""
+                        UPDATE {fq_table("async_operations")}
+                        SET status = 'completed', updated_at = NOW(), completed_at = NOW()
+                        WHERE operation_id = $1 AND status NOT IN ('completed', 'failed', 'cancelled')
+                        RETURNING operation_id
+                        """,
+                        uuid.UUID(operation_id),
+                    )
+                    if row is not None:
+                        completed_in_fallback = True
+                        await self._maybe_update_parent_operation(operation_id, conn)
         except Exception as e:
-            logger.error(f"Failed to mark operation completed and fire webhook {operation_id}: {e}")
+            # Last-resort: the worker poller's post-executor backstop (PR #2608) still
+            # marks the row completed after this returns.
+            logger.error(f"Fallback completion commit failed for {operation_id}: {e}")
+
+        if completed_in_fallback:
+            await self._fire_consolidation_webhook(
+                bank_id=bank_id,
+                operation_id=operation_id,
+                status=status,
+                result=result,
+                error_message=error_message,
+                schema=schema,
+            )
 
     async def _maybe_update_parent_operation(self, child_operation_id: str, conn):
         """Check if this is a child operation and update parent status if all siblings are done.

hindsight-api-slim/tests/test_operation_completion.py

@@ -0,0 +1,132 @@
+"""Unit tests for MemoryEngine operation-completion side-effect isolation.
+
+These are fast, DB-free tests (fake asyncpg-style connections) that pin the
+critical property fixed for issue #2601: a failure in the best-effort side
+effects of completing an operation (webhook outbox insert, parent aggregation)
+must never roll back — or silently swallow — the completion itself, and the
+consolidation webhook must still be delivered on the fallback path.
+"""
+
+import uuid
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from hindsight_api.engine.memory_engine import MemoryEngine
+
+
+class _FakeTx:
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return False
+
+
+class _FakeConn:
+    def __init__(self, fetchrow_result):
+        self._fetchrow_result = fetchrow_result
+        self.fetchrow_calls: list[tuple[str, tuple]] = []
+
+    def transaction(self):
+        return _FakeTx()
+
+    async def fetchrow(self, query, *args):
+        self.fetchrow_calls.append((query, args))
+        return self._fetchrow_result
+
+
+class _FakeAcquire:
+    def __init__(self, conn):
+        self._conn = conn
+
+    async def __aenter__(self):
+        return self._conn
+
+    async def __aexit__(self, exc_type, exc, tb):
+        return False
+
+
+class _FakeBackend:
+    # Route acquire_with_retry down its DatabaseBackend branch without importing one.
+    _wraps_backend = True
+
+    def __init__(self, conn):
+        self._conn = conn
+
+    def acquire(self):
+        return _FakeAcquire(self._conn)
+
+
+def _make_engine(fetchrow_result, *, webhook_raises: bool):
+    """Build a MemoryEngine with only the attributes the completion path touches."""
+    conn = _FakeConn(fetchrow_result)
+    engine = MemoryEngine.__new__(MemoryEngine)
+    engine._get_backend = AsyncMock(return_value=_FakeBackend(conn))
+    engine._maybe_update_parent_operation = AsyncMock()
+    engine._fire_consolidation_webhook = AsyncMock()
+
+    webhook_manager = MagicMock()
+    if webhook_raises:
+        webhook_manager.fire_event_with_conn = AsyncMock(side_effect=RuntimeError("outbox insert failed"))
+    else:
+        webhook_manager.fire_event_with_conn = AsyncMock()
+    engine._webhook_manager = webhook_manager
+    return engine, conn
+
+
+class TestMarkOperationCompletedAndFireWebhook:
+    async def test_happy_path_uses_outbox_and_does_not_double_fire(self):
+        """When the atomic outbox transaction succeeds, the best-effort
+        (non-transactional) webhook fire must NOT run — that would duplicate delivery."""
+        op_id = str(uuid.uuid4())
+        engine, conn = _make_engine({"operation_id": op_id}, webhook_raises=False)
+
+        await engine._mark_operation_completed_and_fire_webhook(
+            operation_id=op_id, bank_id="bank-1", status="completed", result={"observations_created": 2}
+        )
+
+        # Completion committed exactly once, guarded on 'processing'.
+        assert len(conn.fetchrow_calls) == 1
+        assert "NOT IN ('completed', 'failed', 'cancelled')" in conn.fetchrow_calls[0][0]
+        engine._webhook_manager.fire_event_with_conn.assert_awaited_once()
+        engine._fire_consolidation_webhook.assert_not_awaited()
+
+    async def test_webhook_failure_falls_back_to_completion_and_best_effort_webhook(self):
+        """If the outbox transaction fails, the operation must still be completed and
+        the webhook delivered best-effort — not left stuck in 'processing' (issue #2601)."""
+        op_id = str(uuid.uuid4())
+        engine, conn = _make_engine({"operation_id": op_id}, webhook_raises=True)
+
+        await engine._mark_operation_completed_and_fire_webhook(
+            operation_id=op_id, bank_id="bank-1", status="completed", result=None
+        )
+
+        # Two completion UPDATEs: the rolled-back happy path + the fallback commit.
+        assert len(conn.fetchrow_calls) == 2
+        for query, _args in conn.fetchrow_calls:
+            assert "NOT IN ('completed', 'failed', 'cancelled')" in query
+        # Best-effort webhook fired exactly once on the fallback path.
+        engine._fire_consolidation_webhook.assert_awaited_once()
+        _, kwargs = engine._fire_consolidation_webhook.await_args
+        assert kwargs["operation_id"] == op_id
+        assert kwargs["status"] == "completed"
+
+    async def test_already_terminal_row_is_a_noop(self):
+        """Idempotency with the poller backstop (PR #2608): if the row is no longer
+        'processing', do nothing — no parent aggregation, no webhook."""
+        op_id = str(uuid.uuid4())
+        engine, conn = _make_engine(None, webhook_raises=False)
+
+        await engine._mark_operation_completed_and_fire_webhook(
+            operation_id=op_id, bank_id="bank-1", status="completed", result=None
+        )
+
+        assert len(conn.fetchrow_calls) == 1
+        engine._maybe_update_parent_operation.assert_not_awaited()
+        engine._webhook_manager.fire_event_with_conn.assert_not_awaited()
+        engine._fire_consolidation_webhook.assert_not_awaited()
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))