Pipeline resilience: try/finally + startup recovery for stuck images

- process_image_pipeline: wrapped in try/finally to guarantee processing_status
  is always updated, even on unexpected crashes/timeouts
- init_db: recovery query that marks images stuck in 'processing' for >10min
  as ERROR on startup
- Clean rewrite of pipeline.py for consistent indentation

app/database.py

@@ -2,8 +2,10 @@
 Configuration SQLAlchemy — session async
 """
 import logging
+from datetime import datetime, timezone, timedelta
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
 from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy import select, text
 from app.config import settings
 
 logger = logging.getLogger(__name__)
@@ -44,9 +46,9 @@ async def get_db() -> AsyncSession:
 async def init_db():
     """Crée toutes les tables, applique les migrations, et initialise un client par défaut si nécessaire."""
     import secrets
-    from sqlalchemy import select, text
     from app.dependencies.auth import hash_api_key
     from app.models.client import APIClient, ClientPlan
+    from app.models.image import Image, ProcessingStatus
 
     async with engine.begin() as conn:
         from app.models import image  # noqa: F401
@@ -59,7 +61,7 @@ async def init_db():
                 "ALTER TABLE api_clients ADD COLUMN IF NOT EXISTS ai_provider VARCHAR(64)"
             ))
         except Exception:
-            pass  # SQLite ne supporte pas ADD COLUMN IF NOT EXISTS
+            pass
 
         try:
             await conn.execute(text(
@@ -68,8 +70,27 @@ async def init_db():
         except Exception:
             pass
 
-    # Vérifier s'il y a déjà des clients
+    # Vérifier s'il y a déjà des clients + recovery des images bloquées
     async with AsyncSessionLocal() as session:
+        # Recovery : marquer les images bloquées en "processing" depuis > 10 min comme ERROR
+        cutoff = datetime.now(timezone.utc) - timedelta(minutes=10)
+        stuck_query = select(Image).where(
+            Image.processing_status == ProcessingStatus.PROCESSING,
+            Image.processing_started_at < cutoff,
+        )
+        stuck_result = await session.execute(stuck_query)
+        stuck_images = stuck_result.scalars().all()
+        for img in stuck_images:
+            img.processing_status = ProcessingStatus.ERROR
+            img.processing_error = "[Recovery] Pipeline interrompu (timeout ou crash worker)"
+            img.processing_done_at = datetime.now(timezone.utc)
+        if stuck_images:
+            await session.commit()
+            logger.warning(
+                "init_db.recovery",
+                extra={"stuck_images_recovered": len(stuck_images)},
+            )
+
         result = await session.execute(select(APIClient).limit(1))
         if result.scalar_one_or_none() is None:
             # Table vide -> Création du client bootstrap

app/services/pipeline.py

@@ -21,7 +21,6 @@ from app.metrics import (
     hub_pipeline_duration, hub_pipeline_step_duration, hub_pipeline_errors,
     hub_ai_tokens_consumed
 )
-import asyncio
 
 logger = logging.getLogger(__name__)
 
@@ -44,20 +43,17 @@ async def _publish_event(
             payload["data"] = data
         raw = json.dumps(payload)
 
-        # Publier sur le channel dédié à l'image
         await redis.publish(f"pipeline:{image_id}", raw)
 
-        # Buffer de reconnexion (10 derniers événements, TTL 60s)
         buffer_key = f"pipeline:buffer:{image_id}"
         await redis.rpush(buffer_key, raw)
         await redis.ltrim(buffer_key, -10, -1)
         await redis.expire(buffer_key, 60)
 
-        # Publier sur le channel admin pour le monitoring global
         if event in ("pipeline.started", "pipeline.done", "pipeline.error"):
             await redis.publish("pipeline:admin", raw)
     except Exception:
-        pass  # Pub/Sub non critique — ne doit pas bloquer le pipeline
+        pass
 
 
 async def process_image_pipeline(
@@ -70,10 +66,9 @@ async def process_image_pipeline(
       3. Vision AI — description + tags (async)
       4. Sauvegarde finale en BDD
 
-    Le statut est mis à jour à chaque étape pour permettre le polling.
-    Publie des événements Redis sur le channel pipeline:{image_id}.
+    Try/finally garantit que le statut est TOUJOURS mis à jour,
+    même en cas de crash inattendu (timeout, OOM, etc.).
     """
