fix(api): await Kafka acknowledgements and bound HTTP requests

app/api/producer.py

@@ -1,185 +1,91 @@
-import asyncio
+"""Kafka publications complete only on broker delivery callbacks."""
-import logging
+import threading
 from abc import ABC, abstractmethod
-from typing import Any, Callable, Dict, Optional
+from concurrent.futures import Future
 
 from app.common.config import config
-from app.common.logging import setup_logger
-
-logger = setup_logger("ingestion-api")
 
 
 class MessageProducer(ABC):
     @abstractmethod
-    def produce(self, topic: str, key: Optional[str], value: bytes) -> bool:
+    def produce(self, topic: str, key: str | None, value: bytes) -> Future:
-        """Produce a message synchronously or enqueue in broker buffer."""
+        """Return a future resolved by the destination acknowledgement."""
-        pass
 
     @abstractmethod
-    def is_healthy(self) -> bool:
+    def close(self, timeout: float = 5.0) -> None:
-        """Check if producer is healthy and connected to broker."""
-        pass
-
-    @abstractmethod
-    def flush(self, timeout: float = 5.0) -> int:
-        """Flush any pending messages."""
-        pass
-
-    @abstractmethod
-    def close(self) -> None:
-        """Close producer connection."""
         pass
 
 
 class ConfluentKafkaProducer(MessageProducer):
-    """
+    def __init__(self, bootstrap_servers=None):
-    Production Kafka producer implementation wrapping confluent-kafka.
+        from confluent_kafka import Producer
-    Uses 'raw_id' as partition key and LZ4 compression.
+        self._producer = Producer({
-    """
+            "bootstrap.servers": bootstrap_servers or config.kafka.bootstrap_servers,
-
-    def __init__(self, bootstrap_servers: Optional[str] = None):
-        try:
-            import confluent_kafka
-            self._ck = confluent_kafka
-        except ImportError:
-            raise ImportError(
-                "confluent-kafka library is not installed. Run 'pip install confluent-kafka'."
-            )
-
-        self.bootstrap_servers = bootstrap_servers or config.kafka.bootstrap_servers
-        self._producer_config = {
-            "bootstrap.servers": self.bootstrap_servers,
             "compression.type": "lz4",
-            "queue.buffering.max.messages": 100000,
+            "queue.buffering.max.messages": config.api.queue_max_size,
-            "queue.buffering.max.ms": 5,
+            "linger.ms": 5,
             "acks": "all",
-            "retries": 3,
+            "enable.idempotence": True,
-        }
+            "delivery.timeout.ms": int(config.kafka.delivery_timeout_seconds * 1000),
-        self._producer = self._ck.Producer(self._producer_config)
+        })
-        self._healthy = True
+        self._stop = threading.Event()
+        self._closed = False
+        self._thread = threading.Thread(target=self._poll, name="KafkaDelivery", daemon=True)
+        self._thread.start()
 
-    def _delivery_report(self, err: Any, msg: Any) -> None:
+    def _poll(self):
-        if err is not None:
+        while not self._stop.is_set():
-            self._healthy = False
+            self._producer.poll(0.01)
-            raw_id = msg.key().decode("utf-8") if msg and msg.key() else None
-            logger.error(
-                f"Kafka delivery failed: {err}",
-                extra={"raw_id": raw_id, "event": "kafka_produce_failed"},
-            )
-        else:
-            self._healthy = True
 
-    def produce(self, topic: str, key: Optional[str], value: bytes) -> bool:
+    def produce(self, topic, key, value):
+        future = Future()
+        def delivered(error, message):
+            if error is not None:
+                future.set_exception(RuntimeError(str(error)))
+            else:
+                future.set_result(message)
         try:
-            key_bytes = key.encode("utf-8") if key else None
+            if self._closed:
-            self._producer.produce(
+                raise RuntimeError("Producer is closed")
-                topic=topic,
+            self._producer.produce(topic, key=key, value=value, on_delivery=delivered)
-                key=key_bytes,
+        except Exception as exc:
-                value=value,
+            future.set_exception(exc)
-                on_delivery=self._delivery_report,
+        return future
-            )
-            # Serve delivery callback queue non-blockingly
-            self._producer.poll(0)
-            return True
-        except BufferError:
-            logger.warning(
-                "Kafka local producer buffer full",
-                extra={"raw_id": key, "event": "kafka_buffer_full"},
-            )
-            self._producer.poll(0.1)
-            return False
-        except Exception as e:
-            self._healthy = False
-            logger.error(
-                f"Kafka produce error: {e}",
-                extra={"raw_id": key, "event": "kafka_error"},
-            )
-            return False
 
-    def is_healthy(self) -> bool:
+    def flush(self, timeout=5.0):
-        return self._healthy
-
-    def flush(self, timeout: float = 5.0) -> int:
         return self._producer.flush(timeout)
 
