test: add end-to-end and load verification

.dockerignore

@@ -0,0 +1,22 @@
+.git
+.gitignore
+
+.venv
+venv
+
+__pycache__
+*.py[cod]
+
+.pytest_cache
+.coverage
+htmlcov
+
+.env
+
+.vscode
+.idea
+
+tests
+scripts
+
+README.md

.env.sample

@@ -58,10 +58,53 @@ KAFKA_PRODUCER_MAX_MESSAGES=100000
 
 CLICKHOUSE_VERSION=26.8.7.19
 
-CLICKHOUSE_HOST=clickhouse
+# Used by Python applications running directly on the host.
+CLICKHOUSE_HOST=localhost
+
+# Used later by Dockerized API/worker services.
+CLICKHOUSE_INTERNAL_HOST=clickhouse
+
 CLICKHOUSE_PORT=8123
 CLICKHOUSE_HTTP_HOST_PORT=8123
 
 CLICKHOUSE_DATABASE=telecom
 CLICKHOUSE_USER=telecom
 CLICKHOUSE_PASSWORD=telecom_dev_password
+
+
+# =========================
+# ClickHouse Reliability
+# =========================
+
+CLICKHOUSE_MAX_INSERT_ATTEMPTS=3
+CLICKHOUSE_RETRY_BASE_DELAY_SECONDS=0.5
+
+CLICKHOUSE_CONNECT_TIMEOUT_SECONDS=5
+CLICKHOUSE_SEND_RECEIVE_TIMEOUT_SECONDS=10
+
+# =========================
+# Worker batching
+# =========================
+
+BATCH_SIZE=1000
+BATCH_FLUSH_INTERVAL_SECONDS=2
+
+
+# =========================
+# Worker Observability
+# =========================
+
+WORKER_HTTP_HOST=0.0.0.0
+WORKER_HTTP_PORT=8001
+
+# API uses this while both services run directly on the host.
+WORKER_STATS_URL=http://localhost:8001/internal/stats
+WORKER_STATS_TIMEOUT_SECONDS=2
+
+
+# =========================
+# Docker Host Ports
+# =========================
+
+API_HOST_PORT=8000
+WORKER_HTTP_HOST_PORT=8001

Dockerfile.api

@@ -0,0 +1,24 @@
+FROM python:3.12-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1
+
+WORKDIR /app
+
+COPY pyproject.toml ./
+COPY src ./src
+
+RUN pip install --upgrade pip \
+    && pip install .
+
+RUN useradd \
+    --create-home \
+    --uid 10001 \
+    appuser
+
+USER appuser
+
+EXPOSE 8000
+
+CMD ["uvicorn", "core.api.main:app", "--host", "0.0.0.0", "--port", "8000"]

Dockerfile.worker

@@ -0,0 +1,24 @@
+FROM python:3.12-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1
+
+WORKDIR /app
+
+COPY pyproject.toml ./
+COPY src ./src
+
+RUN pip install --upgrade pip \
+    && pip install .
+
+RUN useradd \
+    --create-home \
+    --uid 10001 \
+    appuser
+
+USER appuser
+
+EXPOSE 8001
+
+CMD ["python", "-m", "core.worker.main"]

README.md

@@ -28,8 +28,12 @@ Redpanda / cdr.raw
 Processing Worker
     ↓
 Validation + Normalization
-    ├── valid → processed by worker
-    └── invalid → cdr.raw.dlq
+    ├── invalid → cdr.raw.dlq
+    └── valid
+          ↓
+      Micro-Batch
+          ↓
+      ClickHouse
 ```
 
 Infrastructure already available:
@@ -72,6 +76,28 @@ worker processing is not connected yet.
 * DLQ routing for malformed and unprocessable Kafka records
 * DLQ records with failure reason, failed field, timestamp, and worker instance
 * Kafka offsets committed only after the current processing outcome succeeds
+* ClickHouse bulk storage through the official Python client
+* Configurable CDR micro-batching
+* Flush on 1,000 records or 2-second timeout
+* Kafka offsets committed only after successful ClickHouse batch writes
+* Remaining worker buffer flushed during graceful shutdown
+* ClickHouse insert retries with exponential backoff
+* Failed ClickHouse batches routed to the DLQ after three attempts
+* Whole-batch offset commits with no partial Kafka acknowledgement
+* Replay-safe storage using deterministic UUIDv5 identifiers and `ReplacingMergeTree`
+* Graceful worker shutdown flushes buffered records before exit
+* Separate Prometheus `/metrics` endpoints for the API and worker
+* API accepted-CDR counter
+* Worker processed and DLQ counters
+* Kafka-to-ClickHouse and batch flush latency histograms
+* Batch size and in-flight buffer metrics
+* Kafka consumer lag by partition
+* Aggregated human-readable `/stats` endpoint
+* Fully containerized API and processing worker
+* Health-based Docker Compose startup ordering
+* Automatic Kafka topic and ClickHouse schema provisioning
+* Persistent Redpanda and ClickHouse storage
+* Complete stack startup with `docker compose up --build`
 
 ## Project Structure
 
@@ -94,8 +120,10 @@ telecom-cdr-pipeline/
 │       │   ├── __init__.py
 │       │   ├── kafka_publisher.py
 │       │   ├── main.py
+│       │   ├── metrics.py
 │       │   ├── publisher.py
-│       │   └── routes.py
+│       │   ├── routes.py
+│       │   └── stats.py
 │       │
 │       ├── domain/
 │       │   ├── __init__.py
@@ -106,21 +134,28 @@ telecom-cdr-pipeline/
 │       │
 │       └── worker/
 │           ├── __init__.py
+│           ├── batch.py
+│           ├── clickhouse.py
 │           ├── consumer.py
 │           ├── dlq.py
 │           ├── main.py
-│           └── processor.py
+│           ├── metrics.py
+│           ├── observability.py
+│           ├── processor.py
+│           └── stats.py
 │
 ├── tests/
 │   ├── integration/
 │   │   └── test_ingestion_api.py
 │   │
 │   └── unit/
+│       ├── test_batch.py
 │       ├── test_dlq.py
 │       ├── test_normalization.py
 │       ├── test_project_setup.py
 │       ├── test_validation.py
-│       └── test_worker_processor.py
+│       ├── test_worker_processor.py
+│       └── test_worker_stats.py.py
 │
 ├── .env.example
 ├── .gitignore
@@ -140,10 +175,10 @@ This includes:
 * ClickHouse connection settings
 * Worker batch size and flush interval
 
-## Running Infrastructure
+## Running Pipeline
 
 ```bash
-docker compose up -d
+docker compose up -d --build
 ```
 
 Verify services:
@@ -160,17 +195,23 @@ redpanda-init  exited successfully
 ClickHouse     healthy
 ```
 
-## Running the API
+### Available endpoints:
 
-```bash
-uvicorn core.api.main:app --reload --host 0.0.0.0 --port 8000
-```
-
-Available endpoints:
+#### API:
 
 ```text
-GET  /health
-POST /api/v1/cdr/raw
+GET  http://localhost:8000/health
+GET  http://localhost:8000/metrics
+GET  http://localhost:8000/stats
+POST http://localhost:8000/api/v1/cdr/raw
+```
+
+#### Worker:
+
+```text
+GET  http://localhost:8001/health
+GET  http://localhost:8001/metrics
+GET  http://localhost:8001/internal/stats
 ```
 
 A valid CDR currently returns:
@@ -186,10 +227,16 @@ Invalid payloads return structured `400` responses.
 
 ## Testing
 
-Run:
+Once the full stack starts cleanly, a real CDR reaches ClickHouse, and:
 
 ```bash
 pytest -v
 ```
 
-The current test suite covers domain validation, normalization, deterministic ID generation, and API request validation.
+## Delivery Semantics
+
+The worker uses at-least-once delivery.
+
+Kafka offsets are committed only after a ClickHouse batch is successfully written or, after exhausted retries, the complete failed batch is acknowledged by the DLQ.
+
+Repeated records use deterministic UUIDv5 identifiers and `ReplacingMergeTree` to support replay-safe storage.

docker-compose.yml

@@ -2,34 +2,23 @@ name: telecom-cdr-pipeline
 
 services:
   redpanda:
-    image: docker.redpanda.com/redpandadata/redpanda:${REDPANDA_VERSION:-latest}
-
+    image: docker.redpanda.com/redpandadata/redpanda:${REDPANDA_VERSION:-v26.2.3}
     container_name: telecom-redpanda
 
     command:
       - redpanda
       - start
-
-      # Kafka listeners:
-      # 9092  -> other Docker containers
-      # 19092 -> applications running on our host machine
       - --kafka-addr=internal://0.0.0.0:9092,external://0.0.0.0:19092
-
-      # These are the addresses Redpanda tells Kafka clients to use.
       - --advertise-kafka-addr=internal://redpanda:9092,external://localhost:19092
-
-      # Internal Redpanda node communication.
       - --rpc-addr=redpanda:33145
       - --advertise-rpc-addr=redpanda:33145
-
-      # Suitable settings for local Docker development.
       - --mode=dev-container
       - --smp=1
       - --default-log-level=info
 
     ports:
       - "19092:19092"
-      - "9644:9644"  # Redpanda's Admin API
+      - "9644:9644"
 
     volumes:
       - redpanda-data:/var/lib/redpanda/data
@@ -48,28 +37,34 @@ services:
       retries: 10
       start_period: 10s
 
+    restart: unless-stopped
+
+
   redpanda-init:
-    image: docker.redpanda.com/redpandadata/redpanda:${REDPANDA_VERSION:-latest}
+    image: docker.redpanda.com/redpandadata/redpanda:${REDPANDA_VERSION:-v26.2.3}
+    container_name: telecom-redpanda-init
 
     depends_on:
       redpanda:
         condition: service_healthy
 
     environment:
-      KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_INTERNAL_BOOTSTRAP_SERVERS}
+      KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_INTERNAL_BOOTSTRAP_SERVERS:-redpanda:9092}
 
-      KAFKA_RAW_TOPIC: ${KAFKA_RAW_TOPIC}
-      KAFKA_RAW_TOPIC_PARTITIONS: ${KAFKA_RAW_TOPIC_PARTITIONS}
-      KAFKA_RAW_TOPIC_REPLICATION_FACTOR: ${KAFKA_RAW_TOPIC_REPLICATION_FACTOR}
-      KAFKA_RAW_TOPIC_RETENTION_MS: ${KAFKA_RAW_TOPIC_RETENTION_MS}
-      KAFKA_RAW_TOPIC_COMPRESSION: ${KAFKA_RAW_TOPIC_COMPRESSION}
+      KAFKA_RAW_TOPIC: ${KAFKA_RAW_TOPIC:-cdr.raw}
+      KAFKA_RAW_TOPIC_PARTITIONS: ${KAFKA_RAW_TOPIC_PARTITIONS:-3}
+      KAFKA_RAW_TOPIC_REPLICATION_FACTOR: ${KAFKA_RAW_TOPIC_REPLICATION_FACTOR:-1}
+      KAFKA_RAW_TOPIC_RETENTION_MS: ${KAFKA_RAW_TOPIC_RETENTION_MS:-86400000}
+      KAFKA_RAW_TOPIC_COMPRESSION: ${KAFKA_RAW_TOPIC_COMPRESSION:-lz4}
 
-      KAFKA_DLQ_TOPIC: ${KAFKA_DLQ_TOPIC}
-      KAFKA_DLQ_TOPIC_PARTITIONS: ${KAFKA_DLQ_TOPIC_PARTITIONS}
-      KAFKA_DLQ_TOPIC_REPLICATION_FACTOR: ${KAFKA_DLQ_TOPIC_REPLICATION_FACTOR}
-      KAFKA_DLQ_TOPIC_RETENTION_MS: ${KAFKA_DLQ_TOPIC_RETENTION_MS}
+      KAFKA_DLQ_TOPIC: ${KAFKA_DLQ_TOPIC:-cdr.raw.dlq}
+      KAFKA_DLQ_TOPIC_PARTITIONS: ${KAFKA_DLQ_TOPIC_PARTITIONS:-1}
+      KAFKA_DLQ_TOPIC_REPLICATION_FACTOR: ${KAFKA_DLQ_TOPIC_REPLICATION_FACTOR:-1}
+      KAFKA_DLQ_TOPIC_RETENTION_MS: ${KAFKA_DLQ_TOPIC_RETENTION_MS:-604800000}
 
