fix(transcoder): ack only after successful metadata ready update

services/transcoder/app/consumer.py

@@ -830,8 +830,15 @@ def process_message(body: bytes):
                 thumb_key = None
 
     ok = update_status(video_id, "ready", renditions, thumb_key)
+    if not ok:
+        # ack must follow a successful metadata update — raise so the consumer
+        # sends the message to the retry queue / DLQ instead of acking a video
+        # that would stay "processing" forever. Retry re-runs the whole pipeline
+        # even though renditions are already uploaded (bounded by MAX_RETRIES);
+        # re-uploads are idempotent per rendition key.
+        raise RuntimeError(f"metadata ready update failed for {video_id}")
     # Phase 15 cost: delete raw only after metadata confirmed ready. Non-fatal; lifecycle 7d is safety net.
-    if ok and not KEEP_RAW:
+    if not KEEP_RAW:
         try:
             mc.remove_object(BUCKET, s3_key)
             log.info("cleaned raw s3://%s/%s after ready (KEEP_RAW=false)", BUCKET, s3_key)
@@ -863,6 +870,72 @@ def _get_status(video_id: str) -> str | None:
     return None
 
 
+def handle_message(ch, method, properties, body):
+    """Ack only after process_message succeeds (including the ready metadata update).
+
+    On failure: republish to the retry queue while retries remain, otherwise
+    mark failed and basic_nack(requeue=False) so the broker dead-letters to DLQ.
+    Module-level so retry/DLQ behavior stays testable (issue #50).
+    """
+    headers = {}
+    if properties and getattr(properties, "headers", None):
+        headers = properties.headers or {}
+    retry_count = 0
+    if headers:
+        try:
+            retry_count = int(headers.get("x-retry-count", 0) or 0)
+        except Exception:
+            retry_count = 0
+    try:
+        process_message(body)
+        ch.basic_ack(delivery_tag=method.delivery_tag)
+    except Exception as e:
+        log.exception("process failed: %s", e)
+        if retry_count < MAX_RETRIES:
+            try:
+                new_headers = dict(headers) if headers else {}
+                new_headers["x-retry-count"] = retry_count + 1
+                props = pika.BasicProperties(
+                    delivery_mode=2,
+                    headers=new_headers,
+                    content_type="application/json",
+                )
+                ch.basic_publish(
+                    exchange="",
+                    routing_key=RETRY_QUEUE,
+                    body=body,
+                    properties=props,
+                )
+                ch.basic_ack(delivery_tag=method.delivery_tag)
+                log.info(
+                    "requeued to %s %d/%d (retry %d)",
+                    RETRY_QUEUE,
+                    retry_count + 1,
+                    MAX_RETRIES,
+                    retry_count + 1,
+                )
+            except Exception as pub_e:
+                log.exception("retry publish failed: %s", pub_e)
+                try:
+                    data = json.loads(body)
+                    vid = data.get("video_id")
+                    if vid:
+                        update_status(vid, "failed")
+                except Exception:
+                    pass
+                ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
+        else:
+            try:
+                data = json.loads(body)
+                vid = data.get("video_id")
+                if vid:
+                    update_status(vid, "failed")
+            except Exception:
+                pass
+            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
+            log.error("moved to DLQ %s after %d retries", DLQ, MAX_RETRIES)
+
+
 def main():
     signal.signal(signal.SIGTERM, _handle_sigterm)
     signal.signal(signal.SIGINT, _handle_sigterm)
@@ -910,67 +983,7 @@ def main():
                     continue
                 raise
             channel.basic_qos(prefetch_count=1)
-
-            def on_message(ch, method, properties, body):
-                headers = {}
-                if properties and getattr(properties, "headers", None):
-                    headers = properties.headers or {}
-                retry_count = 0
-                if headers:
-                    try:
-                        retry_count = int(headers.get("x-retry-count", 0) or 0)
-                    except Exception:
-                        retry_count = 0
-                try:
-                    process_message(body)
-                    ch.basic_ack(delivery_tag=method.delivery_tag)
-                except Exception as e:
-                    log.exception("process failed: %s", e)
-                    if retry_count < MAX_RETRIES:
-                        try:
-                            new_headers = dict(headers) if headers else {}
-                            new_headers["x-retry-count"] = retry_count + 1
-                            props = pika.BasicProperties(
-                                delivery_mode=2,
-                                headers=new_headers,
-                                content_type="application/json",
-                            )
-                            ch.basic_publish(
-                                exchange="",
-                                routing_key=RETRY_QUEUE,
-                                body=body,
-                                properties=props,
-                            )
-                            ch.basic_ack(delivery_tag=method.delivery_tag)
-                            log.info(
-                                "requeued to %s %d/%d (retry %d)",
-                                RETRY_QUEUE,
-                                retry_count + 1,
-                                MAX_RETRIES,
-                                retry_count + 1,
-                            )
-                        except Exception as pub_e:
-                            log.exception("retry publish failed: %s", pub_e)
-                            try:
-                                data = json.loads(body)
-                                vid = data.get("video_id")
-                                if vid:
-                                    update_status(vid, "failed")
-                            except Exception:
-                                pass
-                            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
-                    else:
-                        try:
-                            data = json.loads(body)
-                            vid = data.get("video_id")
-                            if vid:
-                                update_status(vid, "failed")
-                        except Exception:
-                            pass
-                        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
-                        log.error("moved to DLQ %s after %d retries", DLQ, MAX_RETRIES)
-
-            channel.basic_consume(queue=QUEUE, on_message_callback=on_message)
+            channel.basic_consume(queue=QUEUE, on_message_callback=handle_message)
             log.info(
                 "consumer ready, waiting for %s (threads=%s preset=%s)",
                 QUEUE,

services/transcoder/tests/test_consumer.py

@@ -1,6 +1,8 @@
 import json
 from unittest.mock import MagicMock, patch
 
+import pytest
+
 import app.consumer as cons
 
 
@@ -357,5 +359,85 @@ def test_process_message_no_delete_when_metadata_update_fails(monkeypatch):
         patch("app.consumer.transcode_thumbnail_from_rendition", return_value=None),
         patch("os.path.exists", return_value=False),
     ):