-    def close(self) -> None:
+    def close(self, timeout=5.0):
-        self.flush()
+        self._closed = True
+        remaining = self.flush(timeout)
+        self._stop.set()
+        self._thread.join(timeout=1)
+        if remaining:
+            raise TimeoutError(f"{remaining} Kafka deliveries remain unconfirmed")
 
 
 class MockMessageProducer(MessageProducer):
-    """
-    In-memory mock producer for offline testing and verification.
-    """
-
     def __init__(self):
         self.messages = []
         self._healthy = True
 
-    def produce(self, topic: str, key: Optional[str], value: bytes) -> bool:
+    def produce(self, topic, key, value):
-        if not self._healthy:
+        future = Future()
-            return False
+        if self._healthy:
-        self.messages.append({"topic": topic, "key": key, "value": value})
+            self.messages.append({"topic": topic, "key": key, "value": value})
-        return True
+            future.set_result(None)
+        else:
+            future.set_exception(ConnectionError("Broker unavailable"))
+        return future
 
-    def is_healthy(self) -> bool:
+    def is_healthy(self):
         return self._healthy
 
-    def set_healthy(self, healthy: bool) -> None:
+    def set_healthy(self, healthy):
         self._healthy = healthy
 
-    def flush(self, timeout: float = 5.0) -> int:
+    def flush(self, timeout=5.0):
         return 0
 
-    def close(self) -> None:
+    def close(self, timeout=5.0):
         pass
-
-
-class AsyncProducerWorker:
-    """
-    Background worker consuming from the bounded async queue and writing to the Kafka producer.
-    """
-
-    def __init__(self, queue: asyncio.Queue, producer: MessageProducer, topic: str):
-        self.queue = queue
-        self.producer = producer
-        self.topic = topic
-        self._running = False
-        self._task: Optional[asyncio.Task] = None
-
-    async def start(self) -> None:
-        self._running = True
-        self._task = asyncio.create_task(self._run_loop())
-
-    async def _run_loop(self) -> None:
-        while self._running:
-            try:
-                item = await self.queue.get()
-                if item is None:  # Sentinel to stop
-                    self.queue.task_done()
-                    break
-
-                key, value = item
-                success = self.producer.produce(self.topic, key, value)
-                if not success:
-                    logger.error(
-                        "Failed to publish message from async queue to Kafka",
-                        extra={"raw_id": key, "event": "queue_dispatch_failed"},
-                    )
-                self.queue.task_done()
-            except asyncio.CancelledError:
-                break
-            except Exception as e:
-                logger.error(
-                    f"Unexpected error in async producer worker: {e}",
-                    extra={"event": "worker_unexpected_error"},
-                )
-
-    async def stop(self) -> None:
-        self._running = False
-        if self._task:
-            self._task.cancel()
-            try:
-                await self._task
-            except asyncio.CancelledError:
-                pass
-            self.producer.flush()

app/api/server.py

@@ -1,285 +1,194 @@
 import asyncio
 import json
-import logging
+import signal
-from typing import Optional, Tuple
 
 from app.api.metrics import api_metrics
-from app.api.producer import (
+from app.api.producer import ConfluentKafkaProducer
-    AsyncProducerWorker,
-    ConfluentKafkaProducer,
-    MessageProducer,
-    MockMessageProducer,
-)
 from app.api.stats import get_aggregated_pipeline_stats
 from app.common.config import config
 from app.common.logging import setup_logger
+from app.common.validation import PayloadError, parse_payload, validate_structure
 
 logger = setup_logger("ingestion-api")
 
 
+class HTTPError(ValueError):
+    def __init__(self, status, reason, detail):
+        super().__init__(detail)
+        self.status, self.reason = status, reason
+
+
 class IngestionAPIServer:
-    """
+    def __init__(self, host=None, port=None, queue_max_size=None, producer=None,
-    Framework-less HTTP/1.1 Ingestion API server built on standard library asyncio.
+                 worker_stats_url=None, max_body_bytes=None, read_timeout=None,
-    Enforces bounded async queue backpressure and dispatches to Kafka.
+                 delivery_timeout=None):
-    """
+        self.host = host if host is not None else config.api.host
-
+        self.port = port if port is not None else config.api.port
-    def __init__(
+        self.queue_max_size = queue_max_size if queue_max_size is not None else config.api.queue_max_size
-        self,
+        self.max_body_bytes = max_body_bytes if max_body_bytes is not None else config.api.max_body_bytes
-        host: Optional[str] = None,
+        self.read_timeout = read_timeout if read_timeout is not None else config.api.read_timeout_seconds
-        port: Optional[int] = None,
+        self.delivery_timeout = delivery_timeout if delivery_timeout is not None else config.kafka.delivery_timeout_seconds
-        queue_max_size: Optional[int] = None,
+        if min(self.queue_max_size, self.max_body_bytes, self.read_timeout, self.delivery_timeout) <= 0:
-        producer: Optional[MessageProducer] = None,
+            raise ValueError("API capacities and timeouts must be positive")
-        worker_stats_url: Optional[str] = None,
-    ):
-        self.host = host or config.api.host
-        self.port = port or config.api.port
-        self.queue_max_size = queue_max_size or config.api.queue_max_size
         self.worker_stats_url = worker_stats_url or config.api.worker_stats_url