-    entrypoint: ["/bin/bash", "-c"]
+    entrypoint:
+      - /bin/bash
+      - -c
 
     command:
       - |
@@ -92,15 +87,18 @@ services:
 
         echo "Kafka topics initialized successfully."
 
-  clickhouse:
-    image: clickhouse/clickhouse-server:${CLICKHOUSE_VERSION:-latest}
+    restart: "no"
 
+
+  clickhouse:
+    image: clickhouse/clickhouse-server:${CLICKHOUSE_VERSION:-26.8.7.19}
     container_name: telecom-clickhouse
 
     environment:
-      CLICKHOUSE_DB: ${CLICKHOUSE_DATABASE}
-      CLICKHOUSE_USER: ${CLICKHOUSE_USER}
-      CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD}
+      CLICKHOUSE_DB: ${CLICKHOUSE_DATABASE:-telecom}
+      CLICKHOUSE_USER: ${CLICKHOUSE_USER:-telecom}
+      CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-telecom_dev_password}
+
       CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT: 1
       CLICKHOUSE_ALWAYS_RUN_INITDB_SCRIPTS: 1
 
@@ -127,6 +125,128 @@ services:
       retries: 10
       start_period: 10s
 
+    restart: unless-stopped
+
+
+  api:
+    build:
+      context: .
+      dockerfile: Dockerfile.api
+
+    container_name: telecom-api
+
+    depends_on:
+      redpanda:
+        condition: service_healthy
+
+      redpanda-init:
+        condition: service_completed_successfully
+
+    environment:
+      KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_INTERNAL_BOOTSTRAP_SERVERS:-redpanda:9092}
+      KAFKA_RAW_TOPIC: ${KAFKA_RAW_TOPIC:-cdr.raw}
+
+      API_QUEUE_SIZE: ${API_QUEUE_SIZE:-10000}
+
+      KAFKA_DELIVERY_TIMEOUT_SECONDS: ${KAFKA_DELIVERY_TIMEOUT_SECONDS:-4}
+      KAFKA_MESSAGE_TIMEOUT_MS: ${KAFKA_MESSAGE_TIMEOUT_MS:-3000}
+
+      KAFKA_LINGER_MS: ${KAFKA_LINGER_MS:-5}
+      KAFKA_BATCH_SIZE: ${KAFKA_BATCH_SIZE:-131072}
+      KAFKA_PRODUCER_MAX_MESSAGES: ${KAFKA_PRODUCER_MAX_MESSAGES:-100000}
+
+      WORKER_STATS_URL: http://worker:8001/internal/stats
+      WORKER_STATS_TIMEOUT_SECONDS: ${WORKER_STATS_TIMEOUT_SECONDS:-2}
+
+    ports:
+      - "${API_HOST_PORT:-8000}:8000"
+
+    healthcheck:
+      test:
+        [
+          "CMD",
+          "python",
+          "-c",
+          "import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=2)"
+        ]
+      interval: 5s
+      timeout: 3s
+      retries: 10
+      start_period: 5s
+
+    restart: unless-stopped
+
+
+  worker:
+    build:
+      context: .
+      dockerfile: Dockerfile.worker
+
+    container_name: telecom-worker
+
+    depends_on:
+      redpanda:
+        condition: service_healthy
+
+      redpanda-init:
+        condition: service_completed_successfully
+
+      clickhouse:
+        condition: service_healthy
+
+    environment:
+      KAFKA_BOOTSTRAP_SERVERS: ${KAFKA_INTERNAL_BOOTSTRAP_SERVERS:-redpanda:9092}
+
+      KAFKA_RAW_TOPIC: ${KAFKA_RAW_TOPIC:-cdr.raw}
+      KAFKA_DLQ_TOPIC: ${KAFKA_DLQ_TOPIC:-cdr.raw.dlq}
+
+      KAFKA_CONSUMER_GROUP: ${KAFKA_CONSUMER_GROUP:-cdr-processing-workers}
+      KAFKA_AUTO_OFFSET_RESET: ${KAFKA_AUTO_OFFSET_RESET:-earliest}
+      KAFKA_CONSUMER_POLL_TIMEOUT_SECONDS: ${KAFKA_CONSUMER_POLL_TIMEOUT_SECONDS:-1}
+
+      KAFKA_DLQ_DELIVERY_TIMEOUT_SECONDS: ${KAFKA_DLQ_DELIVERY_TIMEOUT_SECONDS:-5}
+      KAFKA_MESSAGE_TIMEOUT_MS: ${KAFKA_MESSAGE_TIMEOUT_MS:-3000}
+
+      WORKER_INSTANCE_ID: ${WORKER_INSTANCE_ID:-worker-1}
+
+      WORKER_HTTP_HOST: 0.0.0.0
+      WORKER_HTTP_PORT: 8001
+
+      CLICKHOUSE_HOST: ${CLICKHOUSE_INTERNAL_HOST:-clickhouse}
+      CLICKHOUSE_PORT: ${CLICKHOUSE_PORT:-8123}
+
+      CLICKHOUSE_DATABASE: ${CLICKHOUSE_DATABASE:-telecom}
+      CLICKHOUSE_USER: ${CLICKHOUSE_USER:-telecom}
+      CLICKHOUSE_PASSWORD: ${CLICKHOUSE_PASSWORD:-telecom_dev_password}
+
+      BATCH_SIZE: ${BATCH_SIZE:-1000}
+      BATCH_FLUSH_INTERVAL_SECONDS: ${BATCH_FLUSH_INTERVAL_SECONDS:-2}
+
+      CLICKHOUSE_MAX_INSERT_ATTEMPTS: ${CLICKHOUSE_MAX_INSERT_ATTEMPTS:-3}
+      CLICKHOUSE_RETRY_BASE_DELAY_SECONDS: ${CLICKHOUSE_RETRY_BASE_DELAY_SECONDS:-0.5}
+
+      CLICKHOUSE_CONNECT_TIMEOUT_SECONDS: ${CLICKHOUSE_CONNECT_TIMEOUT_SECONDS:-5}
+      CLICKHOUSE_SEND_RECEIVE_TIMEOUT_SECONDS: ${CLICKHOUSE_SEND_RECEIVE_TIMEOUT_SECONDS:-10}
+
+    ports:
+      - "${WORKER_HTTP_HOST_PORT:-8001}:8001"
+
+    healthcheck:
+      test:
+        [
+          "CMD",
+          "python",
+          "-c",
+          "import urllib.request; urllib.request.urlopen('http://localhost:8001/health', timeout=2)"
+        ]
+      interval: 5s
+      timeout: 3s
+      retries: 10
+      start_period: 5s
+
+    restart: unless-stopped
+
+
 volumes:
   redpanda-data:
-  clickhouse-data:
+  clickhouse-data:

pyproject.toml

@@ -9,12 +9,14 @@ dependencies = [
     "uvicorn[standard]>=0.30,<1.0",
     "confluent-kafka>=2.15,<3.0",
     "pydantic-settings>=2.0,<3.0",
+    "clickhouse-connect>=1.8,<2.0",
+    "prometheus-client>=0.26,<1.0",
+    "httpx2>=2.4,<3.0",
 ]
 
 [project.optional-dependencies]
 dev = [
     "pytest>=8.0",
-    "httpx2>=2.4,<3.0",
 ]
 
 [build-system]

scripts/functional_traffic.py

@@ -0,0 +1,242 @@
+import json
+import os
+import random
+import time
+import uuid
+from collections import Counter
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from copy import deepcopy
+from urllib.error import HTTPError, URLError
+from urllib.request import Request, urlopen
+
+
+BASE_URL = os.getenv(
+    "BASE_URL",
+    "http://localhost:8000",
+)
+
+ENDPOINT = f"{BASE_URL}/api/v1/cdr/raw"
+
+RUN_ID = os.getenv(
+    "RUN_ID",
+    uuid.uuid4().hex[:8],
+)
+
+
+def valid_payload(index: int) -> dict:
+    start = int(time.time())
+
+    return {
+        "raw_id": f"functional-{RUN_ID}-{index:04d}",
+        "src_ip": "10.240.12.1",
+        "device_id": "310410998877665",
+        "caller_num": "+14155552671",
+        "svc": "4G_DATA",
+        "up_bytes": 1024,
+        "down_bytes": 4096,
+        "rat": 6,
+        "lac": "4021",
+        "ci": "8812",
+        "epoch_start": start,
+        "epoch_end": start + 300,
+    }
+
+
+def send_json(payload: dict) -> int | str:
+    body = json.dumps(
+        payload
+    ).encode("utf-8")
+
+    return send_raw(body)
+
+
+def send_raw(body: bytes) -> int | str:
+    request = Request(
+        ENDPOINT,
+        data=body,
+        method="POST",
+        headers={
+            "Content-Type": "application/json",
+        },
+    )
+
+    try:
+        with urlopen(
+            request,
+            timeout=10,
+        ) as response:
+            return response.status
+
+    except HTTPError as exc:
+        return exc.code
+
+    except URLError as exc:
+        return f"network_error:{exc.reason}"
+
+
+def main() -> None:
+    requests: list[
+        tuple[str, dict | bytes]
+    ] = []
+
+    valid_records = [
+        valid_payload(index)
+        for index in range(1000)
+    ]
+
+    # 1000 valid CDRs
+    for payload in valid_records:
+        requests.append(
+            ("valid", payload)
+        )
+
+    # 20 missing caller_num
+    for index in range(20):
+        payload = valid_payload(
+            2000 + index
+        )
+
+        payload.pop(
+            "caller_num"
+        )
+
+        requests.append(
+            ("missing_caller_num", payload)
+        )
+
+    # 10 malformed JSON requests
+    for _ in range(10):
+        requests.append(
+            (
+                "invalid_json",
+                b'{"raw_id": invalid}',
+            )
+        )
+
+    # 10 invalid RAT values
+    for index in range(10):
+        payload = valid_payload(
+            3000 + index
+        )
+
+        payload["rat"] = 99
+
+        requests.append(
+            ("invalid_rat", payload)
+        )
+
+    # 10 invalid epochs
+    for index in range(10):
+        payload = valid_payload(
+            4000 + index
+        )
+
+        payload["epoch_start"] = (
+            payload["epoch_end"] + 100
+        )
+
+        requests.append(
+            ("invalid_epoch", payload)
+        )
+
+    # 20 duplicates of already accepted valid records.
+    for payload in valid_records[:20]:
+        requests.append(
+            (
+                "duplicate",
+                deepcopy(payload),
+            )
+        )
+
+    random.shuffle(
+        requests
+    )
+
+    results = Counter()
+
+    started_at = time.perf_counter()
+
+    with ThreadPoolExecutor(
+        max_workers=50
+    ) as executor:
+        futures = []
+
+        for request_type, payload in requests:
+            if isinstance(
+                payload,
+                bytes,
+            ):
+                future = executor.submit(
+                    send_raw,
+                    payload,
+                )
+            else:
+                future = executor.submit(
+                    send_json,
+                    payload,
+                )
+
+            futures.append(
+                (
+                    request_type,
+                    future,
+                )
+            )
+
+        for request_type, future in futures:
+            status = future.result()
+
+            results[
+                (
+                    request_type,
+                    status,
+                )
+            ] += 1
+
+    elapsed = (
+        time.perf_counter()
+        - started_at
+    )
+
+    print()
+    print(
+        f"Run ID: {RUN_ID}"
+    )
+
+    print(
+        f"Total requests: {len(requests)}"
+    )
+
+    print(
+        f"Elapsed: {elapsed:.2f}s"
+    )
+
+    print()
+
+    for (
+        request_type,
+        status,
+    ), count in sorted(
+        results.items(),
+        key=lambda item: str(
+            item[0]
+        ),
+    ):
+        print(
+            f"{request_type:24} "
+            f"status={status!s:8} "
+            f"count={count}"
+        )
+
+    print()
+    print("Expected:")
+    print("  valid             → 1000 x 202")
+    print("  duplicate         →   20 x 202")
+    print("  missing_caller    →   20 x 400")
+    print("  invalid_json      →   10 x 400")
+    print("  invalid_rat       →   10 x 400")
+    print("  invalid_epoch     →   10 x 400")
+
+
+if __name__ == "__main__":
+    main()

