diff --git a/api/api.py b/api/api.py index d572ba2..9927026 100644 --- a/api/api.py +++ b/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, diff --git a/common/dlq_utils.py b/common/dlq_utils.py index 95793c3..3cfa8a4 100644 --- a/common/dlq_utils.py +++ b/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}" diff --git a/common/lib/queue.py b/common/lib/queue.py index 5569b4f..7ccc4c0 100644 --- a/common/lib/queue.py +++ b/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) diff --git a/common/storage/vcon_mcp/__init__.py b/common/storage/vcon_mcp/__init__.py index 495624c..a488fd0 100644 --- a/common/storage/vcon_mcp/__init__.py +++ b/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 - 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, diff --git a/common/tests/test_storage_dlq.py b/common/tests/test_storage_dlq.py new file mode 100644 index 0000000..24f0247 --- /dev/null +++ b/common/tests/test_storage_dlq.py @@ -0,0 +1,233 @@ +"""Tests for the storage dead-letter path (CON-714). + +A failed storage write used to be logged and dropped: the vCon never reached +the backend, nothing queued it, and the chain reported success. On BDS that +meant a ~100s outage of one backend silently lost every vCon written during it. + +These tests pin the two halves of the fix: failures land on a per-backend DLQ +with the vCon body kept alive long enough to replay, and the vcon-mcp client +retries the failures that are worth retrying. +""" + +from unittest.mock import MagicMock, patch + +import pytest + +from lib.queue import VconQueue + + +class TestEnqueueStorageDlq: + def test_emits_counter_with_storage_dlq_name(self): + """Reuses ``conserver.dlq.count`` so an existing alert on that metric + covers storage failures without a new rule.""" + mock_client = MagicMock() + mock_client.rpush.return_value = 1 + q = VconQueue(client=mock_client) + + with patch("lib.queue.increment_counter") as inc: + q.enqueue_storage_dlq("vcon_mcp", "vcon-uuid-1234") + + mock_client.rpush.assert_called_once_with("DLQ:storage:vcon_mcp", "vcon-uuid-1234") + inc.assert_called_once_with( + "conserver.dlq.count", + attributes={"queue_name": "DLQ:storage:vcon_mcp"}, + ) + + def test_storage_dlq_is_distinct_from_ingress_dlq(self): + """A storage failure must not land on the ingress DLQ: replaying from + there re-runs the whole chain, including transcription.""" + mock_client = MagicMock() + mock_client.rpush.return_value = 1 + q = VconQueue(client=mock_client) + + with patch("lib.queue.increment_counter"): + q.enqueue_storage_dlq("vcon_mcp", "v1") + q.enqueue_dlq("vcon_mcp", "v1") + + pushed = [call.args[0] for call in mock_client.rpush.call_args_list] + assert pushed == ["DLQ:storage:vcon_mcp", "DLQ:vcon_mcp"] + + def test_no_counter_when_rpush_fails(self): + """Only count entries that actually landed.""" + mock_client = MagicMock() + mock_client.rpush.side_effect = RuntimeError("redis down") + q = VconQueue(client=mock_client) + + with patch("lib.queue.increment_counter") as inc: + with pytest.raises(RuntimeError): + q.enqueue_storage_dlq("vcon_mcp", "v1") + + inc.assert_not_called() + + def test_returns_rpush_result(self): + mock_client = MagicMock() + mock_client.rpush.return_value = 42 + q = VconQueue(client=mock_client) + + with patch("lib.queue.increment_counter"): + assert q.enqueue_storage_dlq("vcon_mcp", "v1") == 42 + + +class TestProcessStorageDeadLetters: + """``_process_storage`` must dead-letter on failure instead of dropping.""" + + def _make_request(self): + from main import VconChainRequest + + return VconChainRequest( + chain_details={"name": "test_chain", "links": [], "storages": ["vcon_mcp"]}, + vcon_id="vcon-uuid-1234", + ) + + def test_failed_write_is_dead_lettered_and_ttl_extended(self): + req = self._make_request() + + with patch("main.Storage") as mock_storage, \ + patch("main.queue") as mock_queue, \ + patch("main.VCON_DLQ_EXPIRY", 604800): + mock_storage.return_value.save.side_effect = RuntimeError("401 Unauthorized") + req._process_storage("vcon_mcp") + + mock_queue.enqueue_storage_dlq.assert_called_once_with("vcon_mcp", "vcon-uuid-1234") + mock_queue.set_vcon_ttl.assert_called_once_with("vcon-uuid-1234", 604800) + + def test_successful_write_is_not_dead_lettered(self): + req = self._make_request() + + with patch("main.Storage"), patch("main.queue") as mock_queue: + req._process_storage("vcon_mcp") + + mock_queue.enqueue_storage_dlq.assert_not_called() + + def test_process_storage_still_does_not_raise(self): + """The chain must keep going when one backend fails; other backends + and egress still need to run.""" + req = self._make_request() + + with patch("main.Storage") as mock_storage, patch("main.queue"): + mock_storage.return_value.save.side_effect = RuntimeError("boom") + req._process_storage("vcon_mcp") # must not raise + + def test_ttl_not_extended_when_expiry_disabled(self): + req = self._make_request() + + with patch("main.Storage") as mock_storage, \ + patch("main.queue") as mock_queue, \ + patch("main.VCON_DLQ_EXPIRY", 0): + mock_storage.return_value.save.side_effect = RuntimeError("boom") + req._process_storage("vcon_mcp") + + mock_queue.enqueue_storage_dlq.assert_called_once() + mock_queue.set_vcon_ttl.assert_not_called() + + def test_redis_failure_while_dead_lettering_does_not_propagate(self): + """If Redis is down too the vCon is genuinely lost, but that must not + also take down the chain.""" + req = self._make_request() + + with patch("main.Storage") as mock_storage, patch("main.queue") as mock_queue: + mock_storage.return_value.save.side_effect = RuntimeError("boom") + mock_queue.enqueue_storage_dlq.side_effect = RuntimeError("redis down") + req._process_storage("vcon_mcp") # must not raise + + +class TestStorageDlqReprocessEndpoint: + """``POST /dlq/storage/reprocess`` replays the write, not the chain.""" + + def _call(self, dlq_items, save_side_effect=None, count=1000): + import asyncio + + import api as api_module + + popped = list(dlq_items) + + async def fake_pop(_redis, _storage_name): + return popped.pop(0) if popped else None + + mock_queue = MagicMock() + mock_queue.dequeue_storage_dlq_async = fake_pop + mock_storage = MagicMock() + if save_side_effect is not None: + mock_storage.return_value.save.side_effect = save_side_effect + + # ``redis_async`` is bound by the app's lifespan startup, so it does not + # exist when the endpoint is called directly. + with patch.object(api_module, "queue", mock_queue), \ + patch.object(api_module, "Storage", mock_storage), \ + patch.object(api_module, "redis_async", MagicMock(), create=True): + response = asyncio.run( + api_module.post_storage_dlq_reprocess(storage_name="vcon_mcp", count=count) + ) + return response, mock_storage, mock_queue + + def test_replays_each_vcon_through_storage_save(self): + response, mock_storage, _ = self._call(["v1", "v2", "v3"]) + + assert response.body == b"3" + saved = [call.args[0] for call in mock_storage.return_value.save.call_args_list] + assert saved == ["v1", "v2", "v3"] + + def test_stops_and_requeues_when_backend_still_down(self): + """A still-broken backend must not drain the DLQ into nothing.""" + response, mock_storage, mock_queue = self._call( + ["v1", "v2", "v3"], save_side_effect=RuntimeError("still 401") + ) + + assert response.body == b"0" + # First item is put back, and we stop rather than popping the rest. + mock_queue.enqueue_storage_dlq.assert_called_once_with("vcon_mcp", "v1") + assert mock_storage.return_value.save.call_count == 1 + + def test_empty_dlq_returns_zero(self): + response, mock_storage, _ = self._call([]) + + assert response.body == b"0" + mock_storage.return_value.save.assert_not_called() + + def test_count_bounds_the_work(self): + """Mirrors the CON-575 fix on /dlq/reprocess: bounded per call.""" + response, mock_storage, _ = self._call(["v1", "v2", "v3"], count=2) + + assert response.body == b"2" + assert mock_storage.return_value.save.call_count == 2 + + +class TestVconMcpRetries: + """Transient failures should be retried before the vCon is dead-lettered.""" + + def _adapter(self, opts=None): + from storage.vcon_mcp import _session, default_options + + session = _session(opts if opts is not None else default_options) + return session.get_adapter("http://example.com") + + def test_retries_transient_statuses_only(self): + retry = self._adapter().max_retries + + assert set(retry.status_forcelist) == {429, 502, 503, 504} + # A 401 or 400 is a decision the server already made; retrying it only + # delays the dead-letter. + assert 401 not in retry.status_forcelist + assert 400 not in retry.status_forcelist + + def test_post_is_retryable_because_vcon_mcp_upserts(self): + """urllib3 excludes POST by default. vcon-mcp upserts on the vCon uuid, + so a retried create converges instead of duplicating.""" + retry = self._adapter().max_retries + + assert "POST" in retry.allowed_methods + + def test_defaults_match_documented_options(self): + retry = self._adapter().max_retries + + assert retry.total == 3 + assert retry.backoff_factor == 0.5 + assert retry.respect_retry_after_header is True + + def test_options_override_defaults(self): + retry = self._adapter( + {"transient_retries": 7, "transient_backoff_base_s": 1.5} + ).max_retries + + assert retry.total == 7 + assert retry.backoff_factor == 1.5 diff --git a/conserver/main.py b/conserver/main.py index c2cc145..96b2cce 100644 --- a/conserver/main.py +++ b/conserver/main.py @@ -416,18 +416,45 @@ def _process_storage(self, storage_name: str) -> None: current_span.set_status(Status(StatusCode.ERROR, str(e))) current_span.record_exception(e) logger.error( - "Failed to save vCon %s to storage %s: %s", + "Failed to save vCon %s to storage %s: %s - Moving to storage DLQ", self.vcon_id, storage_name, str(e), exc_info=True ) + # Dead-letter rather than re-raise. Re-raising would send the vCon + # to the ingress DLQ, and replaying from there re-runs the whole + # chain including transcription that already succeeded. This DLQ + # replays the storage write alone. + self._dead_letter_storage(storage_name) finally: duration_ms = round((time.time() - started) * 1000, 3) attrs = {"backend": storage_name, "outcome": outcome} increment_counter("conserver.storage.count", attributes=attrs) record_histogram("conserver.storage.duration_ms", duration_ms, attributes=attrs) + def _dead_letter_storage(self, storage_name: str) -> None: + """Record a failed storage write so the vCon can be replayed later. + + Pushes onto ``DLQ:storage:`` and extends the vCon's Redis TTL + to ``VCON_DLQ_EXPIRY`` so the body outlives the default retention and + is still there to replay. Mirrors what the ingress DLQ path does. + + Never raises: this already runs on an error path, and losing the vCon + outright because Redis also hiccuped is the failure mode being fixed. + """ + try: + queue.enqueue_storage_dlq(storage_name, self.vcon_id) + if VCON_DLQ_EXPIRY > 0: + queue.set_vcon_ttl(self.vcon_id, VCON_DLQ_EXPIRY) + except Exception: + logger.error( + "Could not dead-letter vCon %s for storage %s; it is now lost", + self.vcon_id, + storage_name, + exc_info=True, + ) + def _process_storage_parallel(self, storage_backends: List[str]) -> None: """Save vCon to multiple storage backends concurrently.