+        self.producer = producer if producer is not None else ConfluentKafkaProducer()
+        self._pending = set()
+        self._clients = set()
+        self._server = None
+        self._stopping = False
 
-        self.queue: asyncio.Queue = asyncio.Queue(maxsize=self.queue_max_size)
+    async def start(self):
-        self.producer = producer or ConfluentKafkaProducer()
+        self._server = await asyncio.start_server(self._handle_client, self.host, self.port)
-        self.dispatcher = AsyncProducerWorker(
+        self.port = self._server.sockets[0].getsockname()[1]
-            queue=self.queue,
+        logger.info(f"Listening on {self.host}:{self.port}", extra={"event": "server_started"})
-            producer=self.producer,
-            topic=config.kafka.raw_topic,
-        )
 
-        self._server: Optional[asyncio.Server] = None
+    async def stop(self):
-        self._running = False
+        self._stopping = True
-
-    async def start(self) -> None:
-        """Start the async producer dispatcher and HTTP server."""
-        await self.dispatcher.start()
-        self._server = await asyncio.start_server(
-            self._handle_client, self.host, self.port
-        )
-        self._running = True
-        logger.info(
-            f"Ingestion API listening on http://{self.host}:{self.port} (queue capacity: {self.queue_max_size})",
-            extra={"event": "server_started"},
-        )
-
-    async def stop(self) -> None:
-        """Stop the HTTP server and flush dispatcher queue."""
-        self._running = False
         if self._server:
             self._server.close()
             await self._server.wait_closed()
-        await self.dispatcher.stop()
+        deadline = asyncio.get_running_loop().time() + config.api.shutdown_timeout_seconds
-        self.producer.close()
+        outstanding = self._pending | self._clients
-        logger.info("Ingestion API stopped cleanly", extra={"event": "server_stopped"})
+        if outstanding:
+            _, remaining = await asyncio.wait(outstanding, timeout=max(0, deadline-asyncio.get_running_loop().time()))
+            for task in remaining:
+                if isinstance(task, asyncio.Task):
+                    task.cancel()
+            await asyncio.gather(*(t for t in remaining if isinstance(t, asyncio.Task)), return_exceptions=True)
+        await asyncio.to_thread(self.producer.close, max(0, deadline-asyncio.get_running_loop().time()))
+        logger.info("API drained and stopped", extra={"event": "server_stopped"})
 
-    async def _handle_client(
+    async def _read_request(self, reader):
-        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
+        line = await reader.readline()
-    ) -> None:
+        if not line:
+            return None
         try:
-            # Read request line
+            method, path, version = line.decode("ascii").strip().split()
-            request_line = await reader.readline()
+        except (ValueError, UnicodeError) as exc:
-            if not request_line:
+            raise HTTPError(400, "bad_request", "Invalid request line") from exc
-                writer.close()
+        if version not in ("HTTP/1.0", "HTTP/1.1"):
-                await writer.wait_closed()
+            raise HTTPError(400, "bad_request", "Unsupported HTTP version")
-                return
+        headers = {}
+        header_bytes = len(line)
+        while True:
+            line = await reader.readline()
+            header_bytes += len(line)
+            if header_bytes > 16384:
+                raise HTTPError(400, "bad_request", "Headers exceed 16384 bytes")
+            if line in (b"\r\n", b"\n"):
+                break
+            if not line or b":" not in line:
+                raise HTTPError(400, "bad_request", "Invalid header framing")
+            key, value = line.decode("latin-1").split(":", 1)
+            key, value = key.strip().lower(), value.strip()
+            if key in headers and key in ("content-length", "transfer-encoding"):
+                raise HTTPError(400, "bad_request", "Duplicate framing header")
+            headers[key] = value
+        if "transfer-encoding" in headers:
+            raise HTTPError(400, "bad_request", "Transfer-Encoding is unsupported; use Content-Length")
+        length = headers.get("content-length", "0")
+        if not length.isascii() or not length.isdecimal():
+            raise HTTPError(400, "bad_request", "Invalid Content-Length")
+        if len(length) > 10 or int(length) > self.max_body_bytes:
+            raise HTTPError(413, "payload_too_large", "Request body exceeds configured limit")
+        return method.upper(), path, headers, await reader.readexactly(int(length))
 