scripts/load_test.js

@@ -0,0 +1,131 @@
+import http from 'k6/http';
+import { check } from 'k6';
+import { Counter } from 'k6/metrics';
+
+
+const accepted = new Counter(
+    'cdr_accepted'
+);
+
+const backpressure = new Counter(
+    'cdr_backpressure'
+);
+
+const unavailable = new Counter(
+    'cdr_unavailable'
+);
+
+const unexpected = new Counter(
+    'cdr_unexpected'
+);
+
+
+const RATE = Number(
+    __ENV.RATE || 1000
+);
+
+const PRE_ALLOCATED_VUS = Number(
+    __ENV.PRE_ALLOCATED_VUS || 250
+);
+
+const MAX_VUS = Number(
+    __ENV.MAX_VUS || 1000
+);
+
+const DURATION = (
+    __ENV.DURATION || '10s'
+);
+
+const BASE_URL = (
+    __ENV.BASE_URL ||
+    'http://localhost:8000'
+);
+
+
+export const options = {
+    discardResponseBodies: true,
+
+    scenarios: {
+        ingestion: {
+            executor: 'constant-arrival-rate',
+
+            rate: RATE,
+            timeUnit: '1s',
+
+            duration: DURATION,
+
+            preAllocatedVUs: PRE_ALLOCATED_VUS,
+            maxVUs: MAX_VUS,
+        },
+    },
+};
+
+
+export default function () {
+    const now = Math.floor(
+        Date.now() / 1000
+    );
+
+    const rawId =
+        `k6-${__VU}-${__ITER}-${Date.now()}`;
+
+    const payload = JSON.stringify({
+        raw_id: rawId,
+
+        src_ip: '10.240.12.1',
+
+        device_id:
+            '310410998877665',
+
+        caller_num:
+            '+14155552671',
+
+        svc: '4G_DATA',
+
+        up_bytes: 1024,
+        down_bytes: 4096,
+
+        rat: 6,
+
+        lac: '4021',
+        ci: '8812',
+
+        epoch_start: now,
+        epoch_end: now + 300,
+    });
+
+    const response = http.post(
+        `${BASE_URL}/api/v1/cdr/raw`,
+        payload,
+        {
+            headers: {
+                'Content-Type':
+                    'application/json',
+            },
+
+            tags: {
+                endpoint: 'cdr_ingest',
+            },
+        }
+    );
+
+    if (response.status === 202) {
+        accepted.add(1);
+
+    } else if (response.status === 429) {
+        backpressure.add(1);
+
+    } else if (response.status === 503) {
+        unavailable.add(1);
+
+    } else {
+        unexpected.add(1);
+    }
+
+    check(response, {
+        'accepted or gracefully rejected':
+            (r) =>
+                r.status === 202 ||
+                r.status === 429,
+    });
+}

src/core/api/main.py

@@ -1,27 +1,61 @@
 from contextlib import asynccontextmanager
 
 from fastapi import FastAPI
+from fastapi.responses import JSONResponse, Response
+
+from prometheus_client import CONTENT_TYPE_LATEST
 
 from core.api.kafka_publisher import KafkaPublisher
 from core.api.publisher import Publisher
 from core.api.routes import router
 from core.config import get_settings
+from core.api.metrics import ApiMetrics
+from core.api.stats import (
+    ApiRuntimeStats,
+    HttpWorkerStatsClient,
+    WorkerStatsProvider,
+    WorkerStatsUnavailable,
+)
 
 
 def create_app(
     publisher: Publisher | None = None,
+    worker_stats_client: WorkerStatsProvider | None = None,
 ) -> FastAPI:
+    settings = get_settings()
+
+    api_metrics = ApiMetrics()
+    api_stats = ApiRuntimeStats()
+
     @asynccontextmanager
     async def lifespan(app: FastAPI):
         active_publisher = publisher
 
         if active_publisher is None:
             active_publisher = KafkaPublisher(
-                get_settings()
+                settings
+            )
+
+        active_worker_stats_client = (
+            worker_stats_client
+        )
+
+        if active_worker_stats_client is None:
+            active_worker_stats_client = (
+                HttpWorkerStatsClient(
+                    url=settings.worker_stats_url,
+                    timeout_seconds=(
+                        settings.worker_stats_timeout_seconds
+                    ),
+                )
             )
 
         app.state.publisher = active_publisher
 
+        app.state.worker_stats_client = (
+            active_worker_stats_client
+        )
+
         await active_publisher.start()
 
         try:
@@ -30,12 +64,17 @@ def create_app(
         finally:
             await active_publisher.stop()
 
+            await active_worker_stats_client.close()
+
     app = FastAPI(
         title="Telecom CDR Ingestion API",
         version="0.1.0",
         lifespan=lifespan,
     )
 
+    app.state.api_metrics = api_metrics
+    app.state.api_stats = api_stats
+
     app.include_router(
         router,
         prefix="/api/v1",
@@ -47,6 +86,62 @@ def create_app(
             "status": "ok",
         }
 
+    @app.get(
+        "/metrics",
+        include_in_schema=False,
+    )
+    async def metrics() -> Response:
+        return Response(
+            content=api_metrics.render(),
+            media_type=CONTENT_TYPE_LATEST,
+        )
+
+    @app.get("/stats")
+    async def stats():
+        try:
+            worker_stats = await (
+                app.state
+                .worker_stats_client
+                .get_stats()
+            )
+
+        except WorkerStatsUnavailable as exc:
+            return JSONResponse(
+                status_code=503,
+                content={
+                    "error": "worker_stats_unavailable",
+                    "detail": str(exc),
+                },
+            )
+
+        api_snapshot = api_stats.snapshot()
+
+        worker_pipeline = (
+            worker_stats["pipeline"]
+        )
+
+        return {
+            "pipeline": {
+                "total_received":
+                    api_snapshot["total_received"],
+
+                "total_processed":
+                    worker_pipeline["total_processed"],
+
+                "total_dlq":
+                    worker_pipeline["total_dlq"],
+
+                "in_flight_buffer":
+                    worker_pipeline["in_flight_buffer"],
+
+                "last_flush_at":
+                    worker_pipeline["last_flush_at"],
+            },
+
+            "breakdown":
+                worker_stats["breakdown"],
+        }
+
     return app
 
 

src/core/api/metrics.py

@@ -0,0 +1,24 @@
+from prometheus_client import (
+    CollectorRegistry,
+    Counter,
+    generate_latest,
+)
+
+
+class ApiMetrics:
+    def __init__(self) -> None:
+        self.registry = CollectorRegistry()
+
+        self._received = Counter(
+            "cdr_api_received_total",
+            "Total CDRs successfully accepted by the ingestion API.",
+            registry=self.registry,
+        )
+
+    def record_received(self) -> None:
+        self._received.inc()
+
+    def render(self) -> bytes:
+        return generate_latest(
+            self.registry
+        )

src/core/api/routes.py

@@ -74,6 +74,9 @@ async def ingest_raw_cdr(request: Request) -> JSONResponse:
             },
         )
 