-    # ── Chargement de l'image ─────────────────────────────────
     result = await db.execute(select(Image).where(Image.id == image_id))
     image = result.scalar_one_or_none()
 
@@ -81,7 +76,6 @@ async def process_image_pipeline(
         logger.warning("pipeline.image_not_found", extra={"image_id": image_id})
         return
 
-    # ── Démarrage ─────────────────────────────────────────────
     image.processing_status = ProcessingStatus.PROCESSING
     image.processing_started_at = datetime.now(timezone.utc)
     t_start = time.time()
@@ -93,172 +87,176 @@ async def process_image_pipeline(
     errors: list[str] = []
     file_path = image.file_path
 
-    # ════════════════════════════════════════════════════════════
-    # ÉTAPE 1 — Extraction EXIF
-    # ════════════════════════════════════════════════════════════
     try:
-        logger.info("pipeline.step.start", extra={"image_id": image_id, "step": "exif", "step_num": "1/3"})
-        t0 = time.time()
-        
-        # Maintenant async et utilise le backend
-        exif = await extract_exif(file_path)
+        # ════════════════════════════════════════════════════════
+        # ÉTAPE 1 — Extraction EXIF
+        # ════════════════════════════════════════════════════════
+        try:
+            logger.info("pipeline.step.start", extra={"image_id": image_id, "step": "exif", "step_num": "1/3"})
+            t0 = time.time()
 
-        image.exif_raw         = exif.get("raw")
-        image.exif_make        = exif.get("make")
-        image.exif_model       = exif.get("model")
-        image.exif_lens        = exif.get("lens")
-        image.exif_taken_at    = exif.get("taken_at")
-        image.exif_gps_lat     = exif.get("gps_lat")
-        image.exif_gps_lon     = exif.get("gps_lon")
-        image.exif_altitude    = exif.get("altitude")
-        image.exif_iso         = exif.get("iso")
-        image.exif_aperture    = exif.get("aperture")
-        image.exif_shutter     = exif.get("shutter")
-        image.exif_focal       = exif.get("focal")
-        image.exif_flash       = exif.get("flash")
-        image.exif_orientation = exif.get("orientation")
-        image.exif_software    = exif.get("software")
+            exif = await extract_exif(file_path)
 
-        await db.commit()
-        elapsed = int((time.time() - t0) * 1000)
-        hub_pipeline_step_duration.labels(step="exif").observe(time.time() - t0)
-        
-        logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "exif", "duration_ms": elapsed, "camera": image.exif_make})
-        await _publish_event(redis, image_id, "step.completed", {
-            "step": "exif", "duration_ms": elapsed, "camera": image.exif_make,
-        })
+            image.exif_raw         = exif.get("raw")
+            image.exif_make        = exif.get("make")
+            image.exif_model       = exif.get("model")
+            image.exif_lens        = exif.get("lens")
+            image.exif_taken_at    = exif.get("taken_at")
+            image.exif_gps_lat     = exif.get("gps_lat")
+            image.exif_gps_lon     = exif.get("gps_lon")
+            image.exif_altitude    = exif.get("altitude")
+            image.exif_iso         = exif.get("iso")
+            image.exif_aperture    = exif.get("aperture")
+            image.exif_shutter     = exif.get("shutter")
+            image.exif_focal       = exif.get("focal")
+            image.exif_flash       = exif.get("flash")
+            image.exif_orientation = exif.get("orientation")
+            image.exif_software    = exif.get("software")
 