-            line_str = request_line.decode("utf-8", errors="replace").strip()
+    async def _handle_client(self, reader, writer):
-            parts = line_str.split(" ")
+        task = asyncio.current_task()
-            if len(parts) < 2:
+        self._clients.add(task)
-                await self._send_response(
+        try:
-                    writer, 400, "application/json", b'{"error": "bad_request"}'
+            async with asyncio.timeout(self.read_timeout):
-                )
+                request = await self._read_request(reader)
-                return
+            if request:
-
+                await self._dispatch_route(writer, *request)
-            method = parts[0].upper()
+        except TimeoutError:
-            path = parts[1]
+            await self._error(writer, 408, "request_timeout", "Request read deadline exceeded")
-
+        except HTTPError as exc:
-            # Read headers
+            await self._error(writer, exc.status, exc.reason, str(exc))
-            headers = {}
+        except (asyncio.IncompleteReadError, ValueError) as exc:
-            while True:
+            await self._error(writer, 400, "bad_request", str(exc))
-                header_line = await reader.readline()
+        except (ConnectionError, BrokenPipeError):
-                if not header_line or header_line == b"\r\n" or header_line == b"\n":
-                    break
-                header_str = header_line.decode("utf-8", errors="replace").strip()
-                if ":" in header_str:
-                    key, val = header_str.split(":", 1)
-                    headers[key.strip().lower()] = val.strip()
-
-            # Read body based on Content-Length
-            content_length = int(headers.get("content-length", 0))
-            body = b""
-            if content_length > 0:
-                body = await reader.readexactly(content_length)
-
-            # Route dispatch
-            await self._dispatch_route(writer, method, path, headers, body)
-
-        except asyncio.IncompleteReadError:
             pass
-        except Exception as e:
+        except Exception:
-            logger.error(
+            logger.exception("HTTP handler failure", extra={"event": "request_handler_error"})
-                f"Error handling HTTP request: {e}",
+            await self._error(writer, 500, "internal_error", "Request failed")
-                extra={"event": "request_handler_error"},
-            )
-            try:
-                err_body = json.dumps(
-                    {"error": "internal_error", "detail": str(e)}
-                ).encode("utf-8")
-                await self._send_response(writer, 500, "application/json", err_body)
-            except Exception:
-                pass
         finally:
+            self._clients.discard(task)
+            writer.close()
             try:
-                writer.close()
                 await writer.wait_closed()
-            except Exception:
+            except ConnectionError:
                 pass
 
-    async def _dispatch_route(
+    async def _dispatch_route(self, writer, method, path, headers, body):
-        self,
-        writer: asyncio.StreamWriter,
-        method: str,
-        path: str,
-        headers: dict,
-        body: bytes,
-    ) -> None:
-        # 1. Health check
         if method == "GET" and path in ("/health", "/healthz", "/live"):
-            res = json.dumps(
+            await self._json(writer, 200, {"status": "ok", "in_flight": len(self._pending)})
-                {"status": "ok", "queue_size": self.queue.qsize()}
+        elif method == "GET" and path == "/metrics":
-            ).encode("utf-8")
+            await self._send_response(writer, 200, "text/plain; version=0.0.4; charset=utf-8", api_metrics.get_prometheus_metrics().encode())
-            await self._send_response(writer, 200, "application/json", res)
+        elif method == "GET" and path == "/stats":
-            return
+            await self._json(writer, 200, await get_aggregated_pipeline_stats(self.worker_stats_url))
-
+        elif method == "POST" and path == "/api/v1/cdr/raw":
-        # 2. Prometheus metrics
-        if method == "GET" and path == "/metrics":
-            res = api_metrics.get_prometheus_metrics().encode("utf-8")
-            await self._send_response(
-                writer, 200, "text/plain; version=0.0.4; charset=utf-8", res
-            )
-            return
-
-        # 3. Public Stats summary
-        if method == "GET" and path == "/stats":
-            stats = await get_aggregated_pipeline_stats(self.worker_stats_url)
-            res = json.dumps(stats, indent=2).encode("utf-8")
-            await self._send_response(writer, 200, "application/json", res)
-            return
-
-        # 4. Raw CDR Ingestion endpoint
-        if method == "POST" and path == "/api/v1/cdr/raw":
             await self._handle_raw_cdr_ingestion(writer, body)
-            return
+        else:
+            await self._error(writer, 404, "not_found", f"Cannot {method} {path}")
 
