feat: implement background health check for Inference API with Celery queue management

Author: Avneesh11905

inference_api/presentation/api/main.py

@@ -18,7 +18,6 @@
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    # Any necessary init happens via Dependency Injector singletons if needed
 
     main_api_url = settings.MAIN_API_URL
     webhook_secret = settings.WEBHOOK_SECRET
@@ -43,11 +42,12 @@ async def fire_online_webhook_with_retry():
                 logger.error(f"Failed to send online webhook: {e}. Retrying in 5 seconds...")
                 await asyncio.sleep(5)
 
-    # Run the robust webhook logic in the background so we don't block server startup
-    asyncio.create_task(fire_online_webhook_with_retry())
+    task = asyncio.create_task(fire_online_webhook_with_retry())
 
     yield
 
+    task.cancel()
+
 app = FastAPI(
     title="Eventsnap Inference API",
     description="Local testing server for HF Endpoint Handler",
@@ -64,6 +64,10 @@ async def fire_online_webhook_with_retry():
 
 app.include_router(inference.router)
 
+@app.get("/health", tags=["Health"])
+async def health_check():
+    return {"status": "ok"}
+
 if __name__ == "__main__":
     port = 5000
     logger.info("=" * 50)

main_api/presentation/api/main.py

@@ -9,15 +9,51 @@
 from sqlalchemy import text
 from infrastructure.queue.celery_app import celery_app
 
-# Create the container at startup to wire everything
-container = get_container()
+import asyncio
+import httpx
+import logging
+from config import settings
 
+container = get_container()
+logger = logging.getLogger(__name__)
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     # Any necessary init happens via Dependency Injector singletons if needed
+    
+    async def poll_inference_api():
+        consecutive_failures = 0
+        current_state = "online" 
+        
+        while True:
+            try:
+                async with httpx.AsyncClient() as client:
+                    response = await client.get(f"{settings.INFERENCE_API_URL.rstrip('/')}/health", timeout=5.0)
+                    response.raise_for_status()
+                    
+                consecutive_failures = 0
+                if current_state == "offline":
+                    logger.info("Poller: Inference API is back online! Resuming Celery queue...")
+                    await asyncio.to_thread(celery_app.control.add_consumer, 'celery', reply=True)
+                    current_state = "online"
+                    
+            except Exception as e:
+                consecutive_failures += 1
+                if consecutive_failures >= 3 and current_state == "online":
+                    logger.warning(f"Poller: Inference API health check failed {consecutive_failures} times! Pausing Celery queue...")
+                    await asyncio.to_thread(celery_app.control.cancel_consumer, 'celery', reply=True)
+                    current_state = "offline"
+                    
+            # Wait 10 seconds before next ping
+            await asyncio.sleep(10)
+            
+    # Start the background task
+    task = asyncio.create_task(poll_inference_api())
+    
     yield
-
+    
+    # Clean up on shutdown
+    task.cancel()
 
 app = FastAPI(
     title="Eventsnap Main API (Orchestrator) - Clean",