From 33593609e316697872bf52b22b8ea70366c5b5b9 Mon Sep 17 00:00:00 2001 From: Amirhossein Khalili Date: Sat, 19 Sep 2026 21:30:27 +0330 Subject: [PATCH] fix(api): await Kafka acknowledgements and bound HTTP requests --- app/api/producer.py | 204 +++++------------- app/api/server.py | 383 +++++++++++++--------------------- app/common/config.py | 4 + app/worker/dlq.py | 8 +- tests/test_api_delivery.py | 72 +++++++ tests/test_api_server.py | 30 ++- tests/test_http_boundaries.py | 152 ++++++++++++++ 7 files changed, 447 insertions(+), 406 deletions(-) create mode 100644 tests/test_api_delivery.py create mode 100644 tests/test_http_boundaries.py diff --git a/app/api/producer.py b/app/api/producer.py index 8d4bdbd..0cee827 100644 --- a/app/api/producer.py +++ b/app/api/producer.py @@ -1,185 +1,91 @@ -import asyncio -import logging +"""Kafka publications complete only on broker delivery callbacks.""" +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: - """Produce a message synchronously or enqueue in broker buffer.""" - pass + def produce(self, topic: str, key: str | None, value: bytes) -> Future: + """Return a future resolved by the destination acknowledgement.""" @abstractmethod - def is_healthy(self) -> bool: - """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.""" + def close(self, timeout: float = 5.0) -> None: pass class ConfluentKafkaProducer(MessageProducer): - """ - Production Kafka producer implementation wrapping confluent-kafka. - Uses 'raw_id' as partition key and LZ4 compression. - """ - - 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, + def __init__(self, bootstrap_servers=None): + from confluent_kafka import Producer + self._producer = Producer({ + "bootstrap.servers": bootstrap_servers or config.kafka.bootstrap_servers, "compression.type": "lz4", - "queue.buffering.max.messages": 100000, - "queue.buffering.max.ms": 5, + "queue.buffering.max.messages": config.api.queue_max_size, + "linger.ms": 5, "acks": "all", - "retries": 3, - } - self._producer = self._ck.Producer(self._producer_config) - self._healthy = True + "enable.idempotence": True, + "delivery.timeout.ms": int(config.kafka.delivery_timeout_seconds * 1000), + }) + 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: - if err is not None: - self._healthy = False - 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 _poll(self): + while not self._stop.is_set(): + self._producer.poll(0.01) - 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 - self._producer.produce( - topic=topic, - key=key_bytes, - value=value, - on_delivery=self._delivery_report, - ) - # 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 + if self._closed: + raise RuntimeError("Producer is closed") + self._producer.produce(topic, key=key, value=value, on_delivery=delivered) + except Exception as exc: + future.set_exception(exc) + return future - def is_healthy(self) -> bool: - return self._healthy - - def flush(self, timeout: float = 5.0) -> int: + def flush(self, timeout=5.0): return self._producer.flush(timeout) - def close(self) -> None: - self.flush() + def close(self, timeout=5.0): + 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: - if not self._healthy: - return False - self.messages.append({"topic": topic, "key": key, "value": value}) - return True + def produce(self, topic, key, value): + future = Future() + if self._healthy: + self.messages.append({"topic": topic, "key": key, "value": value}) + 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() diff --git a/app/api/server.py b/app/api/server.py index 4142542..fae84c4 100644 --- a/app/api/server.py +++ b/app/api/server.py @@ -1,285 +1,194 @@ import asyncio import json -import logging -from typing import Optional, Tuple +import signal from app.api.metrics import api_metrics -from app.api.producer import ( - AsyncProducerWorker, - ConfluentKafkaProducer, - MessageProducer, - MockMessageProducer, -) +from app.api.producer import ConfluentKafkaProducer 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: - """ - Framework-less HTTP/1.1 Ingestion API server built on standard library asyncio. - Enforces bounded async queue backpressure and dispatches to Kafka. - """ - - def __init__( - self, - host: Optional[str] = None, - port: Optional[int] = None, - queue_max_size: Optional[int] = None, - producer: Optional[MessageProducer] = None, - 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 + def __init__(self, host=None, port=None, queue_max_size=None, producer=None, + worker_stats_url=None, max_body_bytes=None, read_timeout=None, + 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 + self.queue_max_size = queue_max_size if queue_max_size is not None else config.api.queue_max_size + self.max_body_bytes = max_body_bytes if max_body_bytes is not None else config.api.max_body_bytes + self.read_timeout = read_timeout if read_timeout is not None else config.api.read_timeout_seconds + self.delivery_timeout = delivery_timeout if delivery_timeout is not None else config.kafka.delivery_timeout_seconds + if min(self.queue_max_size, self.max_body_bytes, self.read_timeout, self.delivery_timeout) <= 0: + raise ValueError("API capacities and timeouts must be positive") 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) - self.producer = producer or ConfluentKafkaProducer() - self.dispatcher = AsyncProducerWorker( - queue=self.queue, - producer=self.producer, - topic=config.kafka.raw_topic, - ) + async def start(self): + self._server = await asyncio.start_server(self._handle_client, self.host, self.port) + self.port = self._server.sockets[0].getsockname()[1] + logger.info(f"Listening on {self.host}:{self.port}", extra={"event": "server_started"}) - self._server: Optional[asyncio.Server] = None - self._running = False - - 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 + async def stop(self): + self._stopping = True if self._server: self._server.close() await self._server.wait_closed() - await self.dispatcher.stop() - self.producer.close() - logger.info("Ingestion API stopped cleanly", extra={"event": "server_stopped"}) + deadline = asyncio.get_running_loop().time() + config.api.shutdown_timeout_seconds + outstanding = self._pending | self._clients + 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( - self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter - ) -> None: + async def _read_request(self, reader): + line = await reader.readline() + if not line: + return None try: - # Read request line - request_line = await reader.readline() - if not request_line: - writer.close() - await writer.wait_closed() - return + method, path, version = line.decode("ascii").strip().split() + except (ValueError, UnicodeError) as exc: + raise HTTPError(400, "bad_request", "Invalid request line") from exc + if version not in ("HTTP/1.0", "HTTP/1.1"): + raise HTTPError(400, "bad_request", "Unsupported HTTP version") + 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() - parts = line_str.split(" ") - if len(parts) < 2: - await self._send_response( - writer, 400, "application/json", b'{"error": "bad_request"}' - ) - return - - method = parts[0].upper() - path = parts[1] - - # Read headers - headers = {} - while True: - header_line = await reader.readline() - 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: + async def _handle_client(self, reader, writer): + task = asyncio.current_task() + self._clients.add(task) + try: + async with asyncio.timeout(self.read_timeout): + request = await self._read_request(reader) + if request: + await self._dispatch_route(writer, *request) + except TimeoutError: + await self._error(writer, 408, "request_timeout", "Request read deadline exceeded") + except HTTPError as exc: + await self._error(writer, exc.status, exc.reason, str(exc)) + except (asyncio.IncompleteReadError, ValueError) as exc: + await self._error(writer, 400, "bad_request", str(exc)) + except (ConnectionError, BrokenPipeError): pass - except Exception as e: - logger.error( - f"Error handling HTTP request: {e}", - 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 + except Exception: + logger.exception("HTTP handler failure", extra={"event": "request_handler_error"}) + await self._error(writer, 500, "internal_error", "Request failed") finally: + self._clients.discard(task) + writer.close() try: - writer.close() await writer.wait_closed() - except Exception: + except ConnectionError: pass - async def _dispatch_route( - self, - writer: asyncio.StreamWriter, - method: str, - path: str, - headers: dict, - body: bytes, - ) -> None: - # 1. Health check + async def _dispatch_route(self, writer, method, path, headers, body): if method == "GET" and path in ("/health", "/healthz", "/live"): - res = json.dumps( - {"status": "ok", "queue_size": self.queue.qsize()} - ).encode("utf-8") - await self._send_response(writer, 200, "application/json", res) - return - - # 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._json(writer, 200, {"status": "ok", "in_flight": len(self._pending)}) + elif method == "GET" and path == "/metrics": + await self._send_response(writer, 200, "text/plain; version=0.0.4; charset=utf-8", api_metrics.get_prometheus_metrics().encode()) + elif method == "GET" and path == "/stats": + await self._json(writer, 200, await get_aggregated_pipeline_stats(self.worker_stats_url)) + elif 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 - 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 + async def _handle_raw_cdr_ingestion(self, writer, body): try: - body_text = body.decode("utf-8") - payload = json.loads(body_text) - except json.JSONDecodeError as e: - err = json.dumps( - { - "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) + payload = parse_payload(body) + validate_structure(payload) + except PayloadError as exc: + await self._error(writer, 400, exc.reason, str(exc)) return - - if not isinstance(payload, dict): - err = json.dumps( - {"error": "invalid_payload", "detail": "Expected a JSON object"} - ).encode("utf-8") - await self._send_response(writer, 400, "application/json", err) + if self._stopping: + await self._error(writer, 503, "shutting_down", "API is draining") + return + if len(self._pending) >= self.queue_max_size: + await self._error(writer, 429, "queue_full", "Outstanding publication capacity exhausted") return - - raw_id = payload.get("raw_id") - - # Put in bounded queue without blocking try: - self.queue.put_nowait((raw_id, body)) - except asyncio.QueueFull: - 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) + delivery = asyncio.wrap_future(self.producer.produce(config.kafka.raw_topic, payload["raw_id"], body)) + self._pending.add(delivery) + def completed(future): + self._pending.discard(future) + if not future.cancelled(): + future.exception() # Consume failures after an HTTP timeout/disconnect. + delivery.add_done_callback(completed) + await asyncio.wait_for(asyncio.shield(delivery), self.delivery_timeout) + 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 - res = json.dumps( - {"status": "accepted", "raw_id": raw_id, "queued": True} - ).encode("utf-8") - await self._send_response(writer, 202, "application/json", res) + async def _error(self, writer, status, error, detail): + await self._json(writer, status, {"error": error, "detail": detail}) - async def _send_response( - self, - writer: asyncio.StreamWriter, - status_code: int, - content_type: str, - body: bytes, - ) -> None: - status_reasons = { - 200: "OK", - 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) + async def _json(self, writer, status, payload): + await self._send_response(writer, status, "application/json", json.dumps(payload).encode()) + + async def _send_response(self, writer, status_code, content_type, body): + from http import HTTPStatus + headers = (f"HTTP/1.1 {status_code} {HTTPStatus(status_code).phrase}\r\n" + f"Content-Type: {content_type}\r\nContent-Length: {len(body)}\r\n" + "Connection: close\r\n\r\n") + writer.write(headers.encode() + 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 - while True: - await asyncio.sleep(3600) - except (asyncio.CancelledError, KeyboardInterrupt): + await stopping.wait() + finally: await server.stop() diff --git a/app/common/config.py b/app/common/config.py index 01bc1db..63ac681 100644 --- a/app/common/config.py +++ b/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) diff --git a/app/worker/dlq.py b/app/worker/dlq.py index 1281e6f..7548349 100644 --- a/app/worker/dlq.py +++ b/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 diff --git a/tests/test_api_delivery.py b/tests/test_api_delivery.py new file mode 100644 index 0000000..0fac57c --- /dev/null +++ b/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) diff --git a/tests/test_api_server.py b/tests/test_api_server.py index d84df1d..5251a80 100644 --- a/tests/test_api_server.py +++ b/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 - self.server.dispatcher._task.cancel() - - # Fill the queue (maxsize=2) - await self.server.queue.put(("k1", b"v1")) - await self.server.queue.put(("k2", b"v2")) - - # 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"}, - ) + from concurrent.futures import Future + from tests.test_api_delivery import payload + futures = [asyncio.get_running_loop().create_future() for _ in range(2)] + self.server._pending.update(futures) + body = json.dumps(payload()).encode() + status, _, resp_body = await self._http_request("POST", "/api/v1/cdr/raw", body=body) self.assertEqual(status, 429) - data = json.loads(resp_body.decode("utf-8")) - self.assertEqual(data["error"], "queue_full") + self.assertEqual(json.loads(resp_body)["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", diff --git a/tests/test_http_boundaries.py b/tests/test_http_boundaries.py new file mode 100644 index 0000000..6b54160 --- /dev/null +++ b/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()