-        # 5. Not found
+    async def _handle_raw_cdr_ingestion(self, writer, body):
-        err = json.dumps(
-            {"error": "not_found", "detail": f"Cannot {method} {path}"}
-        ).encode("utf-8")
-        await self._send_response(writer, 404, "application/json", err)
-
-    async def _handle_raw_cdr_ingestion(
-        self, writer: asyncio.StreamWriter, body: bytes
-    ) -> None:
-        # Check broker reachability
-        if not self.producer.is_healthy():
-            err = json.dumps(
-                {
-                    "error": "broker_unavailable",
-                    "detail": "Kafka broker is unreachable",
-                }
-            ).encode("utf-8")
-            await self._send_response(writer, 503, "application/json", err)
-            return
-
-        # Check queue capacity (Backpressure)
-        if self.queue.full():
-            err = json.dumps(
-                {
-                    "error": "queue_full",
-                    "detail": "Internal async buffer is saturated (backpressure)",
-                }
-            ).encode("utf-8")
-            await self._send_response(writer, 429, "application/json", err)
-            return
-
-        # Validate JSON format
         try:
-            body_text = body.decode("utf-8")
+            payload = parse_payload(body)
-            payload = json.loads(body_text)
+            validate_structure(payload)
-        except json.JSONDecodeError as e:
+        except PayloadError as exc:
-            err = json.dumps(
+            await self._error(writer, 400, exc.reason, str(exc))
-                {
-                    "error": "invalid_json",
-                    "detail": f"Unexpected token at position {e.pos}: {e.msg}",
-                }
-            ).encode("utf-8")
-            await self._send_response(writer, 400, "application/json", err)
             return
-
+        if self._stopping:
-        if not isinstance(payload, dict):
+            await self._error(writer, 503, "shutting_down", "API is draining")
-            err = json.dumps(
+            return
-                {"error": "invalid_payload", "detail": "Expected a JSON object"}
+        if len(self._pending) >= self.queue_max_size:
-            ).encode("utf-8")
+            await self._error(writer, 429, "queue_full", "Outstanding publication capacity exhausted")
-            await self._send_response(writer, 400, "application/json", err)
             return
-
-        raw_id = payload.get("raw_id")
-
-        # Put in bounded queue without blocking
         try:
-            self.queue.put_nowait((raw_id, body))
+            delivery = asyncio.wrap_future(self.producer.produce(config.kafka.raw_topic, payload["raw_id"], body))
-        except asyncio.QueueFull:
+            self._pending.add(delivery)
-            err = json.dumps(
+            def completed(future):
-                {
+                self._pending.discard(future)
-                    "error": "queue_full",
+                if not future.cancelled():
-                    "detail": "Internal async buffer is saturated (backpressure)",
+                    future.exception()  # Consume failures after an HTTP timeout/disconnect.
-                }
+            delivery.add_done_callback(completed)
-            ).encode("utf-8")
+            await asyncio.wait_for(asyncio.shield(delivery), self.delivery_timeout)
-            await self._send_response(writer, 429, "application/json", err)
+        except Exception as exc:
+            logger.warning(str(exc), extra={"raw_id": payload["raw_id"], "event": "kafka_delivery_failed"})
+            await self._error(writer, 503, "broker_unavailable", "Delivery unconfirmed; retry with the same raw_id and payload")
             return
-
-        # Increment metric
         api_metrics.increment_received(1)
+        await self._json(writer, 202, {"status": "accepted", "raw_id": payload["raw_id"], "queued": True})
 
-        # 202 Accepted
+    async def _error(self, writer, status, error, detail):
-        res = json.dumps(
+        await self._json(writer, status, {"error": error, "detail": detail})
-            {"status": "accepted", "raw_id": raw_id, "queued": True}
-        ).encode("utf-8")
-        await self._send_response(writer, 202, "application/json", res)
 
-    async def _send_response(
+    async def _json(self, writer, status, payload):
-        self,
+        await self._send_response(writer, status, "application/json", json.dumps(payload).encode())
-        writer: asyncio.StreamWriter,
+
-        status_code: int,
+    async def _send_response(self, writer, status_code, content_type, body):
-        content_type: str,
+        from http import HTTPStatus
-        body: bytes,
+        headers = (f"HTTP/1.1 {status_code} {HTTPStatus(status_code).phrase}\r\n"
-    ) -> None:
+                   f"Content-Type: {content_type}\r\nContent-Length: {len(body)}\r\n"
-        status_reasons = {
+                   "Connection: close\r\n\r\n")
-            200: "OK",
+        writer.write(headers.encode() + body)
-            202: "Accepted",
-            400: "Bad Request",
-            404: "Not Found",
-            429: "Too Many Requests",
-            500: "Internal Server Error",
-            503: "Service Unavailable",
-        }
-        reason = status_reasons.get(status_code, "Unknown")
-        headers = (
-            f"HTTP/1.1 {status_code} {reason}\r\n"
-            f"Content-Type: {content_type}\r\n"
-            f"Content-Length: {len(body)}\r\n"
-            f"Connection: close\r\n"
-            f"\r\n"
-        )
-        writer.write(headers.encode("utf-8") + body)
         await writer.drain()
 
 
 async def main():
     server = IngestionAPIServer()