-        cons.process_message(body)
+        # issue #50: failed ready update must raise so handle_message retries / DLQs
+        # instead of acking a video that stays "processing" forever
+        with pytest.raises(RuntimeError, match="metadata ready update failed"):
+            cons.process_message(body)
         fake_minio.remove_object.assert_not_called()
+
+
+def _handle_message_channel(delivery_tag=1):
+    ch = MagicMock()
+    method = MagicMock(delivery_tag=delivery_tag)
+    return ch, method
+
+
+def test_handle_message_acks_after_success():
+    ch, method = _handle_message_channel()
+    body = json.dumps({"video_id": "vid-1", "s3_key": "raw/vid-1/original.mp4"}).encode()
+    with patch("app.consumer.process_message") as mock_proc:
+        cons.handle_message(ch, method, MagicMock(headers={}), body)
+        mock_proc.assert_called_once_with(body)
+        ch.basic_ack.assert_called_once_with(delivery_tag=1)
+        ch.basic_publish.assert_not_called()
+        ch.basic_nack.assert_not_called()
+
+
+def test_handle_message_metadata_failure_retries_and_does_not_ack_silently():
+    # issue #50 verify: metadata update failed -> retry-queue republish, not a plain ack
+    ch, method = _handle_message_channel()
+    body = json.dumps({"video_id": "vid-1", "s3_key": "raw/vid-1/original.mp4"}).encode()
+    with (
+        patch(
+            "app.consumer.process_message",
+            side_effect=RuntimeError("metadata ready update failed for vid-1"),
+        ) as mock_proc,
+    ):
+        cons.handle_message(ch, method, MagicMock(headers={}), body)
+        mock_proc.assert_called_once_with(body)
+        assert ch.basic_publish.call_count == 1
+        pub_kwargs = ch.basic_publish.call_args[1]
+        assert pub_kwargs["routing_key"] == cons.RETRY_QUEUE
+        assert pub_kwargs["properties"].headers["x-retry-count"] == 1
+        ch.basic_ack.assert_called_once_with(delivery_tag=1)
+        ch.basic_nack.assert_not_called()
+
+
+def test_handle_message_respects_incoming_retry_count():
+    ch, method = _handle_message_channel()
+    body = json.dumps({"video_id": "vid-1", "s3_key": "raw/vid-1/original.mp4"}).encode()
+    props = MagicMock(headers={"x-retry-count": 2})
+    with patch("app.consumer.process_message", side_effect=Exception("boom")):
+        cons.handle_message(ch, method, props, body)
+        pub_kwargs = ch.basic_publish.call_args[1]
+        assert pub_kwargs["properties"].headers["x-retry-count"] == 3
+
+
+def test_handle_message_publish_failure_marks_failed_and_dlqs():
+    # retry publish itself fails (e.g. broker error) -> mark failed, dead-letter, no ack
+    ch, method = _handle_message_channel(delivery_tag=3)
+    body = json.dumps({"video_id": "vid-1", "s3_key": "raw/vid-1/original.mp4"}).encode()
+    ch.basic_publish.side_effect = Exception("broker down")
+    with (
+        patch("app.consumer.process_message", side_effect=Exception("boom")),
+        patch("app.consumer.update_status") as mock_status,
+    ):
+        cons.handle_message(ch, method, MagicMock(headers={}), body)
+        mock_status.assert_called_once_with("vid-1", "failed")
+        ch.basic_nack.assert_called_once_with(delivery_tag=3, requeue=False)
+        ch.basic_ack.assert_not_called()
+
+
+def test_handle_message_moves_to_dlq_after_max_retries():
+    ch, method = _handle_message_channel(delivery_tag=7)
+    body = json.dumps({"video_id": "vid-1", "s3_key": "raw/vid-1/original.mp4"}).encode()
+    props = MagicMock(headers={"x-retry-count": cons.MAX_RETRIES})
+    with (
+        patch("app.consumer.process_message", side_effect=Exception("boom")),
+        patch("app.consumer.update_status") as mock_status,
+    ):
+        cons.handle_message(ch, method, props, body)
+        mock_status.assert_called_once_with("vid-1", "failed")
+        ch.basic_nack.assert_called_once_with(delivery_tag=7, requeue=False)
+        ch.basic_ack.assert_not_called()
+        ch.basic_publish.assert_not_called()