Spaces:
Sleeping
Sleeping
| """ | |
| NeuralVault Distributed Outbox Worker & Dead-Letter Queue (DLQ) Processor | |
| Polls pending outbox_events, processes background tasks with exponential backoff, | |
| and migrates failed events to outbox_dlq after maximum retries. | |
| """ | |
| import json | |
| import time | |
| import asyncio | |
| import logging | |
| from typing import Dict, Any | |
| from core.state import postgres | |
| from embedding_service import embed_text | |
| logger = logging.getLogger("outbox_worker") | |
| async def ensure_dlq_tables(): | |
| if postgres.pool is None: | |
| return | |
| try: | |
| async with postgres.pool.acquire() as conn: | |
| await conn.execute(""" | |
| CREATE TABLE IF NOT EXISTS outbox_events ( | |
| id UUID PRIMARY KEY DEFAULT gen_random_uuid(), | |
| event_type VARCHAR(64) NOT NULL, | |
| payload JSONB NOT NULL, | |
| processed BOOLEAN DEFAULT false, | |
| retry_count INT DEFAULT 0, | |
| created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP | |
| ); | |
| CREATE TABLE IF NOT EXISTS outbox_dlq ( | |
| id UUID PRIMARY KEY DEFAULT gen_random_uuid(), | |
| original_event_id UUID NOT NULL, | |
| event_type VARCHAR(64) NOT NULL, | |
| payload JSONB NOT NULL, | |
| error_reason TEXT NOT NULL, | |
| failed_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP | |
| ); | |
| """) | |
| except Exception as exc: | |
| logger.warning(f"Failed to ensure DLQ tables: {exc}") | |
| async def process_event_with_retry(event_id: str, event_type: str, payload: dict, retry_count: int) -> bool: | |
| """Executes background task logic with retry tracking.""" | |
| try: | |
| if event_type == "REVIEW_REINDEX": | |
| text = payload.get("text", "") | |
| if text: | |
| await embed_text(text) | |
| elif event_type == "PRODUCT_REINDEX": | |
| text = f"{payload.get('name', '')} {payload.get('description', '')}" | |
| if text: | |
| await embed_text(text) | |
| return True | |
| except Exception as exc: | |
| logger.error(f"Event {event_id} failed attempt {retry_count + 1}: {exc}") | |
| return False | |
| async def run_worker_iteration() -> int: | |
| if postgres.pool is None: | |
| return 0 | |
| await ensure_dlq_tables() | |
| processed_count = 0 | |
| try: | |
| async with postgres.pool.acquire() as conn: | |
| records = await conn.fetch(""" | |
| SELECT id::text, event_type, payload, retry_count | |
| FROM outbox_events | |
| WHERE processed = false AND retry_count < 3 | |
| ORDER BY created_at ASC | |
| LIMIT 10 | |
| FOR UPDATE SKIP LOCKED; | |
| """) | |
| for record in records: | |
| event_id = record["id"] | |
| event_type = record["event_type"] | |
| payload = json.loads(record["payload"]) if isinstance(record["payload"], str) else record["payload"] | |
| retries = record["retry_count"] | |
| success = await process_event_with_retry(event_id, event_type, payload, retries) | |
| if success: | |
| await conn.execute("UPDATE outbox_events SET processed = true WHERE id = $1::uuid", event_id) | |
| processed_count += 1 | |
| else: | |
| new_retries = retries + 1 | |
| if new_retries >= 3: | |
| # Move to Dead-Letter Queue (DLQ) | |
| await conn.execute(""" | |
| INSERT INTO outbox_dlq (original_event_id, event_type, payload, error_reason) | |
| VALUES ($1::uuid, $2, $3, $4) | |
| """, event_id, event_type, json.dumps(payload), "Max retries (3) exceeded") | |
| await conn.execute("UPDATE outbox_events SET processed = true WHERE id = $1::uuid", event_id) | |
| logger.error(f"Moved event {event_id} to Dead-Letter Queue (DLQ)") | |
| else: | |
| # Backoff update | |
| await conn.execute("UPDATE outbox_events SET retry_count = $1 WHERE id = $2::uuid", new_retries, event_id) | |
| except Exception as exc: | |
| logger.error(f"Outbox worker iteration error: {exc}") | |
| return processed_count | |
| async def start_worker_daemon(loop_interval: float = 2.0): | |
| logger.info("Starting NeuralVault Outbox Worker Daemon...") | |
| while True: | |
| await run_worker_iteration() | |
| await asyncio.sleep(loop_interval) | |
| if __name__ == "__main__": | |
| asyncio.run(start_worker_daemon()) | |