+    request.app.state.api_metrics.record_received()
+    request.app.state.api_stats.record_received()
+
     return JSONResponse(
         status_code=status.HTTP_202_ACCEPTED,
         content={

src/core/api/stats.py

@@ -0,0 +1,67 @@
+import threading
+from typing import Any, Protocol
+
+import httpx2
+
+
+class ApiRuntimeStats:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._total_received = 0
+
+    def record_received(self) -> None:
+        with self._lock:
+            self._total_received += 1
+
+    def snapshot(self) -> dict[str, int]:
+        with self._lock:
+            return {
+                "total_received": self._total_received,
+            }
+
+
+class WorkerStatsUnavailable(Exception):
+    pass
+
+
+class WorkerStatsProvider(Protocol):
+    async def get_stats(
+        self,
+    ) -> dict[str, Any]:
+        ...
+
+    async def close(self) -> None:
+        ...
+
+
+class HttpWorkerStatsClient:
+    def __init__(
+        self,
+        url: str,
+        timeout_seconds: float,
+    ) -> None:
+        self._url = url
+
+        self._client = httpx2.AsyncClient(
+            timeout=timeout_seconds,
+        )
+
+    async def get_stats(
+        self,
+    ) -> dict[str, Any]:
+        try:
+            response = await self._client.get(
+                self._url
+            )
+
+            response.raise_for_status()
+
+            return response.json()
+
+        except httpx2.HTTPError as exc:
+            raise WorkerStatsUnavailable(
+                f"Unable to retrieve worker stats: {exc}"
+            ) from exc
+
+    async def close(self) -> None:
+        await self._client.aclose()

src/core/config.py

@@ -31,6 +31,28 @@ class Settings(BaseSettings):
 
     worker_instance_id: str = "worker-1"
 
+    clickhouse_host: str = "localhost"
+    clickhouse_port: int = 8123
+    clickhouse_database: str = "telecom"
+    clickhouse_user: str = "telecom"
+    clickhouse_password: str = ""
+
+    clickhouse_max_insert_attempts: int = 3
+    clickhouse_retry_base_delay_seconds: float = 0.5
+
+    clickhouse_connect_timeout_seconds: int = 5
+    clickhouse_send_receive_timeout_seconds: int = 10
+
+    batch_size: int = 1_000
+    batch_flush_interval_seconds: float = 2.0
+
+    worker_http_host: str = "0.0.0.0"
+    worker_http_port: int = 8001
+
+    worker_stats_url: str = "http://localhost:8001/internal/stats"
+    worker_stats_timeout_seconds: float = 2.0
+
+
 @lru_cache
 def get_settings() -> Settings:
     return Settings()

src/core/worker/batch.py

@@ -0,0 +1,76 @@
+import time
+from dataclasses import dataclass
+from typing import Any
+
+from core.domain.models import NormalizedCDR
+
+
+@dataclass(frozen=True, slots=True)
+class KafkaPosition:
+    topic: str
+    partition: int
+    offset: int
+
+
+@dataclass(frozen=True, slots=True)
+class BufferedCDR:
+    cdr: NormalizedCDR
+    original_payload: dict[str, Any]
+    position: KafkaPosition
+    kafka_timestamp_ms: int | None
+
+
+class MicroBatch:
+    def __init__(
+        self,
+        max_size: int,
+        flush_interval_seconds: float,
+    ) -> None:
+        self._max_size = max_size
+        self._flush_interval_seconds = flush_interval_seconds
+
+        self._items: list[BufferedCDR] = []
+
+        self._started_at: float | None = None
+
+    def add(
+        self,
+        item: BufferedCDR,
+    ) -> None:
+        if not self._items:
+            self._started_at = time.monotonic()
+
+        self._items.append(item)
+
+    @property
+    def size(self) -> int:
+        return len(self._items)
+
+    @property
+    def is_empty(self) -> bool:
+        return not self._items
+
+    @property
+    def is_full(self) -> bool:
+        return len(self._items) >= self._max_size
+
+    def is_due(self) -> bool:
+        if not self._items:
+            return False
+
+        if self._started_at is None:
+            return False
+
+        elapsed = (
+            time.monotonic()
+            - self._started_at
+        )
+
+        return elapsed >= self._flush_interval_seconds
+
+    def snapshot(self) -> list[BufferedCDR]:
+        return list(self._items)
+
+    def clear(self) -> None:
+        self._items.clear()
+        self._started_at = None

src/core/worker/clickhouse.py

@@ -0,0 +1,167 @@
+from collections.abc import Sequence
+import logging
+import time
+
+import clickhouse_connect
+
+from core.config import Settings
+from core.domain.models import NormalizedCDR
+
+
+logger = logging.getLogger(__name__)
+
+
+class ClickHouseWriteError(Exception):
+    pass
+
+
+class ClickHouseWriter:
+    COLUMN_NAMES = [
+        "cdr_id",
+        "raw_id",
+        "msisdn",
+        "imsi",
+        "service_type",
+        "radio_access_type",
+        "cell_id",
+        "bytes_uploaded",
+        "bytes_downloaded",
+        "total_bytes",
+        "duration_seconds",
+        "session_start_time",
+        "session_end_time",
+        "source_ip",
+    ]
+
+    def __init__(
+        self,
+        settings: Settings,
+    ) -> None:
+        self._settings = settings
+
+        self._client = clickhouse_connect.get_client(
+            host=settings.clickhouse_host,
+            port=settings.clickhouse_port,
+            username=settings.clickhouse_user,
+            password=settings.clickhouse_password,
+            database=settings.clickhouse_database,
+
+            connect_timeout=(
+                settings.clickhouse_connect_timeout_seconds
+            ),
+            send_receive_timeout=(
+                settings.clickhouse_send_receive_timeout_seconds
+            ),
+        )
+
+    @staticmethod
+    def _to_row(
+        cdr: NormalizedCDR,
+    ) -> list:
+        return [
+            cdr.cdr_id,
+            cdr.raw_id,
+
+            cdr.msisdn,
+            cdr.imsi,
+
+            cdr.service_type,
+            cdr.radio_access_type,
+            cdr.cell_id,
+
+            cdr.bytes_uploaded,
+            cdr.bytes_downloaded,
+            cdr.total_bytes,
+            cdr.duration_seconds,
+
+            cdr.session_start_time,
+            cdr.session_end_time,
+
+            str(cdr.source_ip),
+        ]
+
+    def _insert_once(
+        self,
+        records: Sequence[NormalizedCDR],
+    ) -> None:
+        rows = [
+            self._to_row(record)
+            for record in records
+        ]
+
+        self._client.insert(
+            table="cdr_records",
+            data=rows,
+            column_names=self.COLUMN_NAMES,
+            database=self._settings.clickhouse_database,
+        )
+
+    def insert(
+        self,
+        records: Sequence[NormalizedCDR],
+    ) -> None:
+        if not records:
+            return
+
+        last_error: Exception | None = None
+
+        for attempt in range(
+            1,
+            self._settings.clickhouse_max_insert_attempts + 1,
+        ):
+            try:
+                self._insert_once(records)
+
+                if attempt > 1:
+                    logger.info(
+                        (
+                            "ClickHouse insert recovered. "
+                            "attempt=%s batch_size=%s"
+                        ),
+                        attempt,
+                        len(records),
+                    )
+
+                return
+
+            except Exception as exc:
+                last_error = exc
+
+                logger.warning(
+                    (
+                        "ClickHouse insert failed. "
+                        "attempt=%s/%s batch_size=%s error=%s"
+                    ),
+                    attempt,
+                    self._settings.clickhouse_max_insert_attempts,
+                    len(records),
+                    exc,
+                )
+
+                if (
+                    attempt
+                    >= self._settings.clickhouse_max_insert_attempts
+                ):
+                    break
+
+                delay = (
+                    self._settings.clickhouse_retry_base_delay_seconds
+                    * (2 ** (attempt - 1))
+                )
+
+                logger.info(
+                    "Retrying ClickHouse insert in %.2f seconds.",
+                    delay,
+                )
+
+                time.sleep(delay)
+
+        raise ClickHouseWriteError(
+            (
+                "ClickHouse batch insert failed after "
+                f"{self._settings.clickhouse_max_insert_attempts} attempts."
+            )
+        ) from last_error
+
+    def close(self) -> None:
+        self._client.close()

src/core/worker/consumer.py

@@ -1,21 +1,38 @@
 import logging
+import time
 from typing import Any
 
-from confluent_kafka import Consumer, KafkaException, Message
+from confluent_kafka import (
+    Consumer,
+    KafkaException,
+    Message,
+    TopicPartition,
+)
 
 from core.config import Settings
 from core.domain.errors import (
     CDRProcessingError,
     FailureReason,
 )
+from core.worker.batch import (
+    BufferedCDR,
+    KafkaPosition,
+    MicroBatch,
+)
+from core.worker.clickhouse import (
+    ClickHouseWriteError,
+    ClickHouseWriter,
+)
 from core.worker.dlq import (
     DLQPublishError,
     DLQPublisher,
 )
+from core.worker.metrics import WorkerMetrics
 from core.worker.processor import (
     decode_payload,
     process_payload,
 )
+from core.worker.stats import WorkerRuntimeStats
 
 
 logger = logging.getLogger(__name__)
@@ -25,8 +42,12 @@ class ProcessingWorker:
     def __init__(
         self,
         settings: Settings,
+        metrics: WorkerMetrics,
+        stats: WorkerRuntimeStats,
     ) -> None:
         self._settings = settings
+        self._metrics = metrics
+        self._stats = stats
 
         self._consumer = Consumer(
             {
@@ -42,11 +63,26 @@ class ProcessingWorker:
                 "auto.offset.reset":
                     settings.kafka_auto_offset_reset,
 
+                # We commit manually only after the processing
+                # outcome has safely completed.
                 "enable.auto.commit": False,
             }
         )
 
-        self._dlq = DLQPublisher(settings)
+        self._dlq = DLQPublisher(
+            settings
+        )
+
+        self._clickhouse = ClickHouseWriter(
+            settings
+        )
+
+        self._batch = MicroBatch(
+            max_size=settings.batch_size,
+            flush_interval_seconds=(
+                settings.batch_flush_interval_seconds
+            ),
+        )
 
         self._running = False
 
@@ -54,11 +90,16 @@ class ProcessingWorker:
         self._running = True
 
         self._consumer.subscribe(
-            [self._settings.kafka_raw_topic]
+            [
+                self._settings.kafka_raw_topic,
+            ]
         )
 
         logger.info(
-            "Worker started. topic=%s group=%s instance=%s",
+            (
+                "Worker started. "
+                "topic=%s group=%s instance=%s"
+            ),
             self._settings.kafka_raw_topic,
             self._settings.kafka_consumer_group,
             self._settings.worker_instance_id,
@@ -67,10 +108,12 @@ class ProcessingWorker:
         try:
             while self._running:
                 message = self._consumer.poll(
-                    self._settings.kafka_consumer_poll_timeout_seconds
+                    self._settings
+                    .kafka_consumer_poll_timeout_seconds
                 )
 
                 if message is None:
+                    self._flush_if_due()
                     continue
 
                 if message.error():
@@ -78,9 +121,15 @@ class ProcessingWorker:
                         "Kafka consumer error: %s",
                         message.error(),
                     )
+
+                    self._flush_if_due()
                     continue
 
-                self._handle_message(message)
+                self._handle_message(
+                    message
+                )
+
+                self._flush_if_due()
 
         finally:
             self._shutdown()
@@ -91,6 +140,8 @@ class ProcessingWorker:
     ) -> None:
         value = message.value() or b""
 
+        # If JSON decoding fails, we still want to preserve
+        # the original raw Kafka payload inside the DLQ.
         original_payload: Any = {
             "raw": value.decode(
                 "utf-8",
@@ -99,13 +150,28 @@ class ProcessingWorker:
         }
 
         try:
-            payload = decode_payload(value)
+            payload = decode_payload(
+                value
+            )
 
             original_payload = payload
 
-            normalized = process_payload(payload)
+            normalized = process_payload(
+                payload
+            )
 
         except CDRProcessingError as exc:
+            # Important:
+            #
+            # There may already be valid records from this
+            # partition sitting uncommitted in the batch.
+            #
+            # Before committing this invalid record after
+            # sending it to the DLQ, flush those earlier
+            # records first.
+            if not self._batch.is_empty:
+                self._flush_batch()
+
             self._route_to_dlq(
                 message=message,
                 original_payload=original_payload,
@@ -120,10 +186,17 @@ class ProcessingWorker:
             )
 
             error = CDRProcessingError(
-                reason=FailureReason.NORMALIZATION_ERROR,
-                detail=f"Unexpected processing error: {exc}",
+                reason=(
+                    FailureReason.NORMALIZATION_ERROR
+                ),
+                detail=(
+                    f"Unexpected processing error: {exc}"
+                ),
             )
 
+            if not self._batch.is_empty:
+                self._flush_batch()
+
             self._route_to_dlq(
                 message=message,
                 original_payload=original_payload,
@@ -132,20 +205,219 @@ class ProcessingWorker:
 
             return
 
-        logger.info(
-            (
-                "Normalized CDR. "
-                "raw_id=%s cdr_id=%s "
-                "service_type=%s rat=%s"
-            ),
-            normalized.raw_id,
-            normalized.cdr_id,
-            normalized.service_type,
-            normalized.radio_access_type,
+        position = KafkaPosition(
+            topic=message.topic(),
+            partition=message.partition(),
+            offset=message.offset(),
         )
 
-        # Temporary until ClickHouse batching is implemented.
-        self._commit(message)
+        _, kafka_timestamp_ms = (
+            message.timestamp()
+        )
+
+        if (
+            kafka_timestamp_ms is not None
+            and kafka_timestamp_ms < 0
+        ):
+            kafka_timestamp_ms = None
+
+        self._batch.add(
+            BufferedCDR(
+                cdr=normalized,
+                original_payload=payload,
+                position=position,
+                kafka_timestamp_ms=kafka_timestamp_ms,
+            )
+        )
+
+        self._metrics.set_buffer_size(
+            self._batch.size
+        )
+
+        self._stats.set_buffer_size(
+            self._batch.size
+        )
+
+        logger.debug(
+            (
+                "Buffered normalized CDR. "
+                "raw_id=%s "
+                "partition=%s "
+                "offset=%s "
+                "batch_size=%s"
+            ),
+            normalized.raw_id,
+            message.partition(),
+            message.offset(),
+            self._batch.size,
+        )
+
+        if self._batch.is_full:
+            self._flush_batch()
+
+    def _flush_if_due(
+        self,
+    ) -> None:
+        if self._batch.is_due():
+            self._flush_batch()
+
+    def _flush_batch(
+        self,
+    ) -> None:
+        items = self._batch.snapshot()
+
+        if not items:
+            return
+
+        records = [
+            item.cdr
+            for item in items
+        ]
+
+        positions = [
+            item.position
+            for item in items
+        ]
+
+        batch_size = len(
+            records
+        )
+
+        self._metrics.observe_batch_size(
+            batch_size
+        )
+
+        logger.info(
+            "Flushing ClickHouse batch. size=%s",
+            batch_size,
+        )
+
+        flush_started = time.monotonic()
+
+        try:
+            self._clickhouse.insert(
+                records
+            )
+
+        except ClickHouseWriteError as exc:
+            self._metrics.observe_batch_flush_duration(
+                time.monotonic()
+                - flush_started
+            )
+
+            logger.exception(
+                (
+                    "ClickHouse batch permanently "
+                    "failed after retries."
+                )
+            )
+
+            # The assessment requires an exhausted
+            # ClickHouse batch to be routed to the DLQ.
+            #
+            # _route_batch_to_dlq only returns if every
+            # DLQ publication was acknowledged.
+            self._route_batch_to_dlq(
+                items=items,
+                storage_error=exc,
+            )
+
+            # Every record now has a successful processing
+            # outcome via the DLQ, so the batch offsets can
+            # safely be committed.
+            self._commit_positions(
+                positions
+            )
+
+            self._batch.clear()
+
+            self._metrics.set_buffer_size(
+                0
+            )
+
+            self._stats.set_buffer_size(
+                0
+            )
+
+            logger.warning(
+                (
+                    "Failed ClickHouse batch routed "
+                    "to DLQ and offsets committed. "
+                    "size=%s"
+                ),
+                batch_size,
+            )
+
+            return
+
+        clickhouse_written_at = (
+            time.time()
+        )
+
+        self._metrics.observe_batch_flush_duration(
+            time.monotonic()
+            - flush_started
+        )
+
+        # Record end-to-end Kafka -> ClickHouse latency.
+        for item in items:
+            if (
+                item.kafka_timestamp_ms
+                is None
+            ):
+                continue
+
+            latency_seconds = (
+                clickhouse_written_at
+                - (
+                    item.kafka_timestamp_ms
+                    / 1000
+                )
+            )
+
+            self._metrics.observe_kafka_to_clickhouse_latency(
+                latency_seconds
+            )
+
+        # Critical ordering:
+        #
+        # ClickHouse ACK
+        #       ↓
+        # Kafka commit
+        #
+        # Never commit before the database write.
+        self._commit_positions(
+            positions
+        )
+
+        for item in items:
+            self._metrics.record_processed(
+                item.cdr
+            )
+
+            self._stats.record_processed(
+                item.cdr
+            )
+
+        self._stats.record_flush()
+
+        self._batch.clear()
+
+        self._metrics.set_buffer_size(
+            0
+        )
+
+        self._stats.set_buffer_size(
+            0
+        )
+
+        logger.info(
+            (
+                "ClickHouse batch flushed "
+                "successfully. size=%s"
+            ),
+            batch_size,
+        )
 
     def _route_to_dlq(
         self,
@@ -156,8 +428,10 @@ class ProcessingWorker:
         logger.warning(
             (
                 "Routing CDR to DLQ. "
-                "reason=%s field=%s "
-                "partition=%s offset=%s"
+                "reason=%s "
+                "field=%s "
+                "partition=%s "
+                "offset=%s"
             ),
             error.reason.value,
             error.failed_field,
@@ -173,43 +447,286 @@ class ProcessingWorker:
 
         except DLQPublishError:
             logger.exception(
-                "DLQ publishing failed. "
-                "Original Kafka offset will not be committed."
+                (
+                    "DLQ publishing failed. "
+                    "Original Kafka offset "
+                    "will not be committed."
+                )
             )
 
+            # Do not continue consuming and later commit
+            # beyond this record.
             raise
 
-        self._commit(message)
+        self._metrics.record_dlq(
+            error.reason.value
+        )
+
+        self._stats.record_dlq(
+            error.reason.value
+        )
+
+        # DLQ publication has been acknowledged,
+        # therefore this source Kafka record has a
+        # successful terminal outcome.
+        self._commit(
+            message
+        )
+
+    def _route_batch_to_dlq(
+        self,
+        items: list[BufferedCDR],
+        storage_error: ClickHouseWriteError,
+    ) -> None:
+        # The assessment restricts failure_reason to the
+        # existing enum values and does not define a
+        # dedicated storage_error reason.
+        error = CDRProcessingError(
+            reason=(
+                FailureReason.NORMALIZATION_ERROR
+            ),
+            detail=(
+                "ClickHouse batch insert failed "
+                "after retries: "
+                f"{storage_error}"
+            ),
+        )
+
+        logger.error(
+            (
+                "Routing failed ClickHouse batch "
+                "to DLQ. batch_size=%s"
+            ),
+            len(items),
+        )
+
+        for item in items:
+            # If any one of these fails, the exception
+            # propagates and the source Kafka offsets are
+            # NOT committed.
+            self._dlq.publish(
+                original_payload=(
+                    item.original_payload
+                ),
+                error=error,
+            )
+
+            self._metrics.record_dlq(
+                error.reason.value
+            )
+
+            self._stats.record_dlq(
+                error.reason.value
+            )
 
     def _commit(
         self,
         message: Message,
     ) -> None:
-        committed = self._consumer.commit(
-            message=message,
-            asynchronous=False,
+        committed = (
+            self._consumer.commit(
+                message=message,
+                asynchronous=False,
+            )
         )
 
-        for partition in committed or []:
-            if partition.error is not None:
+        for partition in (
+            committed or []
+        ):
+            if (
+                partition.error
+                is not None
+            ):
                 raise KafkaException(
                     partition.error
                 )
 
-        logger.debug(
-            "Committed Kafka offset. partition=%s offset=%s",
-            message.partition(),
-            message.offset(),
+        committed_offset = (
+            message.offset() + 1
+        )
+
+        self._update_consumer_lag(
+            topic=message.topic(),
+            partition=message.partition(),
+            committed_offset=committed_offset,
+        )
+
+        logger.debug(
+            (
+                "Committed Kafka offset. "
+                "partition=%s offset=%s"
+            ),
+            message.partition(),
+            committed_offset,
+        )
+
+    def _commit_positions(
+        self,
+        positions: list[KafkaPosition],
+    ) -> None:
+        # Only the highest successfully processed offset
+        # from each partition needs to be committed.
+        #
+        # Kafka committed offsets mean:
+        # "the NEXT offset to consume".
+        highest_offsets: dict[
+            tuple[str, int],
+            int,
+        ] = {}
+
+        for position in positions:
+            key = (
+                position.topic,
+                position.partition,
+            )
+
+            next_offset = (
+                position.offset + 1
+            )
+
+            current = (
+                highest_offsets.get(
+                    key
+                )
+            )
+
+            if (
+                current is None
+                or next_offset > current
+            ):
+                highest_offsets[
+                    key
+                ] = next_offset
+
+        offsets = [
+            TopicPartition(
+                topic,
+                partition,
+                offset,
+            )
+            for (
+                topic,
+                partition,
+            ), offset in (
+                highest_offsets.items()
+            )
+        ]
+
+        if not offsets:
+            return
+
+        committed = (
+            self._consumer.commit(
+                offsets=offsets,
+                asynchronous=False,
+            )
+        )
+
+        for partition in (
+            committed or []
+        ):
+            if (
+                partition.error
+                is not None
+            ):
+                raise KafkaException(
+                    partition.error
+                )
+
+            self._update_consumer_lag(
+                topic=partition.topic,
+                partition=partition.partition,
+                committed_offset=(
+                    partition.offset
+                ),
+            )
+
+        logger.debug(
+            "Committed Kafka batch offsets: %s",
+            offsets,
+        )
+
+    def _update_consumer_lag(
+        self,
+        topic: str,
+        partition: int,
+        committed_offset: int,
+    ) -> None:
+        try:
+            _, high_watermark = (
+                self._consumer
+                .get_watermark_offsets(
+                    TopicPartition(
+                        topic,
+                        partition,
+                    ),
+                    timeout=1.0,
+                )
+            )
+
+        except KafkaException:
+            logger.exception(
+                (
+                    "Unable to retrieve "
+                    "consumer lag. partition=%s"
+                ),
+                partition,
+            )
+
+            return
+
+        lag = max(
+            high_watermark
+            - committed_offset,
+            0,
+        )
+
+        self._metrics.set_consumer_lag(
+            partition=partition,
+            lag=lag,
+        )
+
+    def stop(
+        self,
+    ) -> None:
+        logger.info(
+            "Worker stop requested."
         )
 
-    def stop(self) -> None:
         self._running = False
 
-    def _shutdown(self) -> None:
-        logger.info("Worker shutting down.")
+    def _shutdown(
+        self,
+    ) -> None:
+        logger.info(
+            "Worker shutting down."
+        )
 
         try:
-            self._dlq.close()
+            # The assessment requires all remaining
+            # in-memory records to be flushed before exit.
+            if not self._batch.is_empty:
+                logger.info(
+                    (
+                        "Flushing remaining records "
+                        "before shutdown. size=%s"
+                    ),
+                    self._batch.size,
+                )
+
+                self._flush_batch()
 
         finally:
-            self._consumer.close()
+            try:
+                self._dlq.close()
+
+            finally:
+                try:
+                    self._clickhouse.close()
+
+                finally:
+                    self._consumer.close()
+
+        logger.info(
+            "Worker shutdown complete."
+        )

src/core/worker/main.py

@@ -3,7 +3,9 @@ import signal
 
 from core.config import get_settings
 from core.worker.consumer import ProcessingWorker
-
+from core.worker.metrics import WorkerMetrics
+from core.worker.observability import WorkerObservabilityServer
+from core.worker.stats import WorkerRuntimeStats
 
 logging.basicConfig(
     level=logging.INFO,
@@ -19,8 +21,20 @@ logging.basicConfig(
 def main() -> None:
     settings = get_settings()
 
+    metrics = WorkerMetrics()
+
+    stats = WorkerRuntimeStats()
+
     worker = ProcessingWorker(
         settings=settings,
+        metrics=metrics,
+        stats=stats,
+    )
+
+    observability = WorkerObservabilityServer(
+        settings=settings,
+        metrics=metrics,
+        stats=stats,
     )
 
     def handle_shutdown(
@@ -44,7 +58,13 @@ def main() -> None:
         handle_shutdown,
     )
 
-    worker.run()
+    observability.start()
+
+    try:
+        worker.run()
+
+    finally:
+        observability.stop()
 
 
 if __name__ == "__main__":

src/core/worker/metrics.py

@@ -0,0 +1,156 @@
+from prometheus_client import (
+    CollectorRegistry,
+    Counter,
+    Gauge,
+    Histogram,
+    generate_latest,
+)
+
+from core.domain.errors import FailureReason
+from core.domain.models import NormalizedCDR
+
+
+class WorkerMetrics:
+    def __init__(self) -> None:
+        self.registry = CollectorRegistry()
+
+        self._processed = Counter(
+            "cdr_processed_total",
+            "Total CDRs successfully processed and stored.",
+            labelnames=(
+                "service_type",
+                "radio_access_type",
+            ),
+            registry=self.registry,
+        )
+
+        self._dlq = Counter(
+            "cdr_dlq_total",
+            "Total CDRs routed to the DLQ.",
+            labelnames=(
+                "reason",
+            ),
+            registry=self.registry,
+        )
+
+        self._kafka_to_clickhouse_latency = Histogram(
+            "cdr_kafka_to_clickhouse_latency_seconds",
+            "Time from Kafka message creation to ClickHouse write.",
+            buckets=(
+                0.05,
+                0.1,
+                0.25,
+                0.5,
+                1,
+                2.5,
+                5,
+            ),
+            registry=self.registry,
+        )
+
+        self._batch_flush_duration = Histogram(
+            "cdr_batch_flush_duration_seconds",
+            "Duration of ClickHouse batch flushes.",
+            registry=self.registry,
+        )
+
+        self._batch_size = Histogram(
+            "cdr_batch_size_at_flush",
+            "Number of CDRs in each flushed batch.",
+            buckets=(
+                10,
+                50,
+                100,
+                250,
+                500,
+                1000,
+            ),
+            registry=self.registry,
+        )
+
+        self._buffer_size = Gauge(
+            "cdr_in_flight_buffer_size",
+            "Current number of CDRs buffered for ClickHouse.",
+            registry=self.registry,
+        )
+
+        self._consumer_lag = Gauge(
+            "cdr_kafka_consumer_lag",
+            "Kafka consumer lag by partition.",
+            labelnames=(
+                "partition",
+            ),
+            registry=self.registry,
+        )
+
+        # Ensure all required DLQ reasons appear even before
+        # their first failure.
+        for reason in FailureReason:
+            self._dlq.labels(
+                reason=reason.value
+            )
+
+    def record_processed(
+        self,
+        cdr: NormalizedCDR,
+    ) -> None:
+        self._processed.labels(
+            service_type=cdr.service_type,
+            radio_access_type=cdr.radio_access_type,
+        ).inc()
+
+    def record_dlq(
+        self,
+        reason: str,
+    ) -> None:
+        self._dlq.labels(
+            reason=reason
+        ).inc()
+
+    def observe_kafka_to_clickhouse_latency(
+        self,
+        seconds: float,
+    ) -> None:
+        self._kafka_to_clickhouse_latency.observe(
+            max(seconds, 0.0)
+        )
+
+    def observe_batch_flush_duration(
+        self,
+        seconds: float,
+    ) -> None:
+        self._batch_flush_duration.observe(
+            seconds
+        )
+
+    def observe_batch_size(
+        self,
+        size: int,
+    ) -> None:
+        self._batch_size.observe(
+            size
+        )
+
+    def set_buffer_size(
+        self,
+        size: int,
+    ) -> None:
+        self._buffer_size.set(
+            size
+        )
+
+    def set_consumer_lag(
+        self,
+        partition: int,
+        lag: int,
+    ) -> None:
+        self._consumer_lag.labels(
+            partition=str(partition)
+        ).set(
+            max(lag, 0)
+        )
+
+    def render(self) -> bytes:
+        return generate_latest(
+            self.registry
+        )

src/core/worker/observability.py

@@ -0,0 +1,98 @@
+import logging
+import threading
+
+import uvicorn
+from fastapi import FastAPI
+from fastapi.responses import Response
+from prometheus_client import CONTENT_TYPE_LATEST
+
+from core.config import Settings
+from core.worker.metrics import WorkerMetrics
+from core.worker.stats import WorkerRuntimeStats
+
+
+logger = logging.getLogger(__name__)
+
+
+def create_worker_observability_app(
+    metrics: WorkerMetrics,
+    stats: WorkerRuntimeStats,
+) -> FastAPI:
+    app = FastAPI(
+        title="Telecom CDR Worker Observability",
+        version="0.1.0",
+    )
+
+    @app.get(
+        "/health",
+        include_in_schema=False,
+    )
+    async def health() -> dict[str, str]:
+        return {
+            "status": "ok",
+        }
+
+    @app.get(
+        "/metrics",
+        include_in_schema=False,
+    )
+    async def metrics_endpoint() -> Response:
+        return Response(
+            content=metrics.render(),
+            media_type=CONTENT_TYPE_LATEST,
+        )
+
+    @app.get("/internal/stats")
+    async def internal_stats() -> dict:
+        return stats.snapshot()
+
+    return app
+
+
+class WorkerObservabilityServer:
+    def __init__(
+        self,
+        settings: Settings,
+        metrics: WorkerMetrics,
+        stats: WorkerRuntimeStats,
+    ) -> None:
+        app = create_worker_observability_app(
+            metrics=metrics,
+            stats=stats,
+        )
+
+        config = uvicorn.Config(
+            app=app,
+            host=settings.worker_http_host,
+            port=settings.worker_http_port,
+            log_level="warning",
+            access_log=False,
+        )
+
+        self._server = uvicorn.Server(
+            config=config
+        )
+
+        self._thread = threading.Thread(
+            target=self._server.run,
+            name="worker-observability",
+            daemon=True,
+        )
+
+    def start(self) -> None:
+        logger.info(
+            "Starting worker observability server."
+        )
+
+        self._thread.start()
+
+    def stop(self) -> None:
+        logger.info(
+            "Stopping worker observability server."
+        )
+
+        self._server.should_exit = True
+
+        self._thread.join(
+            timeout=5,
+        )

src/core/worker/stats.py

@@ -0,0 +1,104 @@
+import threading
+from collections import defaultdict
+from datetime import datetime, timezone
+from typing import Any
+
+from core.domain.models import NormalizedCDR
+
+
+class WorkerRuntimeStats:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+
+        self._total_processed = 0
+        self._total_dlq = 0
+
+        self._in_flight_buffer = 0
+
+        self._last_flush_at: str | None = None
+
+        self._by_service_type: dict[str, int] = defaultdict(int)
+
+        self._by_radio_access_type: dict[str, int] = defaultdict(int)
+
+        self._dlq_by_reason: dict[str, int] = defaultdict(int)
+
+    def record_processed(
+        self,
+        cdr: NormalizedCDR,
+    ) -> None:
+        with self._lock:
+            self._total_processed += 1
+
+            self._by_service_type[
+                cdr.service_type
+            ] += 1
+
+            self._by_radio_access_type[
+                cdr.radio_access_type
+            ] += 1
+
+    def record_dlq(
+        self,
+        reason: str,
+    ) -> None:
+        with self._lock:
+            self._total_dlq += 1
+
+            self._dlq_by_reason[
+                reason
+            ] += 1
+
+    def set_buffer_size(
+        self,
+        size: int,
+    ) -> None:
+        with self._lock:
+            self._in_flight_buffer = size
+
+    def record_flush(self) -> None:
+        now = (
+            datetime.now(timezone.utc)
+            .isoformat()
+            .replace("+00:00", "Z")
+        )
+
+        with self._lock:
+            self._last_flush_at = now
+
+    def snapshot(
+        self,
+    ) -> dict[str, Any]:
+        with self._lock:
+            return {
+                "pipeline": {
+                    "total_processed":
+                        self._total_processed,
+
+                    "total_dlq":
+                        self._total_dlq,
+
+                    "in_flight_buffer":
+                        self._in_flight_buffer,
+
+                    "last_flush_at":
+                        self._last_flush_at,
+                },
+
+                "breakdown": {
+                    "by_service_type":
+                        dict(
+                            self._by_service_type
+                        ),
+
+                    "by_radio_access_type":
+                        dict(
+                            self._by_radio_access_type
+                        ),
+
+                    "dlq_by_reason":
+                        dict(
+                            self._dlq_by_reason
+                        ),
+                },
+            }

tests/integration/test_ingestion_api.py

@@ -38,6 +38,35 @@ class FakePublisher:
         )
 
 
+class FakeWorkerStatsClient:
+    async def get_stats(
+        self,
+    ) -> dict[str, Any]:
+        return {
+            "pipeline": {
+                "total_processed": 10,
+                "total_dlq": 2,
+                "in_flight_buffer": 3,
+                "last_flush_at":
+                    "2026-09-20T12:00:00Z",
+            },
+            "breakdown": {
+                "by_service_type": {
+                    "4G_DATA": 10,
+                },
+                "by_radio_access_type": {
+                    "LTE_4G": 10,
+                },
+                "dlq_by_reason": {
+                    "invalid_rat_value": 2,
+                },
+            },
+        }
+
+    async def close(self) -> None:
+        pass
+
+
 @pytest.fixture
 def publisher() -> FakePublisher:
     return FakePublisher()
@@ -49,6 +78,7 @@ def client(
 ):
     app = create_app(
         publisher=publisher,
+        worker_stats_client=FakeWorkerStatsClient(),
     )
 
     with TestClient(app) as test_client:
@@ -226,3 +256,74 @@ def test_kafka_failure_returns_503(
     assert response.status_code == 503
 
     assert response.json()["error"] == "kafka_unavailable"
+
+
+def test_metrics_endpoint(
+    client: TestClient,
+) -> None:
+    response = client.get(
+        "/metrics"
+    )
+
+    assert response.status_code == 200
+
+    assert (
+        "cdr_api_received_total"
+        in response.text
+    )
+
+
+def test_stats_endpoint(
+    client: TestClient,
+) -> None:
+    response = client.get(
+        "/stats"
+    )
+
+    assert response.status_code == 200
+
+    body = response.json()
+
+    assert (
+        body["pipeline"]["total_processed"]
+        == 10
+    )
+
+    assert (
+        body["pipeline"]["total_dlq"]
+        == 2
+    )
+
+    assert (
+        body["pipeline"]["in_flight_buffer"]
+        == 3
+    )
+
+    assert (
+        body["breakdown"]
+        ["by_service_type"]
+        ["4G_DATA"]
+        == 10
+    )
+
+
+def test_successful_request_increments_api_metric(
+    client: TestClient,
+) -> None:
+    response = client.post(
+        "/api/v1/cdr/raw",
+        json=valid_payload(),
+    )
+
+    assert response.status_code == 202
+
+    metrics = client.get(
+        "/metrics"
+    )
+
+    assert (
+        'cdr_api_received_total 1.0'
+        in metrics.text
+    )
+
+

tests/unit/test_batch.py

@@ -0,0 +1,108 @@
+from datetime import datetime, timezone
+from ipaddress import IPv4Address
+from uuid import uuid4
+
+from core.domain.models import NormalizedCDR
+from core.worker.batch import (
+    BufferedCDR,
+    KafkaPosition,
+    MicroBatch,
+)
+
+
+def normalized_cdr(
+    raw_id: str,
+) -> NormalizedCDR:
+    now = datetime.now(
+        timezone.utc
+    )
+
+    return NormalizedCDR(
+        cdr_id=uuid4(),
+        raw_id=raw_id,
+
+        msisdn="+19875551234",
+        imsi="310410998877665",
+
+        service_type="4G_DATA",
+        radio_access_type="LTE_4G",
+        cell_id="4021_8812",
+
+        bytes_uploaded=100,
+        bytes_downloaded=200,
+        total_bytes=300,
+        duration_seconds=300,
+
+        session_start_time=now,
+        session_end_time=now,
+
+        source_ip=IPv4Address(
+            "10.240.12.1"
+        ),
+    )
+
+
+def buffered_item(
+    raw_id: str,
+    offset: int,
+) -> BufferedCDR:
+    return BufferedCDR(
+        cdr=normalized_cdr(raw_id),
+
+        original_payload={
+            "raw_id": raw_id,
+        },
+
+        position=KafkaPosition(
+            topic="cdr.raw",
+            partition=0,
+            offset=offset,
+        ),
+
+        kafka_timestamp_ms=None,
+    )
+
+
+def test_batch_flushes_when_full() -> None:
+    batch = MicroBatch(
+        max_size=2,
+        flush_interval_seconds=10,
+    )
+
+    batch.add(
+        buffered_item(
+            raw_id="raw-1",
+            offset=1,
+        )
+    )
+
+    assert batch.is_full is False
+
+    batch.add(
+        buffered_item(
+            raw_id="raw-2",
+            offset=2,
+        )
+    )
+
+    assert batch.is_full is True
+
+
+def test_batch_clear_resets_state() -> None:
+    batch = MicroBatch(
+        max_size=2,
+        flush_interval_seconds=10,
+    )
+
+    batch.add(
+        buffered_item(
+            raw_id="raw-1",
+            offset=1,
+        )
+    )
+
+    batch.clear()
+
+    assert batch.size == 0
+    assert batch.is_empty is True
+    assert batch.is_due() is False

tests/unit/test_worker_stats.py

@@ -0,0 +1,95 @@
+from datetime import datetime, timezone
+from ipaddress import IPv4Address
+from uuid import uuid4
+
+from core.domain.models import NormalizedCDR
+from core.worker.stats import WorkerRuntimeStats
+
+
+def normalized_cdr() -> NormalizedCDR:
+    now = datetime.now(
+        timezone.utc
+    )
+
+    return NormalizedCDR(
+        cdr_id=uuid4(),
+        raw_id="raw-stats-1",
+
+        msisdn="+19875551234",
+        imsi="310410998877665",
+
+        service_type="4G_DATA",
+        radio_access_type="LTE_4G",
+        cell_id="4021_8812",
+
+        bytes_uploaded=100,
+        bytes_downloaded=200,
+        total_bytes=300,
+        duration_seconds=300,
+
+        session_start_time=now,
+        session_end_time=now,
+
+        source_ip=IPv4Address(
+            "10.240.12.1"
+        ),
+    )
+
+
+def test_worker_stats() -> None:
+    stats = WorkerRuntimeStats()
+
+    stats.record_processed(
+        normalized_cdr()
+    )
+
+    stats.record_dlq(
+        "invalid_rat_value"
+    )
+
+    stats.set_buffer_size(4)
+
+    stats.record_flush()
+
+    snapshot = stats.snapshot()
+
+    assert (
+        snapshot["pipeline"]["total_processed"]
+        == 1
+    )
+
+    assert (
+        snapshot["pipeline"]["total_dlq"]
+        == 1
+    )
+
+    assert (
+        snapshot["pipeline"]["in_flight_buffer"]
+        == 4
+    )
+
+    assert (
+        snapshot["breakdown"]
+        ["by_service_type"]
+        ["4G_DATA"]
+        == 1
+    )
+
+    assert (
+        snapshot["breakdown"]
+        ["by_radio_access_type"]
+        ["LTE_4G"]
+        == 1
+    )
+
+    assert (
+        snapshot["breakdown"]
+        ["dlq_by_reason"]
+        ["invalid_rat_value"]
+        == 1
+    )
+
+    assert (
+        snapshot["pipeline"]["last_flush_at"]
+        is not None
+    )

tree

@@ -0,0 +1,772 @@
+# Telecom CDR Pipeline
+
+High-throughput, fault-tolerant telecom Call Detail Record ingestion and analytics pipeline built with FastAPI, Redpanda/Kafka, ClickHouse, and Python.
+
+## Architecture
+
+```text
+Network Element
+      │
+      ▼
+FastAPI Ingestion API
+      │
+      ├── Validation
+      │
+      ├── Bounded Async Queue
+      │
+      └── Kafka Producer
+      │
+      ▼
+Redpanda / cdr.raw
+      │
+      ▼
+Processing Worker
+      │
+      ├── Decode + Validate
+      ├── Normalize
+      ├── Micro-Batch
+      │
+      ├── valid ──────► ClickHouse
+      │
+      └── invalid ────► cdr.raw.dlq
+```
+
+The main services are:
+
+* **FastAPI API** — validates incoming CDRs, applies backpressure, and publishes accepted records to Kafka.
+* **Redpanda** — Kafka-compatible durable buffer between ingestion and processing.
+* **Processing Worker** — validates, normalizes, batches, and persists CDRs.
+* **ClickHouse** — analytical storage for normalized records.
+* **DLQ** — stores malformed or unprocessable records with structured failure information.
+
+## Features
+
+* Bounded asynchronous ingestion queue
+* `202`, `400`, `429`, and `503` ingestion responses
+* Long-lived asynchronous Kafka producer
+* Kafka delivery acknowledgement before `202 Accepted`
+* Kafka consumer with manual offset management
+* Worker-side validation and normalization
+* E.164 phone-number normalization using `phonenumbers`
+* RAT mapping for `3G`, `LTE_4G`, and `5G_NR`
+* Deterministic UUIDv5 `cdr_id` generation
+* Structured Dead Letter Queue
+* Configurable ClickHouse micro-batching
+* Flush on 1,000 records or 2-second timeout
+* ClickHouse retries with exponential backoff
+* At-least-once delivery semantics
+* Replay-safe ClickHouse storage using `ReplacingMergeTree`
+* Graceful worker shutdown with final buffer flush
+* Separate Prometheus metrics endpoints for API and worker
+* Kafka consumer lag monitoring
+* Aggregated `/stats` endpoint
+* Health-based Docker Compose startup ordering
+* Automated Kafka topic provisioning
+* Automated and idempotent ClickHouse initialization
+* Unit, integration, functional, and load testing
+
+---
+
+## Project Structure
+
+Generated directories such as `__pycache__` and `*.egg-info` are omitted below.
+
+```text
+telecom-cdr-pipeline/
+├── Dockerfile.api
+├── Dockerfile.worker
+├── README.md
+├── docker-compose.yml
+├── pyproject.toml
+│
+├── infrastructure/
+│   └── clickhouse/
+│       └── init.sql
+│
+├── scripts/
+│   ├── functional_traffic.py
+│   ├── load_test.js
+│   └── sample-valid-cdr.json
+│
+├── src/
+│   └── core/
+│       ├── __init__.py
+│       ├── config.py
+│       │
+│       ├── api/
+│       │   ├── __init__.py
+│       │   ├── kafka_publisher.py
+│       │   ├── main.py
+│       │   ├── metrics.py
+│       │   ├── publisher.py
+│       │   ├── routes.py
+│       │   └── stats.py
+│       │
+│       ├── domain/
+│       │   ├── __init__.py
+│       │   ├── errors.py
+│       │   ├── models.py
+│       │   ├── normalization.py
+│       │   └── validation.py
+│       │
+│       └── worker/
+│           ├── __init__.py
+│           ├── batch.py
+│           ├── clickhouse.py
+│           ├── consumer.py
+│           ├── dlq.py
+│           ├── main.py
+│           ├── metrics.py
+│           ├── observability.py
+│           ├── processor.py
+│           └── stats.py
+│
+└── tests/
+    ├── integration/
+    │   └── test_ingestion_api.py
+    │
+    └── unit/
+        ├── test_batch.py
+        ├── test_dlq.py
+        ├── test_normalization.py
+        ├── test_project_setup.py
+        ├── test_validation.py
+        ├── test_worker_processor.py
+        └── test_worker_stats.py
+```
+
+---
+
+## Quickstart
+
+### 1. Configure Environment
+
+Create your local environment file from `.env.example`:
+
+```bash
+cp .env.example .env
+```
+
+### 2. Start the Complete Pipeline
+
+```bash
+docker compose up --build -d
+```
+
+Check service status:
+
+```bash
+docker compose ps -a
+```
+
+Expected state:
+
+```text
+telecom-redpanda        healthy
+telecom-clickhouse      healthy
+telecom-api             healthy
+telecom-worker          healthy
+telecom-redpanda-init   exited (0)
+```
+
+### 3. Verify Health
+
+API:
+
+```bash
+curl http://localhost:8000/health
+```
+
+Worker:
+
+```bash
+curl http://localhost:8001/health
+```
+
+### 4. Send a Sample CDR
+
+```bash
+curl -X POST \
+  http://localhost:8000/api/v1/cdr/raw \
+  -H "Content-Type: application/json" \
+  --data-binary "@scripts/sample-valid-cdr.json"
+```
+
+Expected response:
+
+```json
+{
+  "status": "accepted",
+  "raw_id": "raw-test-001"
+}
+```
+
+---
+
+## API Endpoints
+
+### Ingestion API
+
+```text
+POST /api/v1/cdr/raw
+```
+
+Possible responses:
+
+| Status | Meaning                                     |
+| ------ | ------------------------------------------- |
+| `202`  | Payload validated and acknowledged by Kafka |
+| `400`  | Invalid JSON or invalid CDR schema          |
+| `429`  | Bounded ingestion queue is full             |
+| `503`  | Kafka delivery is unavailable               |
+
+### API Observability
+
+```text
+GET http://localhost:8000/health
+GET http://localhost:8000/metrics
+GET http://localhost:8000/stats
+```
+
+### Worker Observability
+
+```text
+GET http://localhost:8001/health
+GET http://localhost:8001/metrics
+GET http://localhost:8001/internal/stats
+```
+
+---
+
+## Micro-Batching
+
+The worker never inserts CDRs into ClickHouse row-by-row.
+
+Normalized records are buffered and flushed when either condition occurs:
+
+```text
+Batch reaches 1,000 records
+            OR
+2 seconds pass after the batch starts
+```
+
+Configuration:
+
+```env
+BATCH_SIZE=1000
+BATCH_FLUSH_INTERVAL_SECONDS=2
+```
+
+This reduces ClickHouse insert overhead while still keeping low-volume traffic latency bounded.
+
+On `SIGTERM`, the worker stops consuming new records, flushes the remaining buffer, commits the corresponding Kafka offsets, and then exits.
+
+---
+
+## Delivery and Offset Strategy
+
+The worker uses **at-least-once delivery**.
+
+Kafka auto-commit is disabled.
+
+The normal processing order is:
+
+```text
+Kafka consume
+      ↓
+Validate + Normalize
+      ↓
+Micro-Batch
+      ↓
+ClickHouse INSERT
+      ↓
+ClickHouse ACK
+      ↓
+Kafka offset commit
+```
+
+Offsets are never committed before the corresponding processing outcome is safely completed.
+
+### ClickHouse Failure
+
+A failed ClickHouse batch is retried with exponential backoff:
+
+```text
+Attempt 1
+   ↓ failure
+0.5s delay
+
+Attempt 2
+   ↓ failure
+1.0s delay
+
+Attempt 3
+   ↓ failure
+Entire batch → DLQ
+```
+
+Offsets are committed only after every DLQ publication succeeds.
+
+If DLQ publishing fails, the offsets remain uncommitted so Kafka can replay the records.
+
+### Worker Crash After ClickHouse Write
+
+If ClickHouse successfully stores a batch but the worker crashes before committing Kafka offsets, Kafka delivers the records again after restart.
+
+This can produce duplicate physical inserts, which is handled by the deduplication strategy below.
+
+### Partial Batch Failure
+
+A batch is treated as a single processing unit.
+
+Offsets are never partially committed for a failed ClickHouse batch.
+
+---
+
+## Deduplication
+
+Each CDR gets a deterministic UUIDv5 generated from its `raw_id`:
+
+```python
+uuid.uuid5(uuid.NAMESPACE_DNS, raw_id)
+```
+
+The same `raw_id` therefore always generates the same `cdr_id`.
+
+ClickHouse uses:
+
+```text
+ReplacingMergeTree(processed_at)
+```
+
+with `cdr_id` included in the sorting key.
+
+This allows records replayed because of at-least-once Kafka delivery to eventually collapse into one logical record.
+
+For immediate query-time deduplication:
+
+```sql
+SELECT *
+FROM telecom.cdr_records FINAL;
+```
+
+---
+
+## Dead Letter Queue
+
+Invalid or unprocessable Kafka records are published to:
+
+```text
+cdr.raw.dlq
+```
+
+DLQ records include:
+
+```json
+{
+  "original_payload": {},
+  "failure_reason": "invalid_rat_value",
+  "failed_field": "rat",
+  "failed_at": "2026-09-20T12:00:00Z",
+  "worker_instance": "worker-1"
+}
+```
+
+Supported failure reasons:
+
+```text
+missing_required_field
+invalid_rat_value
+invalid_epoch
+malformed_json
+normalization_error
+```
+
+A bad record does not crash the worker.
+
+---
+
+## Monitoring
+
+### API Metrics
+
+```bash
+curl http://localhost:8000/metrics
+```
+
+Includes:
+
+```text
+cdr_api_received_total
+```
+
+### Worker Metrics
+
+```bash
+curl http://localhost:8001/metrics
+```
+
+Includes:
+
+```text
+cdr_processed_total
+cdr_dlq_total
+
+cdr_kafka_to_clickhouse_latency_seconds
+cdr_batch_flush_duration_seconds
+cdr_batch_size_at_flush
+
+cdr_in_flight_buffer_size
+cdr_kafka_consumer_lag
+```
+
+### Pipeline Stats
+
+```bash
+curl http://localhost:8000/stats
+```
+
+Example:
+
+```json
+{
+  "pipeline": {
+    "total_received": 1000,
+    "total_processed": 995,
+    "total_dlq": 5,
+    "in_flight_buffer": 0,
+    "last_flush_at": "2026-09-20T12:00:00Z"
+  },
+  "breakdown": {
+    "by_service_type": {},
+    "by_radio_access_type": {},
+    "dlq_by_reason": {}
+  }
+}
+```
+
+---
+
+## ClickHouse Verification
+
+Open a ClickHouse client:
+
+```bash
+docker compose exec clickhouse \
+  clickhouse-client \
+  --user telecom \
+  --password telecom_dev_password \
+  --database telecom
+```
+
+Count records by service:
+
+```sql
+SELECT
+    service_type,
+    count() AS total
+FROM cdr_records
+GROUP BY service_type;
+```
+
+Average duration by RAT:
+
+```sql
+SELECT
+    radio_access_type,
+    avg(duration_seconds) AS avg_duration
+FROM cdr_records
+GROUP BY radio_access_type;
+```
+
+Top cells by traffic:
+
+```sql
+SELECT
+    cell_id,
+    sum(total_bytes) AS total_traffic
+FROM cdr_records
+GROUP BY cell_id
+ORDER BY total_traffic DESC
+LIMIT 10;
+```
+
+Check logical deduplication:
+
+```sql
+SELECT
+    raw_id,
+    count()
+FROM cdr_records FINAL
+GROUP BY raw_id
+HAVING count() > 1;
+```
+
+Expected result:
+
+```text
+0 rows
+```
+
+---
+
+## Automated Tests
+
+Install development dependencies:
+
+```bash
+pip install -e ".[dev]"
+```
+
+Run all unit and integration tests:
+
+```bash
+pytest -v
+```
+
+---
+
+## Functional Traffic Verification
+
+The functional traffic script exercises the pipeline with the assessment scenarios, including valid records, invalid requests, and duplicate `raw_id` values.
+
+Run:
+
+```bash
+python scripts/functional_traffic.py
+```
+
+The assessment scenarios are:
+
+| Scenario                  | Count | Expected                                      |
+| ------------------------- | ----: | --------------------------------------------- |
+| Valid records             | 1,000 | Stored in ClickHouse                          |
+| Missing `caller_num`      |    20 | Rejected / DLQ when worker is tested directly |
+| Invalid JSON              |    10 | API `400`; DLQ when worker is bypassed        |
+| Unknown `rat=99`          |    10 | Rejected / DLQ when worker is tested directly |
+| `epoch_start > epoch_end` |    10 | Rejected / DLQ when worker is tested directly |
+| Duplicate `raw_id`        |    20 | One logical ClickHouse record per `raw_id`    |
+
+Invalid records submitted through the public API are rejected before Kafka. Worker-side DLQ handling is independently exercised by publishing malformed or invalid records directly to `cdr.raw` and by the automated worker tests.
+
+### Functional Traffic Result
+
+Paste the final `functional_traffic.py` output below:
+
+```text
+Total requests: 1070
+Elapsed: 5.21s
+
+duplicate                status=202      count=20
+invalid_epoch            status=400      count=10
+invalid_json             status=400      count=10
+invalid_rat              status=400      count=10
+missing_caller_num       status=400      count=20
+valid                    status=202      count=1000
+
+Expected:
+  valid             → 1000 x 202
+  duplicate         →   20 x 202
+  missing_caller    →   20 x 400
+  invalid_json      →   10 x 400
+  invalid_rat       →   10 x 400
+  invalid_epoch     →   10 x 400
+```
+
+---
+
+## Required k6 Burst Test
+
+The assessment requires:
+
+```text
+500 records in less than 5 seconds
+0 × 429 responses
+No data loss
+```
+
+The included k6 script uses a configurable constant arrival rate.
+
+Run the required assessment scenario with:
+
+```bash
+docker run --rm -i \
+  --network=host \
+  -e RATE=125 \
+  -e DURATION=4s \
+  -e PRE_ALLOCATED_VUS=500 \
+  -e MAX_VUS=750 \
+  grafana/k6 run - \
+  < scripts/load_test.js
+```
+
+This schedules approximately:
+
+```text
+125 requests/second × 4 seconds = 500 requests
+```
+
+A successful result should show approximately:
+
+```text
+iterations             500
+http_reqs              500
+cdr_accepted           500
+cdr_backpressure       0
+cdr_unavailable        0
+http_req_failed        0%
+dropped_iterations     0
+```
+
+After the run, allow the worker to finish processing and verify that no records were lost.
+
+Check worker lag:
+
+```bash
+curl -s http://localhost:8001/metrics \
+  | grep cdr_kafka_consumer_lag
+```
+
+Check pipeline statistics:
+
+```bash
+curl http://localhost:8000/stats
+```
+
+### Required k6 Result
+
+Paste the final required 500-record burst result below:
+
+```text
+    TOTAL RESULTS 
+
+    checks_total.......: 4087   272.979956/s
+    checks_succeeded...: 34.47% 1409 out of 4087
+    checks_failed......: 65.52% 2678 out of 4087
+
+    ✗ accepted or gracefully rejected
+      ↳  34% — ✓ 1409 / ✗ 2678
+
+    CUSTOM
+    cdr_accepted...................: 1409   94.110291/s
+    cdr_unavailable................: 2678   178.869665/s
+
+    HTTP
+    http_req_duration..............: avg=5.26s min=44.55ms med=4.15s max=11.2s p(90)=10.42s p(95)=10.7s
+      { expected_response:true }...: avg=2.33s min=44.55ms med=1.75s max=9.94s p(90)=5.75s  p(95)=6.1s 
+    http_req_failed................: 65.52% 2678 out of 4087
+    http_reqs......................: 4087   272.979956/s
+
+    EXECUTION
+    dropped_iterations.............: 5912   394.875826/s
+    iteration_duration.............: avg=5.26s min=46.48ms med=4.15s max=11.2s p(90)=10.42s p(95)=10.7s
+    iterations.....................: 4087   272.979956/s
+    vus............................: 214    min=214          max=2000
+    vus_max........................: 2000   min=1500         max=2000
+
+    NETWORK
+    data_received..................: 862 kB 58 kB/s
+    data_sent......................: 1.6 MB 105 kB/s
+
+running (15.0s), 0000/2000 VUs, 4087 complete and 0 interrupted iterations
+ingestion ✓ [ 100% ] 0000/2000 VUs  10s  1000.00 iters/s
+```
+
+---
+
+## Optional Stress Test
+
+The required assessment test is the 500-record burst above.
+
+A larger stress test can optionally be executed with:
+
+```bash
+docker run --rm -i \
+  --network=host \
+  -e RATE=1000 \
+  -e DURATION=10s \
+  -e PRE_ALLOCATED_VUS=1500 \
+  -e MAX_VUS=2000 \
+  grafana/k6 run - \
+  < scripts/load_test.js
+```
+
+This is useful for identifying throughput and latency limits but is not required for the assessment's 500-record burst criterion.
+
+---
+
+## Configuration Reference
+
+Configuration is provided through environment variables. Local defaults are documented in `.env.example`.
+
+| Variable                                  | Default                                | Purpose                                |
+| ----------------------------------------- | -------------------------------------- | -------------------------------------- |
+| `REDPANDA_VERSION`                        | `v26.2.3`                              | Redpanda image version                 |
+| `KAFKA_BOOTSTRAP_SERVERS`                 | `localhost:19092`                      | Kafka address for host applications    |
+| `KAFKA_INTERNAL_BOOTSTRAP_SERVERS`        | `redpanda:9092`                        | Kafka address inside Docker            |
+| `KAFKA_RAW_TOPIC`                         | `cdr.raw`                              | Raw CDR topic                          |
+| `KAFKA_RAW_TOPIC_PARTITIONS`              | `3`                                    | Raw topic partitions                   |
+| `KAFKA_RAW_TOPIC_REPLICATION_FACTOR`      | `1`                                    | Raw topic replication                  |
+| `KAFKA_RAW_TOPIC_RETENTION_MS`            | `86400000`                             | Raw topic retention                    |
+| `KAFKA_RAW_TOPIC_COMPRESSION`             | `lz4`                                  | Raw topic compression                  |
+| `KAFKA_DLQ_TOPIC`                         | `cdr.raw.dlq`                          | Dead Letter Queue topic                |
+| `KAFKA_DLQ_TOPIC_PARTITIONS`              | `1`                                    | DLQ partitions                         |
+| `KAFKA_DLQ_TOPIC_REPLICATION_FACTOR`      | `1`                                    | DLQ replication                        |
+| `KAFKA_DLQ_TOPIC_RETENTION_MS`            | `604800000`                            | DLQ retention                          |
+| `KAFKA_CONSUMER_GROUP`                    | `cdr-processing-workers`               | Worker consumer group                  |
+| `KAFKA_AUTO_OFFSET_RESET`                 | `earliest`                             | Consumer fallback offset               |
+| `KAFKA_CONSUMER_POLL_TIMEOUT_SECONDS`     | `1`                                    | Worker Kafka poll timeout              |
+| `KAFKA_DLQ_DELIVERY_TIMEOUT_SECONDS`      | `5`                                    | DLQ delivery timeout                   |
+| `KAFKA_DELIVERY_TIMEOUT_SECONDS`          | `4`                                    | API Kafka acknowledgement timeout      |
+| `KAFKA_MESSAGE_TIMEOUT_MS`                | `3000`                                 | Kafka message timeout                  |
+| `KAFKA_LINGER_MS`                         | `5`                                    | Producer batching delay                |
+| `KAFKA_BATCH_SIZE`                        | `131072`                               | Kafka producer batch size              |
+| `KAFKA_PRODUCER_MAX_MESSAGES`             | `100000`                               | Kafka producer internal queue limit    |
+| `API_QUEUE_SIZE`                          | `10000`                                | Bounded API ingestion queue            |
+| `CLICKHOUSE_VERSION`                      | `26.8.7.19`                            | ClickHouse image version               |
+| `CLICKHOUSE_HOST`                         | `localhost`                            | ClickHouse host for local applications |
+| `CLICKHOUSE_INTERNAL_HOST`                | `clickhouse`                           | ClickHouse Docker service hostname     |
+| `CLICKHOUSE_PORT`                         | `8123`                                 | ClickHouse HTTP port                   |
+| `CLICKHOUSE_HTTP_HOST_PORT`               | `8123`                                 | Host-exposed ClickHouse port           |
+| `CLICKHOUSE_DATABASE`                     | `telecom`                              | ClickHouse database                    |
+| `CLICKHOUSE_USER`                         | `telecom`                              | ClickHouse user                        |
+| `CLICKHOUSE_PASSWORD`                     | `telecom_dev_password`                 | Local development password             |
+| `BATCH_SIZE`                              | `1000`                                 | Worker micro-batch maximum size        |
+| `BATCH_FLUSH_INTERVAL_SECONDS`            | `2`                                    | Worker time-based flush interval       |
+| `CLICKHOUSE_MAX_INSERT_ATTEMPTS`          | `3`                                    | Maximum ClickHouse insert attempts     |
+| `CLICKHOUSE_RETRY_BASE_DELAY_SECONDS`     | `0.5`                                  | Exponential retry base delay           |
+| `CLICKHOUSE_CONNECT_TIMEOUT_SECONDS`      | `5`                                    | ClickHouse connection timeout          |
+| `CLICKHOUSE_SEND_RECEIVE_TIMEOUT_SECONDS` | `10`                                   | ClickHouse request timeout             |
+| `WORKER_INSTANCE_ID`                      | `worker-1`                             | Worker instance identifier             |
+| `WORKER_HTTP_HOST`                        | `0.0.0.0`                              | Worker observability bind address      |
+| `WORKER_HTTP_PORT`                        | `8001`                                 | Worker observability port              |
+| `WORKER_STATS_URL`                        | `http://localhost:8001/internal/stats` | Worker stats URL for host-run API      |
+| `WORKER_STATS_TIMEOUT_SECONDS`            | `2`                                    | Worker stats HTTP timeout              |
+| `API_HOST_PORT`                           | `8000`                                 | API host port                          |
+| `WORKER_HTTP_HOST_PORT`                   | `8001`                                 | Worker observability host port         |
+
+Inside Docker Compose, service-to-service addresses override host defaults where required:
+
+```text
+Kafka:      redpanda:9092
+ClickHouse: clickhouse:8123
+Worker:     http://worker:8001
+```
+
+---
+
+## Shutdown
+
+Stop the stack while preserving Kafka and ClickHouse data:
+
+```bash
+docker compose down
+```
+
+Remove containers and persistent development data:
+
+```bash
+docker compose down -v
+```
+
+Use `-v` only when a completely clean local environment is desired.