-    except Exception as e:
-        msg = f"EXIF : {str(e)}"
-        errors.append(msg)
-        hub_pipeline_errors.labels(step="exif").inc()
-        logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "exif", "error": str(e)})
+            await db.commit()
+            elapsed = int((time.time() - t0) * 1000)
+            hub_pipeline_step_duration.labels(step="exif").observe(time.time() - t0)
 
-    # ════════════════════════════════════════════════════════════
-    # ÉTAPE 2 — OCR
-    # ════════════════════════════════════════════════════════════
-    try:
-        logger.info("pipeline.step.start", extra={"image_id": image_id, "step": "ocr", "step_num": "2/3"})
-        t0 = time.time()
-        
-        # Maintenant async et utilise le backend
-        ocr = await extract_text(file_path)
+            logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "exif", "duration_ms": elapsed, "camera": image.exif_make})
+            await _publish_event(redis, image_id, "step.completed", {
+                "step": "exif", "duration_ms": elapsed, "camera": image.exif_make,
+            })
 
-        # Fallback AI si OCR classique échoue ou ne trouve rien
-        if not ocr.get("has_text", False):
-            logger.info("pipeline.ocr.fallback", extra={"image_id": image_id, "reason": "tesseract_empty"})
-            ai_ocr = await extract_text_with_ai(file_path)
+        except Exception as e:
+            msg = f"EXIF : {str(e)}"
+            errors.append(msg)
+            hub_pipeline_errors.labels(step="exif").inc()
+            logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "exif", "error": str(e)})
 
-            if ai_ocr.get("has_text"):
-                ocr = ai_ocr
-                logger.info("pipeline.ocr.fallback_success", extra={"image_id": image_id, "chars": len(ocr.get("text", ""))})
-            else:
-                logger.info("pipeline.ocr.fallback_empty", extra={"image_id": image_id})
+        # ════════════════════════════════════════════════════════
+        # ÉTAPE 2 — OCR
+        # ════════════════════════════════════════════════════════
+        try:
+            logger.info("pipeline.step.start", extra={"image_id": image_id, "step": "ocr", "step_num": "2/3"})
+            t0 = time.time()
 
-        image.ocr_text       = ocr.get("text")
-        image.ocr_language   = ocr.get("language")
-        image.ocr_confidence = ocr.get("confidence")
-        image.ocr_has_text   = ocr.get("has_text", False)
+            ocr = await extract_text(file_path)
 
-        await db.commit()
-        elapsed = int((time.time() - t0) * 1000)
-        hub_pipeline_step_duration.labels(step="ocr").observe(time.time() - t0)
+            if not ocr.get("has_text", False):
+                logger.info("pipeline.ocr.fallback", extra={"image_id": image_id, "reason": "tesseract_empty"})
+                ai_ocr = await extract_text_with_ai(file_path)
 
-        logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "ocr", "duration_ms": elapsed, "has_text": image.ocr_has_text})
-        await _publish_event(redis, image_id, "step.completed", {
-            "step": "ocr", "duration_ms": elapsed, "has_text": image.ocr_has_text,
-        })
+                if ai_ocr.get("has_text"):
+                    ocr = ai_ocr
+                    logger.info("pipeline.ocr.fallback_success", extra={"image_id": image_id, "chars": len(ocr.get("text", ""))})
+                else:
+                    logger.info("pipeline.ocr.fallback_empty", extra={"image_id": image_id})
 
-    except Exception as e:
-        msg = f"OCR : {str(e)}"
-        errors.append(msg)
-        hub_pipeline_errors.labels(step="ocr").inc()
-        logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "ocr", "error": str(e)})
+            image.ocr_text       = ocr.get("text")
+            image.ocr_language   = ocr.get("language")
+            image.ocr_confidence = ocr.get("confidence")
+            image.ocr_has_text   = ocr.get("has_text", False)
 
