From 163b80d360a9128f1672652d08e78422f8a4ff0a Mon Sep 17 00:00:00 2001 From: portable Date: Thu, 20 Aug 2026 17:36:57 +0100 Subject: [PATCH 1/4] fix(outbox): add lease-based claim/reclaim to prevent stranded events - Add claimed_at column to OutboxEvent for lease tracking - Atomic claim via UPDATE ... WHERE status IN (pending,failed) - Reclaim stale processing events older than 5 minutes - Validate UUID format in process_position_opened_task - Clear claimed_at on failure for proper retry - Alembic migration for claimed_at column - Tests for failed publish, stale reclaim, UUID validation Closes #417 --- .../add_claimed_at_to_outbox_event.py | 31 ++ quantara/web_app/db/models.py | 5 + quantara/web_app/tasks/outbox_relay.py | 71 +++-- quantara/web_app/tests/test_outbox_relay.py | 282 ++++++++++++++++++ 4 files changed, 372 insertions(+), 17 deletions(-) create mode 100644 quantara/web_app/alembic/versions/add_claimed_at_to_outbox_event.py create mode 100644 quantara/web_app/tests/test_outbox_relay.py diff --git a/quantara/web_app/alembic/versions/add_claimed_at_to_outbox_event.py b/quantara/web_app/alembic/versions/add_claimed_at_to_outbox_event.py new file mode 100644 index 00000000..2bef0a22 --- /dev/null +++ b/quantara/web_app/alembic/versions/add_claimed_at_to_outbox_event.py @@ -0,0 +1,31 @@ +"""add claimed_at to outbox_event + +Revision ID: add_claimed_at_outbox_rev +Revises: outbox_event_table_rev +Create Date: 2026-08-20 00:00:00.000000 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = "add_claimed_at_outbox_rev" +down_revision = "outbox_event_table_rev" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column( + "event_outbox", + sa.Column( + "claimed_at", + sa.DateTime(), + nullable=True, + ), + ) + + +def downgrade() -> None: + op.drop_column("event_outbox", "claimed_at") diff --git a/quantara/web_app/db/models.py b/quantara/web_app/db/models.py index a2506f27..569b130e 100644 --- a/quantara/web_app/db/models.py +++ b/quantara/web_app/db/models.py @@ -161,6 +161,10 @@ class Vault(Base): DateTime, nullable=False, default=func.now(), onupdate=func.now() ) + __table_args__ = ( + UniqueConstraint("user_id", "symbol", name="uq_vault_user_symbol"), + ) + class TransactionStatus(PyEnum): """ @@ -243,6 +247,7 @@ class OutboxEvent(Base): status = Column(String, nullable=False, default="pending") # pending, processing, processed, failed retry_count = Column(Integer, nullable=False, default=0) error_message = Column(String, nullable=True) + claimed_at = Column(DateTime, nullable=True) created_at = Column(DateTime, nullable=False, default=func.now()) updated_at = Column( DateTime, nullable=False, default=func.now(), onupdate=func.now() diff --git a/quantara/web_app/tasks/outbox_relay.py b/quantara/web_app/tasks/outbox_relay.py index fe23aa8e..64034b2b 100644 --- a/quantara/web_app/tasks/outbox_relay.py +++ b/quantara/web_app/tasks/outbox_relay.py @@ -5,9 +5,11 @@ import os import asyncio import json +import uuid import sentry_sdk from datetime import datetime, timedelta from celery import Celery +from sqlalchemy import and_, or_ from sqlalchemy.orm import Session from web_app.db.database import SessionLocal, init_db from web_app.db.models import OutboxEvent, Position, Status, Transaction, TransactionStatus @@ -31,6 +33,16 @@ enable_utc=True, ) +STALE_PROCESSING_INTERVAL_MINUTES = 5 + + +def _is_valid_uuid(value: str) -> bool: + try: + uuid.UUID(value) + return True + except ValueError: + return False + @celery_app.task(bind=True, max_retries=5, default_retry_delay=10) def process_position_opened_task(self, event_id: str): @@ -38,7 +50,11 @@ def process_position_opened_task(self, event_id: str): Celery task that consumes the PositionOpened event from the outbox. """ logger.info("processing_position_opened_task_started", event_id=event_id) - + + if not _is_valid_uuid(event_id): + logger.error("invalid_event_id_format", event_id=event_id) + return + init_db() db: Session = SessionLocal() try: @@ -107,7 +123,7 @@ def process_position_opened_task(self, event_id: str): except Exception as exc: db.rollback() logger.exception("outbox_event_processing_failed", event_id=event_id, error=str(exc)) - + # Update event status to failed and increment retry try: with SessionLocal() as fail_session: @@ -116,6 +132,7 @@ def process_position_opened_task(self, event_id: str): evt.status = "failed" evt.retry_count += 1 evt.error_message = str(exc) + evt.claimed_at = None fail_session.commit() except Exception as update_err: logger.error("failed_to_update_outbox_event_status", error=str(update_err)) @@ -133,15 +150,15 @@ def __init__(self, max_retries: int = 5): def process_pending_events(self): """ - Scans event_outbox for pending/failed events and publishes them to Celery. - Also flags events older than 24h with a Sentry warning. + Scans event_outbox for pending/failed/stale-processing events and + publishes them to Celery. Uses an atomic claim to prevent double-dispatch. """ logger.info("outbox_relay_scan_started") db: Session = SessionLocal() try: # Check for events older than 24h that are not processed cutoff_24h_naive = datetime.now() - timedelta(hours=24) - + old_events = db.query(OutboxEvent).filter( OutboxEvent.status != "processed", OutboxEvent.created_at < cutoff_24h_naive @@ -152,19 +169,44 @@ def process_pending_events(self): logger.warning("outbox_event_older_than_24h", event_id=str(event.id), created_at=str(event.created_at)) sentry_sdk.capture_message(msg, level="warning") - # Fetch pending or failed events - pending_events = db.query(OutboxEvent).filter( + # Reclaim stale "processing" events whose claim has expired + stale_cutoff = datetime.now() - timedelta(minutes=STALE_PROCESSING_INTERVAL_MINUTES) + reclaimed = db.query(OutboxEvent).filter( + OutboxEvent.status == "processing", + OutboxEvent.claimed_at.isnot(None), + OutboxEvent.claimed_at < stale_cutoff, + OutboxEvent.retry_count < self.max_retries, + ).update( + {"status": "pending", "claimed_at": None}, + synchronize_session="fetch", + ) + if reclaimed: + logger.info("reclaimed_stale_processing_events", count=reclaimed) + db.commit() + + # Fetch pending or failed events (now includes freshly reclaimed ones) + candidate_events = db.query(OutboxEvent).filter( OutboxEvent.status.in_(["pending", "failed"]), - OutboxEvent.retry_count < self.max_retries + OutboxEvent.retry_count < self.max_retries, ).all() - if not pending_events: + if not candidate_events: logger.info("no_pending_outbox_events") return - for event in pending_events: - # Mark as processing - event.status = "processing" + for event in candidate_events: + # Atomic claim: only transition to processing if still pending/failed + claimed = db.query(OutboxEvent).filter( + OutboxEvent.id == event.id, + OutboxEvent.status.in_(["pending", "failed"]), + ).update( + {"status": "processing", "claimed_at": datetime.now()}, + synchronize_session="fetch", + ) + if not claimed: + logger.info("outbox_event_already_claimed", event_id=str(event.id)) + continue + db.commit() # Publish to Celery @@ -175,8 +217,3 @@ def process_pending_events(self): logger.exception("outbox_relay_scan_failed", error=str(e)) finally: db.close() - - -if __name__ == "__main__": - relay = OutboxRelay() - relay.process_pending_events() diff --git a/quantara/web_app/tests/test_outbox_relay.py b/quantara/web_app/tests/test_outbox_relay.py new file mode 100644 index 00000000..436bf019 --- /dev/null +++ b/quantara/web_app/tests/test_outbox_relay.py @@ -0,0 +1,282 @@ +""" +Tests for OutboxRelay and related helpers. +These tests use SQLAlchemy with SQLite in-memory databases so they do not +require PostgreSQL or Redis. +""" + +import json +import uuid +from datetime import datetime, timedelta +from unittest.mock import MagicMock, patch + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from web_app.db.database import Base +from web_app.db.models import OutboxEvent +from web_app.tasks.outbox_relay import ( + STALE_PROCESSING_INTERVAL_MINUTES, + OutboxRelay, + _is_valid_uuid, + process_position_opened_task, +) + + +@pytest.fixture +def db_session(): + engine = create_engine("sqlite:///:memory:") + Base.metadata.create_all(engine) + Session = sessionmaker(bind=engine) + session = Session() + yield session + session.close() + + +@pytest.fixture +def relay(): + return OutboxRelay(max_retries=5) + + +# --------------------------------------------------------------------------- +# UUID validation +# --------------------------------------------------------------------------- + +class TestUUIDValidation: + def test_valid_uuid(self): + assert _is_valid_uuid(str(uuid.uuid4())) is True + + def test_invalid_uuid(self): + assert _is_valid_uuid("not-a-uuid") is False + + def test_empty_string(self): + assert _is_valid_uuid("") is False + + def test_truncated_uuid(self): + valid = str(uuid.uuid4()) + assert _is_valid_uuid(valid[:30]) is False + + +# --------------------------------------------------------------------------- +# OutboxRelay.process_pending_events +# --------------------------------------------------------------------------- + +class TestProcessPendingEvents: + @patch("web_app.tasks.outbox_relay.SessionLocal") + @patch("web_app.tasks.outbox_relay.process_position_opened_task") + @patch("web_app.tasks.outbox_relay.init_db") + def test_pending_event_is_published( + self, mock_init_db, mock_delay, mock_session_local, db_session + ): + event = OutboxEvent( + id=uuid.uuid4(), + event_type="PositionOpened", + payload=json.dumps({"position_id": "p1", "transaction_hash": "0xabc"}), + status="pending", + ) + db_session.add(event) + db_session.commit() + mock_session_local.return_value = db_session + + relay = OutboxRelay(max_retries=5) + relay.process_pending_events() + + updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + assert updated.status == "processing" + assert updated.claimed_at is not None + mock_delay.assert_called_once_with(str(event.id)) + + @patch("web_app.tasks.outbox_relay.SessionLocal") + @patch("web_app.tasks.outbox_relay.process_position_opened_task") + @patch("web_app.tasks.outbox_relay.init_db") + def test_failed_event_is_requeued( + self, mock_init_db, mock_delay, mock_session_local, db_session + ): + event = OutboxEvent( + id=uuid.uuid4(), + event_type="PositionOpened", + payload=json.dumps({"position_id": "p1", "transaction_hash": "0xabc"}), + status="failed", + retry_count=1, + error_message="previous error", + ) + db_session.add(event) + db_session.commit() + mock_session_local.return_value = db_session + + relay = OutboxRelay(max_retries=5) + relay.process_pending_events() + + updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + assert updated.status == "processing" + assert updated.claimed_at is not None + mock_delay.assert_called_once_with(str(event.id)) + + @patch("web_app.tasks.outbox_relay.SessionLocal") + @patch("web_app.tasks.outbox_relay.process_position_opened_task") + @patch("web_app.tasks.outbox_relay.init_db") + def test_stale_processing_event_is_reclaimed( + self, mock_init_db, mock_delay, mock_session_local, db_session + ): + stale_time = datetime.now() - timedelta(minutes=STALE_PROCESSING_INTERVAL_MINUTES + 1) + event = OutboxEvent( + id=uuid.uuid4(), + event_type="PositionOpened", + payload=json.dumps({"position_id": "p1", "transaction_hash": "0xabc"}), + status="processing", + claimed_at=stale_time, + retry_count=0, + ) + db_session.add(event) + db_session.commit() + mock_session_local.return_value = db_session + + relay = OutboxRelay(max_retries=5) + relay.process_pending_events() + + updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + assert updated.status == "processing" + assert updated.claimed_at is not None + mock_delay.assert_called_once_with(str(event.id)) + + @patch("web_app.tasks.outbox_relay.SessionLocal") + @patch("web_app.tasks.outbox_relay.process_position_opened_task") + @patch("web_app.tasks.outbox_relay.init_db") + def test_recent_processing_event_is_not_reclaimed( + self, mock_init_db, mock_delay, mock_session_local, db_session + ): + recent_time = datetime.now() - timedelta(minutes=1) + event = OutboxEvent( + id=uuid.uuid4(), + event_type="PositionOpened", + payload=json.dumps({"position_id": "p1", "transaction_hash": "0xabc"}), + status="processing", + claimed_at=recent_time, + retry_count=0, + ) + db_session.add(event) + db_session.commit() + mock_session_local.return_value = db_session + + relay = OutboxRelay(max_retries=5) + relay.process_pending_events() + + updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + assert updated.status == "processing" + mock_delay.assert_not_called() + + @patch("web_app.tasks.outbox_relay.SessionLocal") + @patch("web_app.tasks.outbox_relay.process_position_opened_task") + @patch("web_app.tasks.outbox_relay.init_db") + def test_event_exceeding_max_retries_is_skipped( + self, mock_init_db, mock_delay, mock_session_local, db_session + ): + event = OutboxEvent( + id=uuid.uuid4(), + event_type="PositionOpened", + payload=json.dumps({"position_id": "p1", "transaction_hash": "0xabc"}), + status="pending", + retry_count=5, + ) + db_session.add(event) + db_session.commit() + mock_session_local.return_value = db_session + + relay = OutboxRelay(max_retries=5) + relay.process_pending_events() + + updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + assert updated.status == "pending" + mock_delay.assert_not_called() + + @patch("web_app.tasks.outbox_relay.SessionLocal") + @patch("web_app.tasks.outbox_relay.process_position_opened_task") + @patch("web_app.tasks.outbox_relay.init_db") + def test_already_claimed_event_is_not_double_dispatched( + self, mock_init_db, mock_delay, mock_session_local, db_session + ): + now = datetime.now() + event = OutboxEvent( + id=uuid.uuid4(), + event_type="PositionOpened", + payload=json.dumps({"position_id": "p1", "transaction_hash": "0xabc"}), + status="pending", + ) + db_session.add(event) + db_session.commit() + + original_query = db_session.query + + call_count = [0] + + def patched_query(*args, **kwargs): + q = original_query(*args, **kwargs) + + original_update = q.update + + def patched_update(*uargs, **ukwargs): + call_count[0] += 1 + if call_count[0] == 2: + return 0 + return original_update(*uargs, **ukwargs) + + q.update = patched_update + return q + + db_session.query = patched_query + mock_session_local.return_value = db_session + + relay = OutboxRelay(max_retries=5) + relay.process_pending_events() + + mock_delay.assert_not_called() + + +# --------------------------------------------------------------------------- +# process_position_opened_task – UUID validation +# --------------------------------------------------------------------------- + +class TestProcessPositionOpenedTask: + @patch("web_app.tasks.outbox_relay.init_db") + def test_invalid_uuid_returns_early(self, mock_init_db): + task = MagicMock() + process_position_opened_task(task, event_id="not-a-uuid") + task.retry.assert_not_called() + + @patch("web_app.tasks.outbox_relay.SessionLocal") + @patch("web_app.tasks.outbox_relay.init_db") + def test_valid_uuid_not_found_returns_early(self, mock_init_db, mock_session_local, db_session): + mock_session_local.return_value = db_session + task = MagicMock() + process_position_opened_task(task, event_id=str(uuid.uuid4())) + task.retry.assert_not_called() + + @patch("web_app.tasks.outbox_relay.SessionLocal") + @patch("web_app.tasks.outbox_relay.init_db") + def test_failed_event_clears_claimed_at(self, mock_init_db, mock_session_local, db_session): + event = OutboxEvent( + id=uuid.uuid4(), + event_type="PositionOpened", + payload=json.dumps({"position_id": "p1", "transaction_hash": "0xabc"}), + status="processing", + claimed_at=datetime.now(), + ) + db_session.add(event) + db_session.commit() + mock_session_local.return_value = db_session + + task = MagicMock() + with patch("web_app.tasks.outbox_relay.DashboardMixin") as mock_dash: + mock_dash.get_current_prices = MagicMock( + side_effect=RuntimeError("pricing down") + ) + with patch("web_app.tasks.outbox_relay.PositionDBConnector") as mock_pdbc: + instance = mock_pdbc.return_value + instance.get_object.return_value = None + with pytest.raises(Exception): + process_position_opened_task(task, event_id=str(event.id)) + + updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + assert updated.status == "failed" + assert updated.claimed_at is None + assert updated.retry_count == 1 From a29c550a75bd97259a3ce98227a818dbed5a7bb6 Mon Sep 17 00:00:00 2001 From: portable Date: Fri, 21 Aug 2026 13:21:37 +0100 Subject: [PATCH 2/4] =?UTF-8?q?fix(tests):=20fix=20outbox=20relay=20test?= =?UTF-8?q?=20failures=20=E2=80=94=20use=20StaticPool,=20fix=20Celery=20in?= =?UTF-8?q?vocation,=20fix=20mock=20assertions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- quantara/web_app/tests/test_outbox_relay.py | 30 ++++++++++----------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/quantara/web_app/tests/test_outbox_relay.py b/quantara/web_app/tests/test_outbox_relay.py index 436bf019..825aa900 100644 --- a/quantara/web_app/tests/test_outbox_relay.py +++ b/quantara/web_app/tests/test_outbox_relay.py @@ -12,6 +12,7 @@ import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool from web_app.db.database import Base from web_app.db.models import OutboxEvent @@ -25,7 +26,11 @@ @pytest.fixture def db_session(): - engine = create_engine("sqlite:///:memory:") + engine = create_engine( + "sqlite:///:memory:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) Base.metadata.create_all(engine) Session = sessionmaker(bind=engine) session = Session() @@ -84,7 +89,7 @@ def test_pending_event_is_published( updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "processing" assert updated.claimed_at is not None - mock_delay.assert_called_once_with(str(event.id)) + mock_delay.delay.assert_called_once_with(str(event.id)) @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -110,7 +115,7 @@ def test_failed_event_is_requeued( updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "processing" assert updated.claimed_at is not None - mock_delay.assert_called_once_with(str(event.id)) + mock_delay.delay.assert_called_once_with(str(event.id)) @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -137,7 +142,7 @@ def test_stale_processing_event_is_reclaimed( updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "processing" assert updated.claimed_at is not None - mock_delay.assert_called_once_with(str(event.id)) + mock_delay.delay.assert_called_once_with(str(event.id)) @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -163,7 +168,7 @@ def test_recent_processing_event_is_not_reclaimed( updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "processing" - mock_delay.assert_not_called() + mock_delay.delay.assert_not_called() @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -187,7 +192,7 @@ def test_event_exceeding_max_retries_is_skipped( updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "pending" - mock_delay.assert_not_called() + mock_delay.delay.assert_not_called() @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -229,7 +234,7 @@ def patched_update(*uargs, **ukwargs): relay = OutboxRelay(max_retries=5) relay.process_pending_events() - mock_delay.assert_not_called() + mock_delay.delay.assert_not_called() # --------------------------------------------------------------------------- @@ -239,17 +244,13 @@ def patched_update(*uargs, **ukwargs): class TestProcessPositionOpenedTask: @patch("web_app.tasks.outbox_relay.init_db") def test_invalid_uuid_returns_early(self, mock_init_db): - task = MagicMock() - process_position_opened_task(task, event_id="not-a-uuid") - task.retry.assert_not_called() + process_position_opened_task.run(event_id="not-a-uuid") @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.init_db") def test_valid_uuid_not_found_returns_early(self, mock_init_db, mock_session_local, db_session): mock_session_local.return_value = db_session - task = MagicMock() - process_position_opened_task(task, event_id=str(uuid.uuid4())) - task.retry.assert_not_called() + process_position_opened_task.run(event_id=str(uuid.uuid4())) @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.init_db") @@ -265,7 +266,6 @@ def test_failed_event_clears_claimed_at(self, mock_init_db, mock_session_local, db_session.commit() mock_session_local.return_value = db_session - task = MagicMock() with patch("web_app.tasks.outbox_relay.DashboardMixin") as mock_dash: mock_dash.get_current_prices = MagicMock( side_effect=RuntimeError("pricing down") @@ -274,7 +274,7 @@ def test_failed_event_clears_claimed_at(self, mock_init_db, mock_session_local, instance = mock_pdbc.return_value instance.get_object.return_value = None with pytest.raises(Exception): - process_position_opened_task(task, event_id=str(event.id)) + process_position_opened_task.run(event_id=str(event.id)) updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "failed" From 956b319d06b3fd14d028368a71ad40949dfc78dd Mon Sep 17 00:00:00 2001 From: temiport25 Date: Fri, 21 Aug 2026 14:36:58 +0100 Subject: [PATCH 3/4] fix(tests): fix outbox relay test failures - Convert event_id to uuid.UUID in process_position_opened_task to fix SQLite UUID string comparison ('str' object has no attribute 'hex') - Use fresh sessions in tests that query after relay/task closes the original session (DetachedInstanceError) - Track .update() calls in test_relay_worker_dispatches_task to properly simulate synchronize_session='fetch' behavior --- quantara/web_app/tasks/outbox_relay.py | 6 ++++-- quantara/web_app/tests/test_outbox.py | 13 ++++++++++-- quantara/web_app/tests/test_outbox_relay.py | 23 ++++++++++++++------- 3 files changed, 31 insertions(+), 11 deletions(-) diff --git a/quantara/web_app/tasks/outbox_relay.py b/quantara/web_app/tasks/outbox_relay.py index 64034b2b..e6b1a951 100644 --- a/quantara/web_app/tasks/outbox_relay.py +++ b/quantara/web_app/tasks/outbox_relay.py @@ -55,11 +55,13 @@ def process_position_opened_task(self, event_id: str): logger.error("invalid_event_id_format", event_id=event_id) return + event_uuid = uuid.UUID(event_id) + init_db() db: Session = SessionLocal() try: # 1. Fetch the outbox event - event = db.query(OutboxEvent).filter(OutboxEvent.id == event_id).first() + event = db.query(OutboxEvent).filter(OutboxEvent.id == event_uuid).first() if not event: logger.error("outbox_event_not_found", event_id=event_id) return @@ -127,7 +129,7 @@ def process_position_opened_task(self, event_id: str): # Update event status to failed and increment retry try: with SessionLocal() as fail_session: - evt = fail_session.query(OutboxEvent).filter(OutboxEvent.id == event_id).first() + evt = fail_session.query(OutboxEvent).filter(OutboxEvent.id == event_uuid).first() if evt: evt.status = "failed" evt.retry_count += 1 diff --git a/quantara/web_app/tests/test_outbox.py b/quantara/web_app/tests/test_outbox.py index ffab4e1d..a28c1cd6 100644 --- a/quantara/web_app/tests/test_outbox.py +++ b/quantara/web_app/tests/test_outbox.py @@ -54,17 +54,26 @@ def test_relay_worker_dispatches_task(): mock_event.created_at = datetime.now() mock_db = MagicMock() - # Query returns old events (empty) then pending events (mock_event) mock_query = MagicMock() mock_db.query.return_value = mock_query mock_filter = MagicMock() mock_query.filter.return_value = mock_filter mock_filter.all.side_effect = [[], [mock_event]] + update_calls = [] + + def track_update(values, **kwargs): + update_calls.append(values) + for k, v in values.items(): + setattr(mock_event, k, v) + return 1 + + mock_filter.update.side_effect = track_update + with patch("web_app.tasks.outbox_relay.SessionLocal", return_value=mock_db), \ patch("web_app.tasks.outbox_relay.process_position_opened_task.delay") as mock_delay, \ patch("web_app.tasks.outbox_relay.sentry_sdk.capture_message") as mock_sentry: - + relay = OutboxRelay(max_retries=5) relay.process_pending_events() diff --git a/quantara/web_app/tests/test_outbox_relay.py b/quantara/web_app/tests/test_outbox_relay.py index 825aa900..e48e2fa9 100644 --- a/quantara/web_app/tests/test_outbox_relay.py +++ b/quantara/web_app/tests/test_outbox_relay.py @@ -34,6 +34,7 @@ def db_session(): Base.metadata.create_all(engine) Session = sessionmaker(bind=engine) session = Session() + session._session_factory = Session yield session session.close() @@ -86,10 +87,12 @@ def test_pending_event_is_published( relay = OutboxRelay(max_retries=5) relay.process_pending_events() - updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + fresh = db_session._session_factory() + updated = fresh.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "processing" assert updated.claimed_at is not None mock_delay.delay.assert_called_once_with(str(event.id)) + fresh.close() @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -112,10 +115,12 @@ def test_failed_event_is_requeued( relay = OutboxRelay(max_retries=5) relay.process_pending_events() - updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + fresh = db_session._session_factory() + updated = fresh.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "processing" assert updated.claimed_at is not None mock_delay.delay.assert_called_once_with(str(event.id)) + fresh.close() @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -139,10 +144,12 @@ def test_stale_processing_event_is_reclaimed( relay = OutboxRelay(max_retries=5) relay.process_pending_events() - updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + fresh = db_session._session_factory() + updated = fresh.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "processing" assert updated.claimed_at is not None mock_delay.delay.assert_called_once_with(str(event.id)) + fresh.close() @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -166,8 +173,6 @@ def test_recent_processing_event_is_not_reclaimed( relay = OutboxRelay(max_retries=5) relay.process_pending_events() - updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() - assert updated.status == "processing" mock_delay.delay.assert_not_called() @patch("web_app.tasks.outbox_relay.SessionLocal") @@ -190,9 +195,11 @@ def test_event_exceeding_max_retries_is_skipped( relay = OutboxRelay(max_retries=5) relay.process_pending_events() - updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + fresh = db_session._session_factory() + updated = fresh.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "pending" mock_delay.delay.assert_not_called() + fresh.close() @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -276,7 +283,9 @@ def test_failed_event_clears_claimed_at(self, mock_init_db, mock_session_local, with pytest.raises(Exception): process_position_opened_task.run(event_id=str(event.id)) - updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + fresh = db_session._session_factory() + updated = fresh.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "failed" assert updated.claimed_at is None assert updated.retry_count == 1 + fresh.close() From 262a752222777e5d577305a0dfd76293ab630849 Mon Sep 17 00:00:00 2001 From: portableDD Date: Fri, 21 Aug 2026 14:50:29 +0100 Subject: [PATCH 4/4] fix(tests): stub session.close() to prevent DetachedInstanceError The relay's finally block closes the session, invalidating ORM objects for post-relay assertions. Stubbing close() as a no-op keeps the session usable while preserving cleanup in the fixture teardown. --- quantara/web_app/tests/test_outbox_relay.py | 25 ++++++++------------- 1 file changed, 9 insertions(+), 16 deletions(-) diff --git a/quantara/web_app/tests/test_outbox_relay.py b/quantara/web_app/tests/test_outbox_relay.py index e48e2fa9..3fb3c4f9 100644 --- a/quantara/web_app/tests/test_outbox_relay.py +++ b/quantara/web_app/tests/test_outbox_relay.py @@ -34,8 +34,9 @@ def db_session(): Base.metadata.create_all(engine) Session = sessionmaker(bind=engine) session = Session() - session._session_factory = Session + session.close = lambda: None yield session + session.close = session.__class__.close.__get__(session) session.close() @@ -87,12 +88,10 @@ def test_pending_event_is_published( relay = OutboxRelay(max_retries=5) relay.process_pending_events() - fresh = db_session._session_factory() - updated = fresh.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "processing" assert updated.claimed_at is not None mock_delay.delay.assert_called_once_with(str(event.id)) - fresh.close() @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -115,12 +114,10 @@ def test_failed_event_is_requeued( relay = OutboxRelay(max_retries=5) relay.process_pending_events() - fresh = db_session._session_factory() - updated = fresh.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "processing" assert updated.claimed_at is not None mock_delay.delay.assert_called_once_with(str(event.id)) - fresh.close() @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -144,12 +141,10 @@ def test_stale_processing_event_is_reclaimed( relay = OutboxRelay(max_retries=5) relay.process_pending_events() - fresh = db_session._session_factory() - updated = fresh.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "processing" assert updated.claimed_at is not None mock_delay.delay.assert_called_once_with(str(event.id)) - fresh.close() @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -173,6 +168,8 @@ def test_recent_processing_event_is_not_reclaimed( relay = OutboxRelay(max_retries=5) relay.process_pending_events() + updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + assert updated.status == "processing" mock_delay.delay.assert_not_called() @patch("web_app.tasks.outbox_relay.SessionLocal") @@ -195,11 +192,9 @@ def test_event_exceeding_max_retries_is_skipped( relay = OutboxRelay(max_retries=5) relay.process_pending_events() - fresh = db_session._session_factory() - updated = fresh.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "pending" mock_delay.delay.assert_not_called() - fresh.close() @patch("web_app.tasks.outbox_relay.SessionLocal") @patch("web_app.tasks.outbox_relay.process_position_opened_task") @@ -283,9 +278,7 @@ def test_failed_event_clears_claimed_at(self, mock_init_db, mock_session_local, with pytest.raises(Exception): process_position_opened_task.run(event_id=str(event.id)) - fresh = db_session._session_factory() - updated = fresh.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() + updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() assert updated.status == "failed" assert updated.claimed_at is None assert updated.retry_count == 1 - fresh.close()