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..e6b1a951 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,12 +50,18 @@ 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 + + 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 @@ -107,15 +125,16 @@ 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: - 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 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 +152,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 +171,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 +219,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.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 new file mode 100644 index 00000000..3fb3c4f9 --- /dev/null +++ b/quantara/web_app/tests/test_outbox_relay.py @@ -0,0 +1,284 @@ +""" +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 sqlalchemy.pool import StaticPool + +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:", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, + ) + Base.metadata.create_all(engine) + Session = sessionmaker(bind=engine) + session = Session() + session.close = lambda: None + yield session + session.close = session.__class__.close.__get__(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.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.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.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.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.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.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): + 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 + 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") + 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 + + 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.run(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