-    # ════════════════════════════════════════════════════════════
-    # ÉTAPE 3 — Vision AI (description + tags)
-    # ════════════════════════════════════════════════════════════
-    try:
-        logger.info("pipeline.step.start", extra={"image_id": image_id, "step": "ai", "step_num": "3/3"})
-        t0 = time.time()
+            await db.commit()
+            elapsed = int((time.time() - t0) * 1000)
+            hub_pipeline_step_duration.labels(step="ocr").observe(time.time() - t0)
 
-        # Charger les préférences AI du client
-        provider_override = None
-        model_override = None
-        if image.client_id:
-            from app.models.client import APIClient
-            client_result = await db.execute(
-                select(APIClient).where(APIClient.id == image.client_id)
+            logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "ocr", "duration_ms": elapsed, "has_text": image.ocr_has_text})
+            await _publish_event(redis, image_id, "step.completed", {
+                "step": "ocr", "duration_ms": elapsed, "has_text": image.ocr_has_text,
+            })
+
+        except Exception as e:
+            msg = f"OCR : {str(e)}"
+            errors.append(msg)
+            hub_pipeline_errors.labels(step="ocr").inc()
+            logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "ocr", "error": str(e)})
+
+        # ════════════════════════════════════════════════════════
+        # ÉTAPE 3 — Vision AI (description + tags)
+        # ════════════════════════════════════════════════════════
+        try:
+            logger.info("pipeline.step.start", extra={"image_id": image_id, "step": "ai", "step_num": "3/3"})
+            t0 = time.time()
+
+            provider_override = None
+            model_override = None
+            if image.client_id:
+                from app.models.client import APIClient
+                client_result = await db.execute(
+                    select(APIClient).where(APIClient.id == image.client_id)
+                )
+                owner = client_result.scalar_one_or_none()
+                if owner:
+                    provider_override = owner.ai_provider
+                    model_override = owner.ai_model
+
+            ai = await analyze_image(
+                file_path=file_path,
+                ocr_hint=image.ocr_text,
+                provider_override=provider_override,
+                model_override=model_override,
             )
-            owner = client_result.scalar_one_or_none()
-            if owner:
-                provider_override = owner.ai_provider
-                model_override = owner.ai_model
 
-        ai = await analyze_image(
-            file_path=file_path,
-            ocr_hint=image.ocr_text,
-            provider_override=provider_override,
-            model_override=model_override,
-        )
+            image.ai_description   = ai.get("description")
+            image.ai_tags          = ai.get("tags", [])
+            image.ai_confidence    = ai.get("confidence")
+            image.ai_model_used    = ai.get("model")
+            image.ai_processed_at  = datetime.now(timezone.utc)
+            image.ai_prompt_tokens = ai.get("prompt_tokens")
+            image.ai_output_tokens = ai.get("output_tokens")
 
-        image.ai_description   = ai.get("description")
-        image.ai_tags          = ai.get("tags", [])
-        image.ai_confidence    = ai.get("confidence")
-        image.ai_model_used    = ai.get("model")
-        image.ai_processed_at  = datetime.now(timezone.utc)
-        image.ai_prompt_tokens = ai.get("prompt_tokens")
-        image.ai_output_tokens = ai.get("output_tokens")
+            await db.commit()
+            elapsed = int((time.time() - t0) * 1000)
+            hub_pipeline_step_duration.labels(step="ai").observe(time.time() - t0)
 
-        await db.commit()
-        elapsed = int((time.time() - t0) * 1000)
-        hub_pipeline_step_duration.labels(step="ai").observe(time.time() - t0)
-        
-        # Token metrics
-        if image.ai_prompt_tokens or image.ai_output_tokens:
-            total_tokens = (image.ai_prompt_tokens or 0) + (image.ai_output_tokens or 0)
-            hub_ai_tokens_consumed.labels(
-                client_id=image.client_id, 
-                model=image.ai_model_used or "unknown"
-            ).inc(total_tokens)
+            if image.ai_prompt_tokens or image.ai_output_tokens:
+                total_tokens = (image.ai_prompt_tokens or 0) + (image.ai_output_tokens or 0)
+                hub_ai_tokens_consumed.labels(
+                    client_id=image.client_id,
+                    model=image.ai_model_used or "unknown"
+                ).inc(total_tokens)
 
