feat: add worker retry and replay safety

.env.sample

@@ -72,6 +72,16 @@ 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
 # =========================

README.md

@@ -81,6 +81,11 @@ worker processing is not connected yet.
 * 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
 
 ## Project Structure
 
@@ -205,3 +210,11 @@ 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.

src/core/config.py

@@ -37,6 +37,12 @@ class Settings(BaseSettings):
     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
 

src/core/worker/batch.py

@@ -1,5 +1,6 @@
 import time
 from dataclasses import dataclass
+from typing import Any
 
 from core.domain.models import NormalizedCDR
 
@@ -14,6 +15,7 @@ class KafkaPosition:
 @dataclass(frozen=True, slots=True)
 class BufferedCDR:
     cdr: NormalizedCDR
+    original_payload: dict[str, Any]
     position: KafkaPosition
 
 

src/core/worker/clickhouse.py

@@ -1,4 +1,6 @@
 from collections.abc import Sequence
+import logging
+import time
 
 import clickhouse_connect
 
@@ -6,6 +8,9 @@ from core.config import Settings
 from core.domain.models import NormalizedCDR
 
 
+logger = logging.getLogger(__name__)
+
+
 class ClickHouseWriteError(Exception):
     pass
 
@@ -40,6 +45,13 @@ class ClickHouseWriter:
             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
@@ -68,6 +80,22 @@ class ClickHouseWriter:
             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],
@@ -75,23 +103,65 @@ class ClickHouseWriter:
         if not records:
             return
 
-        rows = [
-            self._to_row(record)
-            for record in records
-        ]
+        last_error: Exception | None = None
 
-        try:
-            self._client.insert(
-                table="cdr_records",
-                data=rows,
-                column_names=self.COLUMN_NAMES,
-                database=self._settings.clickhouse_database,
+        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."
             )
-
-        except Exception as exc:
-            raise ClickHouseWriteError(
-                f"ClickHouse batch insert failed: {exc}"
-            ) from exc
+        ) from last_error
 
     def close(self) -> None:
         self._client.close()

src/core/worker/consumer.py

@@ -13,7 +13,10 @@ from core.worker.batch import (
     KafkaPosition,
     MicroBatch,
 )
-from core.worker.clickhouse import ClickHouseWriter
+from core.worker.clickhouse import (
+    ClickHouseWriteError,
+    ClickHouseWriter,
+)
 from core.config import Settings
 from core.domain.errors import (
     CDRProcessingError,
@@ -171,6 +174,7 @@ class ProcessingWorker:
         self._batch.add(
             BufferedCDR(
                 cdr=normalized,
+                original_payload=payload,
                 position=position,
             )
         )
@@ -257,19 +261,47 @@ class ProcessingWorker:
             for item in items
         ]
 
+        positions = [
+            item.position
+            for item in items
+        ]
+
         logger.info(
             "Flushing ClickHouse batch. size=%s",
             len(records),
         )
 
-        self._clickhouse.insert(
-            records
-        )
+        try:
+            self._clickhouse.insert(
+                records
+            )
 
-        positions = [
-            item.position
-            for item in items
-        ]
+        except ClickHouseWriteError as exc:
+            logger.exception(
+                "ClickHouse batch permanently failed."
+            )
+
+            self._route_batch_to_dlq(
+                items=items,
+                storage_error=exc,
+            )
+
+            # Only reached if EVERY DLQ publish succeeded.
+            self._commit_positions(
+                positions
+            )
+
+            self._batch.clear()
+
+            logger.warning(
+                (
+                    "Failed ClickHouse batch routed to DLQ "
+                    "and Kafka offsets committed. size=%s"
+                ),
+                len(records),
+            )
+
+            return
 
         self._commit_positions(
             positions
@@ -332,6 +364,33 @@ class ProcessingWorker:
                     partition.error
                 )
 
+    def _route_batch_to_dlq(
+        self,
+        items: list[BufferedCDR],
+        storage_error: ClickHouseWriteError,
+    ) -> None:
+        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:
+            self._dlq.publish(
+                original_payload=item.original_payload,
+                error=error,
+            )
+
     def stop(self) -> None:
         self._running = False
 

tests/unit/test_batch.py

@@ -48,6 +48,11 @@ def buffered_item(
 ) -> BufferedCDR:
     return BufferedCDR(
         cdr=normalized_cdr(raw_id),
+
+        original_payload={
+            "raw_id": raw_id,
+        },
+
         position=KafkaPosition(
             topic="cdr.raw",
             partition=0,