diff --git a/hindsight-api-slim/hindsight_api/engine/memory_engine.py b/hindsight-api-slim/hindsight_api/engine/memory_engine.py index 4bf0255412..fc1ed5c6e9 100644 --- a/hindsight-api-slim/hindsight_api/engine/memory_engine.py +++ b/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. diff --git a/hindsight-api-slim/tests/test_operation_completion.py b/hindsight-api-slim/tests/test_operation_completion.py new file mode 100644 index 0000000000..c73665fbd5 --- /dev/null +++ b/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"]))