fix(storage): send webhook HTTP failures to the recovery queue

common/storage/webhook/__init__.py

@@ -11,6 +11,7 @@
 default_options = {
     "webhook-urls": [],
     "headers": {},
+    "timeout": 30,
 }
 
 
@@ -37,10 +38,16 @@ def save(vcon_uuid, opts=default_options):
             f"webhook storage: posting vcon {vcon_uuid} to webhook url: {url}"
         )
         webhook_start = time.time()
-        resp = requests.post(url, json=json_dict, headers=headers)
+        resp = requests.post(
+            url, json=json_dict, headers=headers,
+            timeout=opts.get("timeout", default_options["timeout"]),
+        )
         webhook_duration = round(time.time() - webhook_start, 3)
         logger.info(
             f"webhook storage response for {vcon_uuid}: {resp.status_code} {resp.text}"
         )
         record_histogram("conserver.webhook.duration", webhook_duration,
                          attributes={"status_code": str(resp.status_code)})
+
+        # Propagate HTTP failures so the conserver retains this write in its storage DLQ.
+        resp.raise_for_status()

common/tests/test_webhook_storage_failures.py

@@ -0,0 +1,45 @@
+"""HTTP failure propagation uses an opaque probe, not a customer vCon."""
+
+from unittest.mock import patch
+
+import pytest
+import requests
+from storage import webhook
+
+
+@pytest.mark.parametrize("status", [401, 500, 503])
+def test_http_failure_reaches_storage_dlq(status):
+    from main import VconChainRequest
+
+    response = requests.Response()
+    response.status_code = status
+    response._content = b"{}"
+    response.url = "https://receiver.invalid/probe"
+    request = VconChainRequest(
+        chain_details={"name": "test", "links": [], "storages": ["test_webhook"]},
+        vcon_id="synthetic-id",
+    )
+    with (
+        patch.object(webhook, "VconRedis") as bodies,
+        patch.object(webhook.requests, "post", return_value=response),
+        patch("main.Storage") as storage,
+        patch("main.queue") as queue,
+    ):
+        bodies.return_value.get_vcon.return_value.to_dict.return_value = {"synthetic_retry_probe": True}
+        storage.return_value.save.side_effect = lambda vid: webhook.save(vid, {"webhook-urls": [response.url]})
+        request._process_storage("test_webhook")
+        queue.enqueue_storage_dlq.assert_called_once_with("test_webhook", "synthetic-id", retention_seconds=604800)
+
+
+@pytest.mark.parametrize("options,expected", [({}, 30), ({"timeout": 4}, 4)])
+def test_success_has_bounded_request_timeout(options, expected):
+    response = requests.Response()
+    response.status_code = 201
+    response._content = b"{}"
+    with (
+        patch.object(webhook, "VconRedis") as bodies,
+        patch.object(webhook.requests, "post", return_value=response) as post,
+    ):
+        bodies.return_value.get_vcon.return_value.to_dict.return_value = {"synthetic_retry_probe": True}
+        webhook.save("synthetic-id", {"webhook-urls": ["https://receiver.invalid/probe"], **options})
+        assert post.call_args.kwargs["timeout"] == expected

docs/storage-recovery.md

@@ -51,3 +51,11 @@ update both base images; updating only workers queues failures without providing
 the recovery endpoint. Preserve the previous images for rollback. Rolling back
 must not clear `DLQ:storage:*`; old images cannot replay these queues. After rollout,
 verify versions, health, storage-error alerts, and queue depth before a small replay.
+
+## Generic webhook backends
+
+`storage.webhook` propagates HTTP 4xx/5xx and transport failures to the same storage
+DLQ. Its `timeout` option defaults to 30 seconds per request. It does not automatically
+retry arbitrary POST receivers because their idempotency contracts are unknown.
+Before replaying a backend with multiple webhook URLs, confirm every receiver can
+tolerate duplicate writes: an earlier URL may have succeeded before a later one failed.