+    stopping = asyncio.Event()
+    loop = asyncio.get_running_loop()
+    for sig in (signal.SIGTERM, signal.SIGINT):
+        try:
+            loop.add_signal_handler(sig, stopping.set)
+        except NotImplementedError:
+            signal.signal(sig, lambda *_: loop.call_soon_threadsafe(stopping.set))
     await server.start()
     try:
-        # Keep running until cancelled
+        await stopping.wait()
-        while True:
+    finally:
-            await asyncio.sleep(3600)
-    except (asyncio.CancelledError, KeyboardInterrupt):
         await server.stop()
 
 

app/common/config.py

@@ -25,6 +25,9 @@ class APIConfig:
     host: str = get_env_str("API_HOST", "0.0.0.0")
     port: int = get_env_int("API_PORT", 8000)
     queue_max_size: int = get_env_int("API_QUEUE_MAX_SIZE", 10000)
+    max_body_bytes: int = get_env_int("API_MAX_BODY_BYTES", 65536)
+    read_timeout_seconds: float = get_env_float("API_READ_TIMEOUT_SECONDS", 10)
+    shutdown_timeout_seconds: float = get_env_float("SHUTDOWN_TIMEOUT_SECONDS", 45)
     worker_stats_url: str = get_env_str(
         "WORKER_STATS_URL", "http://worker:8001/internal/stats"
     )
@@ -44,6 +47,7 @@ class KafkaConfig:
     dlq_topic: str = get_env_str("KAFKA_DLQ_TOPIC", "cdr.raw.dlq")
     consumer_group: str = get_env_str("KAFKA_CONSUMER_GROUP", "cdr-processing-group")
     auto_offset_reset: str = get_env_str("KAFKA_AUTO_OFFSET_RESET", "earliest")
+    delivery_timeout_seconds: float = get_env_float("KAFKA_DELIVERY_TIMEOUT_SECONDS", 10)
 
 
 @dataclass(frozen=True)

app/worker/dlq.py

