fix(conserver): dead-letter failed storage writes instead of dropping them (CON-714)

api/api.py

@@ -43,7 +43,7 @@
 from starlette.status import HTTP_403_FORBIDDEN
 
 from config import Configuration
-from dlq_utils import get_ingress_list_dlq_name
+from dlq_utils import get_ingress_list_dlq_name, get_storage_dlq_name
 from lib.context_utils import store_context_async, extract_otel_trace_context
 from lib.logging_utils import init_logger
 from lib.metrics import increment_counter
@@ -1179,6 +1179,82 @@ async def post_dlq_reprocess(
         raise HTTPException(status_code=500, detail="Failed to reprocess DLQ")
 
 
+@api_router.post(
+    "/dlq/storage/reprocess",
+    status_code=200,
+    summary="Retry failed storage writes",
+    description="Re-attempt the storage write for vCons in a storage backend's DLQ",
+    tags=["dlq"],
+)
+async def post_storage_dlq_reprocess(
+    storage_name: str = Query(..., description="Name of the storage backend to retry"),
+    count: Annotated[
+        int,
+        Query(
+            ge=1,
+            le=100000,
+            description="Max items to retry in this call. Callers drive the loop client-side for large DLQs to avoid HTTP timeouts.",
+        ),
+    ] = 1000,
+) -> JSONResponse:
+    """Re-attempt up to ``count`` failed writes to a storage backend.
+
+    Only the storage write is replayed, not the chain that produced the vCon.
+    An item whose retry fails again goes back on the DLQ, so a backend that is
+    still down does not drain the queue into nothing. Draining stops at the
+    first such failure rather than spinning through every item.
+
+    Returns the number of vCons successfully written.
+    """
+    try:
+        storage = Storage(storage_name=storage_name)
+    except Exception as e:
+        logger.error(f"Unknown storage backend {storage_name}: {str(e)}")
+        raise HTTPException(status_code=404, detail=f"Unknown storage backend: {storage_name}")
+
+    succeeded = 0
+    try:
+        for _ in range(count):
+            vcon_id = await queue.dequeue_storage_dlq_async(redis_async, storage_name)
+            if vcon_id is None:
+                break
+            try:
+                storage.save(vcon_id)
+                succeeded += 1
+            except Exception as e:
+                # Put it back and stop: the backend is still unhealthy, and
+                # popping the rest would only re-queue them one at a time.
+                logger.warning(
+                    f"Storage DLQ retry failed for vCon {vcon_id} on {storage_name}: {e}"
+                )
+                queue.enqueue_storage_dlq(storage_name, vcon_id)
+                break
+        return JSONResponse(content=succeeded)
+    except Exception as e:
+        logger.error(f"Error reprocessing storage DLQ: {str(e)}")
+        raise HTTPException(status_code=500, detail="Failed to reprocess storage DLQ")
+
+
+@api_router.get(
+    "/dlq/storage",
+    status_code=200,
+    summary="Get storage DLQ contents",
+    description="Get list of vCons whose write to a storage backend failed",
+    tags=["dlq"],
+)
+async def get_storage_dlq_vcons(
+    storage_name: str = Query(..., description="Name of the storage backend")
+) -> JSONResponse:
+    """Get all vCon ids in a storage backend's dead letter queue."""
+    try:
+        dlq_name = get_storage_dlq_name(storage_name)
+        vcons = await redis_async.lrange(dlq_name, 0, -1)
+        return JSONResponse(content=vcons)
+    except Exception as e:
+        logger.error(f"Error reading storage DLQ: {str(e)}")
+        raise HTTPException(status_code=500, detail="Failed to read storage DLQ")
+
+
 @api_router.get(
     "/dlq",
     status_code=200,

common/dlq_utils.py

@@ -2,3 +2,14 @@
 
 def get_ingress_list_dlq_name(ingress_list: str) -> str:
     return f"DLQ:{ingress_list}"
+
+
+def get_storage_dlq_name(storage_name: str) -> str:
+    """DLQ for vCons whose write to a storage backend failed.
+
+    Kept separate from the ingress DLQ because replaying one of these means
+    re-attempting only the storage write. Replaying through the ingress list
+    would re-run the whole chain, including transcription that already
+    succeeded.
+    """
+    return f"DLQ:storage:{storage_name}"

common/lib/queue.py

@@ -108,6 +108,20 @@ def enqueue_dlq(self, ingress_list: str, vcon_id: str) -> int:
         increment_counter("conserver.dlq.count", attributes={"queue_name": dlq_name})
         return result
 
+    def enqueue_storage_dlq(self, storage_name: str, vcon_id: str) -> int:
+        """RPUSH a vCon onto the DLQ for a storage backend that failed to write.
+
+        Emits the same ``conserver.dlq.count{queue_name}`` counter as
+        :meth:`enqueue_dlq`, so an alert on that metric covers storage
+        failures without needing a new rule.
+        """
+        from dlq_utils import get_storage_dlq_name
+
+        dlq_name = get_storage_dlq_name(storage_name)
+        result = self._client.rpush(dlq_name, vcon_id)
+        increment_counter("conserver.dlq.count", attributes={"queue_name": dlq_name})
+        return result
+
     def queue_length(self, list_name: str) -> int:
         return self._client.llen(list_name)
 
@@ -144,3 +158,15 @@ async def dequeue_dlq_async(self, redis_async, ingress_list: str):
 
         dlq_name = get_ingress_list_dlq_name(ingress_list)
         return await redis_async.lpop(dlq_name)
+
+    async def dequeue_storage_dlq_async(self, redis_async, storage_name: str):
+        """Async LPOP one vCon id off a storage backend's DLQ.
+
+        Returns the popped vCon id, or ``None`` if the DLQ is empty. Pairs
+        with :meth:`enqueue_storage_dlq` (RPUSH) for oldest-failure-first
+        ordering.
+        """
+        from dlq_utils import get_storage_dlq_name
+
+        dlq_name = get_storage_dlq_name(storage_name)
+        return await redis_async.lpop(dlq_name)

common/storage/vcon_mcp/__init__.py

@@ -14,10 +14,14 @@
 - base_url: Base URL of vcon-mcp REST API (e.g. http://localhost:3000/api/v1)
 - api_key: Optional. API key for Authorization: Bearer <api_key>
 - timeout: Optional. Request timeout in seconds (default: 30)
+- transient_retries: Optional. Retries for transient failures (default: 3)
+- transient_backoff_base_s: Optional. Backoff factor in seconds (default: 0.5)
 """
 
 from typing import Optional, Dict, Any
 import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
 from lib.logging_utils import init_logger
 from lib.vcon_redis import VconRedis
 
@@ -27,8 +31,48 @@
     "base_url": "http://127.0.0.1:3000/api/v1",
     "api_key": "",
     "timeout": 30,
+    "transient_retries": 3,
+    "transient_backoff_base_s": 0.5,
 }
 
+# Retryable because the request never got a verdict, or got one that says
+# "later". Everything else (401, 400, 404, ...) is a decision the server has
+# already made, and repeating the call only delays the inevitable.
+_RETRY_STATUSES = (429, 502, 503, 504)
+
+
+def _session(opts: Dict[str, Any]) -> requests.Session:
+    """Session that retries transient failures with exponential backoff.
+
+    urllib3 retries connection errors and read timeouts in addition to the
+    status codes listed, and honours Retry-After on 429/503.
+
+    POST is included in ``allowed_methods``, which urllib3 excludes by default
+    as non-idempotent. That is safe here specifically because vcon-mcp upserts
+    on the vCon uuid (``onConflict: 'id'``), so a retried create converges on
+    one row instead of duplicating.
+    """
+    retries = opts.get("transient_retries", default_options["transient_retries"])
+    backoff = opts.get(
+        "transient_backoff_base_s", default_options["transient_backoff_base_s"]
+    )
+    retry = Retry(
+        total=retries,
+        connect=retries,
+        read=retries,
+        status=retries,
+        status_forcelist=_RETRY_STATUSES,
+        allowed_methods=frozenset(["GET", "POST", "DELETE"]),
+        backoff_factor=backoff,
+        raise_on_status=False,
+        respect_retry_after_header=True,
+    )
+    session = requests.Session()
+    adapter = HTTPAdapter(max_retries=retry)
+    session.mount("http://", adapter)
+    session.mount("https://", adapter)
+    return session
+
 
 def _headers(opts: Dict[str, Any]) -> Dict[str, str]:
     """Build request headers, including optional Bearer token."""
@@ -67,7 +111,7 @@ def save(vcon_uuid: str, opts: Dict[str, Any] = None) -> None:
         payload = vcon.to_dict()
         url = _url(opts, "vcons")
         timeout = opts.get("timeout", default_options["timeout"])
-        resp = requests.post(
+        resp = _session(opts).post(
             url,
             json=payload,
             headers=_headers(opts),
@@ -107,7 +151,7 @@ def get(vcon_uuid: str, opts: Dict[str, Any] = None) -> Optional[dict]:
     try:
         url = _url(opts, f"vcons/{vcon_uuid}")
         timeout = opts.get("timeout", default_options["timeout"])
-        resp = requests.get(
+        resp = _session(opts).get(
             url,
             headers=_headers(opts),
             timeout=timeout,
@@ -147,7 +191,7 @@ def delete(vcon_uuid: str, opts: Dict[str, Any] = None) -> bool:
     try:
         url = _url(opts, f"vcons/{vcon_uuid}")
         timeout = opts.get("timeout", default_options["timeout"])
-        resp = requests.delete(
+        resp = _session(opts).delete(
             url,
             headers=_headers(opts),
             timeout=timeout,