-        logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "ai", "duration_ms": elapsed, "tags_count": len(image.ai_tags or [])})
-        await _publish_event(redis, image_id, "step.completed", {
-            "step": "ai", "duration_ms": elapsed, "tags_count": len(image.ai_tags or []),
-        })
+            logger.info("pipeline.step.done", extra={"image_id": image_id, "step": "ai", "duration_ms": elapsed, "tags_count": len(image.ai_tags or [])})
+            await _publish_event(redis, image_id, "step.completed", {
+                "step": "ai", "duration_ms": elapsed, "tags_count": len(image.ai_tags or []),
+            })
 
-    except Exception as e:
-        msg = f"AI Vision : {str(e)}"
-        errors.append(msg)
-        hub_pipeline_errors.labels(step="ai").inc()
-        logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "ai", "error": str(e)})
+        except Exception as e:
+            msg = f"AI Vision : {str(e)}"
+            errors.append(msg)
+            hub_pipeline_errors.labels(step="ai").inc()
+            logger.error("pipeline.step.error", extra={"image_id": image_id, "step": "ai", "error": str(e)})
 
-    # ════════════════════════════════════════════════════════════
-    # FINALISATION
-    # ════════════════════════════════════════════════════════════
-    image.processing_done_at = datetime.now(timezone.utc)
+    finally:
+        # ════════════════════════════════════════════════════════
+        # FINALISATION — toujours exécuté, même après un crash
+        # ════════════════════════════════════════════════════════
+        try:
+            image.processing_done_at = datetime.now(timezone.utc)
 
-    if errors:
-        if image.ai_description:
-            image.processing_status = ProcessingStatus.DONE
-            image.processing_error = f"Avertissements : {'; '.join(errors)}"
-        else:
-            image.processing_status = ProcessingStatus.ERROR
-            image.processing_error = "; ".join(errors)
-    else:
-        image.processing_status = ProcessingStatus.DONE
-        image.processing_error = None
+            if errors:
+                if image.ai_description:
+                    image.processing_status = ProcessingStatus.DONE
+                    image.processing_error = f"Avertissements : {'; '.join(errors)}"
+                else:
+                    image.processing_status = ProcessingStatus.ERROR
+                    image.processing_error = "; ".join(errors)
+            else:
+                image.processing_status = ProcessingStatus.DONE
+                image.processing_error = None
 
-    await db.commit()
-    hub_pipeline_duration.observe(time.time() - t_start)
-    
-    logger.info("pipeline.completed", extra={
-        "image_id": image_id,
-        "status": image.processing_status.value,
-        "errors": len(errors),
-    })
+            await db.commit()
+            hub_pipeline_duration.observe(time.time() - t_start)
 
-    if errors:
-        await _publish_event(redis, image_id, "pipeline.error", {"errors": errors})
-    else:
-        await _publish_event(redis, image_id, "pipeline.done")
+            logger.info("pipeline.completed", extra={
+                "image_id": image_id,
+                "status": image.processing_status.value,
+                "errors": len(errors),
+            })
+
+            if errors:
+                await _publish_event(redis, image_id, "pipeline.error", {"errors": errors})
+            else:
+                await _publish_event(redis, image_id, "pipeline.done")
+        except Exception as final_error:
+            # Si même la finalisation échoue, log seulement — ne pas laisser l'image bloquée
+            logger.critical("pipeline.finalization_failed", extra={
+                "image_id": image_id,
+                "error": str(final_error),
+            })