@@ -28,6 +28,8 @@ def build_dlq_envelope(
     """
     Constructs a standardized DLQ envelope according to specification.
     """
+    if failure_reason not in ALLOWED_FAILURE_REASONS:
+        raise ValueError(f"Unsupported DLQ reason: {failure_reason}")
     return {
         "original_payload": original_payload,
         "failure_reason": failure_reason,
@@ -56,11 +58,13 @@ def route_to_dlq(
     )
     envelope_bytes = json.dumps(envelope, default=str).encode("utf-8")
 
-    success = producer.produce(topic=topic, key=raw_id, value=envelope_bytes)
+    producer.produce(topic=topic, key=raw_id, value=envelope_bytes).result(
+        timeout=config.kafka.delivery_timeout_seconds + 1
+    )
     worker_metrics.record_dlq(failure_reason, 1)
 
     logger.warning(
         f"Record routed to DLQ: reason={failure_reason}, field={failed_field}",
         extra={"raw_id": raw_id, "event": "routed_to_dlq"},
     )
-    return success
+    return True

tests/test_api_delivery.py

@@ -0,0 +1,72 @@
+import asyncio
+import json
+import unittest
+from concurrent.futures import Future
+
+from app.api.producer import MockMessageProducer
+from app.api.server import IngestionAPIServer
+
+
+def payload():
+    return dict(raw_id="delivery-test", device_id="123", caller_num="+19875551234",
+                svc="4G_DATA", rat=6, lac="1", ci="2", up_bytes=10,
+                down_bytes=20, epoch_start=1726070000, epoch_end=1726070300)
+
+
+class PendingProducer(MockMessageProducer):
+    def __init__(self):
+        super().__init__()
+        self.pending = []
+
+    def produce(self, topic, key, value):
+        future = Future()
+        self.pending.append(future)
+        return future
+
+
+class Writer:
+    def __init__(self):
+        self.data = b""
+
+    def write(self, value):
+        self.data += value
+
+    async def drain(self):
+        pass
+
+
+class TestAcknowledgedAPI(unittest.IsolatedAsyncioTestCase):
+    async def test_response_waits_for_delivery_and_bounds_outstanding(self):
+        producer = PendingProducer()
+        server = IngestionAPIServer(producer=producer, queue_max_size=1)
+        writer = Writer()
+        task = asyncio.create_task(server._handle_raw_cdr_ingestion(writer, json.dumps(payload()).encode()))
+        await asyncio.sleep(0.02)
+        self.assertEqual(writer.data, b"", "202 must await the broker")
+        other = Writer()
+        await server._handle_raw_cdr_ingestion(other, json.dumps(payload()).encode())
+        self.assertIn(b"429", other.data)
+        producer.pending[0].set_result(None)
+        await task
+        self.assertIn(b"202", writer.data)
+
+    async def test_missing_fields_wrong_types_and_utf8_return_400(self):
+        server = IngestionAPIServer(producer=MockMessageProducer())
+        for body in (b"{}", b"\xff", json.dumps(payload() | {"raw_id": 42}).encode(),
+                     json.dumps(payload() | {"rat": True}).encode()):
+            with self.subTest(body=body):
+                writer = Writer()
+                await server._handle_raw_cdr_ingestion(writer, body)
+                self.assertIn(b"400", writer.data)
+
+    async def test_failed_delivery_returns_503_and_next_request_recovers(self):
+        producer = MockMessageProducer()
+        server = IngestionAPIServer(producer=producer)
+        producer.set_healthy(False)
+        writer = Writer()
+        await server._handle_raw_cdr_ingestion(writer, json.dumps(payload()).encode())
+        self.assertIn(b"503", writer.data)
+        producer.set_healthy(True)
+        writer = Writer()
+        await server._handle_raw_cdr_ingestion(writer, json.dumps(payload()).encode())
+        self.assertIn(b"202", writer.data)

tests/test_api_server.py

@@ -129,28 +129,22 @@ class TestIngestionAPIServer(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(data["error"], "invalid_payload")
 
     async def test_backpressure_returns_429_when_queue_full(self):
-        # Cancel dispatcher task so it won't consume from queue
+        from concurrent.futures import Future
-        self.server.dispatcher._task.cancel()
+        from tests.test_api_delivery import payload
-
+        futures = [asyncio.get_running_loop().create_future() for _ in range(2)]
-        # Fill the queue (maxsize=2)
+        self.server._pending.update(futures)
-        await self.server.queue.put(("k1", b"v1"))
+        body = json.dumps(payload()).encode()
-        await self.server.queue.put(("k2", b"v2"))
+        status, _, resp_body = await self._http_request("POST", "/api/v1/cdr/raw", body=body)
-
-        # Third request must be rejected with 429
-        body = json.dumps({"raw_id": "k3"}).encode("utf-8")
-        status, _, resp_body = await self._http_request(
-            "POST",
-            "/api/v1/cdr/raw",
-            body=body,
-            headers={"Content-Type": "application/json"},
-        )
         self.assertEqual(status, 429)
-        data = json.loads(resp_body.decode("utf-8"))
+        self.assertEqual(json.loads(resp_body)["error"], "queue_full")
-        self.assertEqual(data["error"], "queue_full")
+        for future in futures:
+            future.set_result(None)
+        self.server._pending.clear()
 
     async def test_broker_unavailable_returns_503(self):
         self.mock_producer.set_healthy(False)
-        body = json.dumps({"raw_id": "test"}).encode("utf-8")
+        from tests.test_api_delivery import payload
+        body = json.dumps(payload()).encode("utf-8")
         status, _, resp_body = await self._http_request(
             "POST",
             "/api/v1/cdr/raw",

tests/test_http_boundaries.py

@@ -0,0 +1,152 @@
+"""Exercise HTTP framing and broker acknowledgement through real TCP sockets."""
+import asyncio
+import json
+import unittest
+from concurrent.futures import Future
+
+from app.api.server import IngestionAPIServer
+
+
+class PendingProducer:
+    def __init__(self):
+        self.deliveries = []
+        self.produced = asyncio.Event()
+        self.closed = False
+
+    def produce(self, topic, key, value):
+        future = Future()
+        self.deliveries.append(future)
+        self.produced.set()
+        return future
+
+    def is_healthy(self):
+        return True
+
+    def flush(self, timeout=5):
+        return sum(not future.done() for future in self.deliveries)
+
+    def close(self, timeout=5):
+        self.closed = True
+
+
+class HTTPBoundaryTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.producer = PendingProducer()
+        self.server = None
+        self.writers = []
+        self.tasks = []
+
+    async def asyncTearDown(self):
+        for future in self.producer.deliveries:
+            if not future.done():
+                future.set_result(None)
+        for writer in self.writers:
+            writer.close()
+            await writer.wait_closed()
+        if self.server is not None:
+            await asyncio.wait_for(self.server.stop(), timeout=2)
+        for task in self.tasks:
+            if not task.done():
+                task.cancel()
+        await asyncio.gather(*self.tasks, return_exceptions=True)
+
+    async def start_server(self, **options):
+        self.server = IngestionAPIServer(host="127.0.0.1", port=0,
+                                          producer=self.producer, **options)
+        await self.server.start()
+        self.port = self.server._server.sockets[0].getsockname()[1]
+
+    async def send(self, request):
+        reader, writer = await asyncio.open_connection("127.0.0.1", self.port)
+        self.writers.append(writer)
+        writer.write(request)
+        await writer.drain()
+        task = asyncio.create_task(reader.read())
+        self.tasks.append(task)
+        return task
+
+    def valid_request(self):
+        body = json.dumps(dict(raw_id="http-boundary", device_id="123",
+                               caller_num="0019875551234", svc="4G_DATA", rat=6,
+                               up_bytes=100, down_bytes=200, lac="1", ci="2",
+                               epoch_start=1726070000, epoch_end=1726070300)).encode()
+        return (b"POST /api/v1/cdr/raw HTTP/1.1\r\nHost: localhost\r\n"
+                + f"Content-Length: {len(body)}\r\n\r\n".encode() + body)
+
+    async def assert_response(self, task, status):
+        response = await asyncio.wait_for(asyncio.shield(task), timeout=2)
+        headers, body = response.split(b"\r\n\r\n", 1)
+        self.assertEqual(int(headers.split(b" ")[1]), status, response)
+        decoded = json.loads(body)
+        if status >= 400:
+            self.assertIn("error", decoded)
+        return decoded
+
+    async def test_malformed_framing_returns_structured_400(self):
+        await self.start_server()
+        headers = [b"Content-Length: invalid", b"Content-Length: -1",
+                   b"Content-Length: 2\r\nContent-Length: 3",
+                   b"Transfer-Encoding: chunked",
+                   b"Transfer-Encoding: chunked\r\nContent-Length: 2"]
+        for header in headers:
+            with self.subTest(header=header):
+                task = await self.send(b"POST /api/v1/cdr/raw HTTP/1.1\r\n" + header + b"\r\n\r\n{}")
+                await self.assert_response(task, 400)
+        self.assertEqual(self.producer.deliveries, [])
+
+    async def test_oversized_declared_body_is_rejected_before_reading(self):
+        await self.start_server(max_body_bytes=32)
+        task = await self.send(b"POST /api/v1/cdr/raw HTTP/1.1\r\nContent-Length: 33\r\n\r\n")
+        await self.assert_response(task, 413)
+        self.assertEqual(self.producer.deliveries, [])
+
+    async def test_incomplete_body_times_out_with_structured_408(self):
+        await self.start_server(read_timeout=0.03)
+        task = await self.send(b"POST /api/v1/cdr/raw HTTP/1.1\r\nContent-Length: 20\r\n\r\n{")
+        await self.assert_response(task, 408)
+        self.assertEqual(self.producer.deliveries, [])
+
+    async def test_response_waits_for_broker_acknowledgement(self):
+        await self.start_server(delivery_timeout=1)
+        task = await self.send(self.valid_request())
+        await asyncio.wait_for(self.producer.produced.wait(), timeout=1)
+        await asyncio.sleep(0.03)
+        self.assertFalse(task.done(), "HTTP accepted before broker acknowledgement")
+        self.producer.deliveries[0].set_result(None)
+        await self.assert_response(task, 202)
+
+    async def test_timeout_holds_capacity_until_delivery_finishes(self):
+        await self.start_server(queue_max_size=1, delivery_timeout=0.03)
+        first = await self.send(self.valid_request())
+        await self.assert_response(first, 503)
+        self.assertEqual(len(self.producer.deliveries), 1)
+        self.assertFalse(self.producer.deliveries[0].cancelled())
+        await self.assert_response(await self.send(self.valid_request()), 429)
+        self.producer.deliveries[0].set_result(None)
+        await asyncio.sleep(0.03)
+        self.server.delivery_timeout = 1.0
+        self.producer.produced.clear()
+        third = await self.send(self.valid_request())
+        await asyncio.wait_for(self.producer.produced.wait(), timeout=1)
+        self.assertEqual(len(self.producer.deliveries), 2)
+        self.producer.deliveries[1].set_result(None)
+        await self.assert_response(third, 202)
+
+    async def test_shutdown_waits_for_outstanding_acknowledgement(self):
+        await self.start_server(delivery_timeout=1)
+        request = await self.send(self.valid_request())
+        await asyncio.wait_for(self.producer.produced.wait(), timeout=1)
+        stop = asyncio.create_task(self.server.stop())
+        self.tasks.append(stop)
+        await asyncio.sleep(0.03)
+        self.assertFalse(stop.done(), "Shutdown discarded an outstanding publication")
+        self.assertFalse(self.producer.closed)
+        self.producer.deliveries[0].set_result(None)
+        await asyncio.wait_for(stop, timeout=2)
+        await self.assert_response(request, 202)
+        self.assertTrue(self.producer.closed)
+        self.server = None
+
+
+if __name__ == "__main__":
+    unittest.main()