diff --git a/services/transcoder/app/consumer.py b/services/transcoder/app/consumer.py index bff701c..a1acf02 100644 --- a/services/transcoder/app/consumer.py +++ b/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, diff --git a/services/transcoder/tests/test_consumer.py b/services/transcoder/tests/test_consumer.py index ec46662..16dc336 100644 --- a/services/transcoder/tests/test_consumer.py +++ b/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()