fix: improve service resilience for restarts and startup

- Add processing state tracking to extractor for restart recovery
  - Tracks which files have been fully processed in .processing_state.json
  - Automatically resumes processing incomplete files after restart
  - Support FORCE_REPROCESS=true to reprocess all files

- Fix RabbitMQ connection issues during startup
  - Increase circuit breaker thresholds (3→5 failures, 30s→60s recovery)
  - Add configurable STARTUP_DELAY for all services
  - Implement additional startup retry logic with exponential backoff
  - Services retry up to 5 times over ~75 seconds before giving up

- Apply fixes to all message-consuming services
  - tableinator: Fixed connection failures and circuit breaker issues
  - graphinator: Added same resilience improvements
  - Both services now handle RabbitMQ unavailability gracefully

This ensures the system recovers properly from container restarts and
handles service startup ordering issues without excessive error logs.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: SimplicityGuy

common/rabbitmq_resilient.py

@@ -122,11 +122,12 @@ def __init__(self, connection_url: str, max_retries: int = 5, heartbeat: int = 6
         self._lock = asyncio.Lock()
 
         # Circuit breaker for RabbitMQ failures
+        # Use higher threshold and longer recovery for startup scenarios
         self.circuit_breaker = CircuitBreaker(
             CircuitBreakerConfig(
                 name="AsyncRabbitMQ",
-                failure_threshold=3,
-                recovery_timeout=30,
+                failure_threshold=5,  # Allow more attempts before opening
+                recovery_timeout=60,  # Give more time for RabbitMQ to start
                 expected_exception=(AMQPConnectionError, AMQPChannelError, ConnectionClosed),
             )
         )

extractor/extractor.py

@@ -1,6 +1,7 @@
 import asyncio
 import contextlib
 import logging
+import os
 import signal
 import sys
 import threading
@@ -654,6 +655,33 @@ def _flush_pending_messages(self) -> None:
             logger.warning("⚠️ Messages will be retried on next flush")
 
 
+def _load_processing_state(discogs_root: Path) -> dict[str, bool]:
+    """Load processing state from file."""
+    state_file = Path(discogs_root) / ".processing_state.json"
+    if not state_file.exists():
+        return {}
+
+    try:
+        with state_file.open("rb") as f:
+            data = loads(f.read())
+            # Ensure we return the correct type
+            return {k: bool(v) for k, v in data.items()}
+    except Exception as e:
+        logger.warning(f"⚠️ Failed to load processing state: {e}")
+        return {}
+
+
+def _save_processing_state(discogs_root: Path, state: dict[str, bool]) -> None:
+    """Save processing state to file."""
+    state_file = Path(discogs_root) / ".processing_state.json"
+
+    try:
+        with state_file.open("wb") as f:
+            f.write(dumps(state, option=OPT_INDENT_2))
+    except Exception as e:
+        logger.warning(f"⚠️ Failed to save processing state: {e}")
+
+
 async def process_file_async(discogs_data_file: str, config: ExtractorConfig) -> None:
     """Process a single file asynchronously."""
     try:
@@ -691,6 +719,27 @@ async def process_discogs_data(config: ExtractorConfig) -> bool:
         logger.warning("⚠️ No data files to process")
         return True
 
+    # Check processing state to handle restarts
+    processing_state = _load_processing_state(config.discogs_root)
+    files_to_process = []
+
+    for data_file in data_files:
+        if data_file not in processing_state or not processing_state[data_file]:
+            files_to_process.append(data_file)
+            logger.info(f"📋 Will process: {data_file}")
+        else:
+            logger.info(f"✅ Already processed: {data_file}")
+
+    if not files_to_process:
+        logger.info("✅ All files have been processed previously")
+        # Force reprocessing if explicitly requested via environment variable
+        if os.environ.get("FORCE_REPROCESS", "").lower() == "true":
+            logger.info("🔄 FORCE_REPROCESS=true, will reprocess all files")
+            files_to_process = data_files
+            processing_state = {}  # Clear processing state
+        else:
+            return True
+
     # Process files concurrently with a semaphore to limit concurrent files
     semaphore = asyncio.Semaphore(3)  # Limit to 3 concurrent files
     tasks = []
@@ -700,8 +749,12 @@ async def process_with_semaphore(file: str) -> None:
             if shutdown_requested:
                 return
             await process_file_async(file, config)
+            # Mark file as processed
+            processing_state[file] = True
+            _save_processing_state(config.discogs_root, processing_state)
+            logger.info(f"✅ Marked {file} as processed")
 
-    for data_file in data_files:
+    for data_file in files_to_process:
         if shutdown_requested:
             break
         tasks.append(asyncio.create_task(process_with_semaphore(data_file)))
@@ -882,6 +935,8 @@ async def main_async() -> None:
     logger.info(
         "🚀 Starting Discogs data extractor with concurrent processing and periodic checks"
     )
+    logger.info("📋 Will check for incomplete processing from previous runs")
+    logger.info("💡 Tip: Set FORCE_REPROCESS=true to reprocess all files")
 
     # Start health server
     health_server = HealthServer(8000, get_health_data)

graphinator/graphinator.py

@@ -1,6 +1,7 @@
 import asyncio
 import contextlib
 import logging
+import os
 import signal
 import time
 from asyncio import run
@@ -812,6 +813,14 @@ async def main() -> None:
     setup_logging("graphinator", log_file=Path("/logs/graphinator.log"))
     logger.info("🚀 Starting Neo4j graphinator service")
 
+    # Add startup delay for dependent services
+    startup_delay = int(os.environ.get("STARTUP_DELAY", "5"))
+    if startup_delay > 0:
+        logger.info(
+            f"⏳ Waiting {startup_delay} seconds for dependent services to start..."
+        )
+        await asyncio.sleep(startup_delay)
+
     # Start health server
     health_server = HealthServer(8001, get_health_data)
     health_server.start_background()
@@ -882,15 +891,40 @@ async def main() -> None:
     # Initialize resilient RabbitMQ connection
     rabbitmq = AsyncResilientRabbitMQ(
         connection_url=config.amqp_connection,
+        max_retries=10,  # More retries for startup
         heartbeat=600,
         connection_attempts=10,
         retry_delay=5.0,
     )
 
-    try:
-        amqp_connection = await rabbitmq.connect()
-    except Exception as e:
-        logger.error(f"❌ Failed to connect to AMQP broker: {e}")
+    # Try to connect with additional retry logic for startup
+    max_startup_retries = 5
+    startup_retry = 0
+    amqp_connection = None
+
+    while startup_retry < max_startup_retries and not shutdown_requested:
+        try:
+            logger.info(
+                f"🐰 Attempting to connect to RabbitMQ (attempt {startup_retry + 1}/{max_startup_retries})"
+            )
+            amqp_connection = await rabbitmq.connect()
+            break
+        except Exception as e:
+            startup_retry += 1
+            if startup_retry < max_startup_retries:
+                wait_time = min(30, 5 * startup_retry)  # Exponential backoff up to 30s
+                logger.warning(
+                    f"⚠️ RabbitMQ connection failed: {e}. Retrying in {wait_time} seconds..."
+                )
+                await asyncio.sleep(wait_time)
+            else:
+                logger.error(
+                    f"❌ Failed to connect to AMQP broker after {max_startup_retries} attempts: {e}"
+                )
+                return
+
+    if amqp_connection is None:
+        logger.error("❌ No AMQP connection available")
         return
 
     async with amqp_connection:

tableinator/tableinator.py

@@ -1,6 +1,7 @@
 import asyncio
 import contextlib
 import logging
+import os
 import signal
 import time
 from asyncio import run
@@ -197,6 +198,14 @@ async def main() -> None:
     setup_logging("tableinator", log_file=Path("/logs/tableinator.log"))
     logger.info("🚀 Starting PostgreSQL tableinator service with connection pooling")
 
+    # Add startup delay for dependent services
+    startup_delay = int(os.environ.get("STARTUP_DELAY", "5"))
+    if startup_delay > 0:
+        logger.info(
+            f"⏳ Waiting {startup_delay} seconds for dependent services to start..."
+        )
+        await asyncio.sleep(startup_delay)
+
     # Start health server
     health_server = HealthServer(8002, get_health_data)
     health_server.start_background()
@@ -316,15 +325,40 @@ async def main() -> None:
     # Initialize resilient RabbitMQ connection
     rabbitmq = AsyncResilientRabbitMQ(
         connection_url=config.amqp_connection,
+        max_retries=10,  # More retries for startup
         heartbeat=600,
         connection_attempts=10,
         retry_delay=5.0,
     )
 
-    try:
-        amqp_connection = await rabbitmq.connect()
-    except Exception as e:
-        logger.error(f"❌ Failed to connect to AMQP broker: {e}")
+    # Try to connect with additional retry logic for startup
+    max_startup_retries = 5
+    startup_retry = 0
+    amqp_connection = None
+
+    while startup_retry < max_startup_retries and not shutdown_requested:
+        try:
+            logger.info(
+                f"🐰 Attempting to connect to RabbitMQ (attempt {startup_retry + 1}/{max_startup_retries})"
+            )
+            amqp_connection = await rabbitmq.connect()
+            break
+        except Exception as e:
+            startup_retry += 1
+            if startup_retry < max_startup_retries:
+                wait_time = min(30, 5 * startup_retry)  # Exponential backoff up to 30s
+                logger.warning(
+                    f"⚠️ RabbitMQ connection failed: {e}. Retrying in {wait_time} seconds..."
+                )
+                await asyncio.sleep(wait_time)
+            else:
+                logger.error(
+                    f"❌ Failed to connect to AMQP broker after {max_startup_retries} attempts: {e}"
+                )
+                return
+
+    if amqp_connection is None:
+        logger.error("❌ No AMQP connection available")
         return
 
     async with amqp_connection: