fix(worker): commit confirmed partition offsets and drain safely

app/worker/clickhouse.py

@@ -141,7 +141,8 @@ class ClickHouseService(BaseClickHouseClient):
                     f"Retrying in {delay:.2f}s...",
                     extra={"event": "clickhouse_insert_retry"},
                 )
-                time.sleep(delay)
+                if attempt < self.max_attempts:
+                    time.sleep(delay)
 
         logger.error(
             f"All {self.max_attempts} ClickHouse insert attempts failed: {last_exception}",

app/worker/consumer.py

@@ -6,6 +6,7 @@ from typing import Any, List, Optional, Tuple
 from app.api.producer import ConfluentKafkaProducer, MessageProducer
 from app.common.config import config
 from app.common.logging import setup_logger
+from app.common.validation import PayloadError, parse_payload
 from app.worker.clickhouse import BaseClickHouseClient, ClickHouseService
 from app.worker.dlq import route_to_dlq
 from app.worker.internal_server import WorkerInternalServer
@@ -47,8 +48,12 @@ class ProcessingWorker:
 
         # Micro-batch buffer: list of (NormalizedRecord, message)
         self.buffer: List[Tuple[NormalizedRecord, Any]] = []
-        self.last_flush_time = time.time()
+        self.last_flush_time = time.monotonic()
         self.running = False
+        self.failed = False
+        self._last_lag_time = 0.0
+        self._shutdown_started = False
+        self._deadline_alarm_installed = False
 
     def _init_consumer(self) -> None:
         if self._consumer is None:
@@ -59,7 +64,8 @@ class ProcessingWorker:
                     "bootstrap.servers": config.kafka.bootstrap_servers,
                     "group.id": config.kafka.consumer_group,
                     "auto.offset.reset": config.kafka.auto_offset_reset,
-                    "enable.auto.commit": False,  # Manual offset commit post ClickHouse ACK
+                    "enable.auto.commit": False,
+                    "enable.auto.offset.store": False,
                 }
                 self._consumer = confluent_kafka.Consumer(conf)
             except ImportError:
@@ -74,120 +80,103 @@ class ProcessingWorker:
             extra={"event": "shutdown_signal_received"},
         )
         self.running = False
+        if not self._shutdown_started:
+            self._shutdown_started = True
+            if self._deadline_alarm_installed:
+                signal.setitimer(signal.ITIMER_REAL, config.api.shutdown_timeout_seconds)
 
-    def flush_buffer(self) -> None:
-        """
-        Flushes buffered normalized records to ClickHouse and commits Kafka offsets.
-        If ClickHouse write fails all retries, routes batch to DLQ.
-        """
-        if not self.buffer:
-            self.last_flush_time = time.time()
-            return
+    def _shutdown_timeout(self, signum, frame):
+        # SystemExit bypasses retry/fallback Exception handlers. Uncommitted
+        # source offsets remain replayable instead of extending shutdown forever.
+        self.failed = True
+        self.running = False
+        logger.error("Shutdown deadline exceeded; uncommitted records will replay",
+                     extra={"event": "shutdown_timeout"})
+        raise SystemExit(1)
 
-        batch_records = [item[0] for item in self.buffer]
-        batch_messages = [item[1] for item in self.buffer]
-        batch_count = len(batch_records)
-
-        start_time = time.time()
-        try:
-            self.clickhouse.insert_batch(batch_records)
-            flush_duration = time.time() - start_time
-
-            # Commit offsets only after ClickHouse ACK
-            if self._consumer and hasattr(self._consumer, "commit"):
-                try:
-                    self._consumer.commit(asynchronous=False)
-                except Exception as e:
-                    logger.error(
-                        f"Kafka offset commit failed after ClickHouse ACK: {e}",
-                        extra={"event": "kafka_commit_failed"},
-                    )
-
-            # Record metrics
-            now_iso = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
-            worker_metrics.set_last_flush_at(now_iso)
-            worker_metrics.batch_flush_duration.observe(flush_duration)
-            worker_metrics.batch_size_at_flush.observe(batch_count)
-
-            for rec in batch_records:
-                worker_metrics.record_processed(
-                    rec.service_type, rec.radio_access_type, 1
-                )
-
-            logger.info(
-                f"Successfully flushed {batch_count} records to ClickHouse in {flush_duration:.3f}s",
-                extra={"event": "clickhouse_batch_flushed"},
-            )
-
-        except Exception as e:
-            logger.error(
-                f"Batch insert to ClickHouse failed after retries: {e}. Routing batch to DLQ.",
-                extra={"event": "ch_batch_fatal_error"},
-            )
-            # Route all records in batch to DLQ to prevent pipeline deadlock
-            for rec, msg in self.buffer:
-                route_to_dlq(
-                    producer=self.dlq_producer,
-                    topic=self.dlq_topic,
-                    original_payload=vars(rec),
-                    failure_reason="ch_insert_failed",
-                    failed_field=None,
-                    raw_id=rec.raw_id,
-                )
-
-            # Commit offsets to prevent infinite replay stall
-            if self._consumer and hasattr(self._consumer, "commit"):
-                try:
-                    self._consumer.commit(asynchronous=False)
-                except Exception as ce:
-                    logger.error(f"Commit after DLQ fallback failed: {ce}")
-
-        finally:
-            self.buffer.clear()
-            worker_metrics.set_in_flight_buffer(0)
-            self.last_flush_time = time.time()
-
-    def process_single_message(self, message: Any) -> None:
-        """
-        Parses, validates, and either buffers or routes message to DLQ.
-        """
-        value_bytes = message.value() if hasattr(message, "value") else message
-        if value_bytes is None:
-            return
-
-        record, failure_reason, failed_field, raw_id = normalize_cdr_payload(
-            value_bytes
+    def _commit(self, messages):
+        from confluent_kafka import TopicPartition
+        offsets = {}
+        for message in messages:
+            topic = message.topic() if hasattr(message, "topic") else config.kafka.raw_topic
+            key = (topic, message.partition())
+            offsets[key] = max(offsets.get(key, 0), message.offset() + 1)
+        result = self._consumer.commit(
+            offsets=[TopicPartition(topic, partition, offset) for (topic, partition), offset in offsets.items()],
+            asynchronous=False,
         )
+        for partition in result or []:
+            if partition.error:
+                raise RuntimeError(f"Partition commit failed: {partition.error}")
 
-        if failure_reason:
-            payload_str = (
-                value_bytes.decode("utf-8", errors="replace")
-                if isinstance(value_bytes, bytes)
-                else str(value_bytes)
-            )
-            route_to_dlq(
-                producer=self.dlq_producer,
-                topic=self.dlq_topic,
-                original_payload=payload_str,
-                failure_reason=failure_reason,
-                failed_field=failed_field,
-                raw_id=raw_id,
-            )
-            # Commit DLQ record offset so it is not reprocessed
-            if self._consumer and hasattr(self._consumer, "commit"):
-                try:
-                    self._consumer.commit(message=message, asynchronous=False)
-                except Exception:
-                    pass
+    @staticmethod
+    def _original(message):
+        data = message.value()
+        try:
+            return parse_payload(data)
+        except (PayloadError, TypeError, AttributeError):
+            return data.decode("utf-8", errors="replace") if isinstance(data, bytes) else str(data)
+
+    def flush_buffer(self):
+        if not self.buffer:
+            self.last_flush_time = time.monotonic()
             return
+        records = [record for record, _ in self.buffer]
+        started = time.monotonic()
+        try:
+            try:
+                self.clickhouse.insert_batch(records)
+            except Exception as exc:
+                logger.error(str(exc), extra={"event": "ch_batch_fatal_error"})
+                for record, message in self.buffer:
+                    route_to_dlq(self.dlq_producer, self.dlq_topic, self._original(message),
+                                 "ch_insert_failed", raw_id=record.raw_id)
+            else:
+                worker_metrics.batch_flush_duration.observe(time.monotonic() - started)
+                worker_metrics.batch_size_at_flush.observe(len(records))
+                worker_metrics.set_last_flush_at(datetime.now(timezone.utc).isoformat())
+                for record, message in self.buffer:
+                    worker_metrics.record_processed(record.service_type, record.radio_access_type)
+                    if hasattr(message, "timestamp"):
+                        _, timestamp_ms = message.timestamp()
+                        if timestamp_ms is not None and timestamp_ms >= 0:
+                            worker_metrics.kafka_to_ch_latency.observe(max(0, time.time() - timestamp_ms / 1000))
+            self._commit([message for _, message in self.buffer])
+        except Exception:
+            self.failed = True
+            self.running = False
+            raise
+        self.buffer.clear()
+        worker_metrics.set_in_flight_buffer(0)
+        self.last_flush_time = time.monotonic()
 
-        if record:
-            self.buffer.append((record, message))
-            worker_metrics.set_in_flight_buffer(len(self.buffer))
-
-            # Flush if batch size threshold reached
-            if len(self.buffer) >= self.batch_size:
+    def process_single_message(self, message):
+        try:
+            record, reason, field, raw_id = normalize_cdr_payload(message.value() or b"")
+            if reason:
                 self.flush_buffer()
+                route_to_dlq(self.dlq_producer, self.dlq_topic, self._original(message),
+                             reason, field, raw_id)
+                self._commit([message])
+            else:
+                self.buffer.append((record, message))
+                worker_metrics.set_in_flight_buffer(len(self.buffer))
+                if len(self.buffer) >= self.batch_size:
+                    self.flush_buffer()
+        except Exception:
+            self.failed = True
+            self.running = False
+            raise
+
+    def update_consumer_lag(self):
+        partitions = self._consumer.assignment()
+        if not partitions:
+            return
+        committed = self._consumer.committed(partitions, timeout=1)
+        for partition in committed:
+            low, high = self._consumer.get_watermark_offsets(partition, timeout=1)
+            offset = partition.offset if partition.offset >= 0 else low
+            worker_metrics.set_consumer_lag(partition.partition, max(0, high - offset))
 
     def start(self) -> None:
         """
@@ -199,6 +188,9 @@ class ProcessingWorker:
         # Register signals for graceful shutdown
         signal.signal(signal.SIGTERM, self._handle_signal)
         signal.signal(signal.SIGINT, self._handle_signal)
+        if hasattr(signal, "setitimer"):
+            signal.signal(signal.SIGALRM, self._shutdown_timeout)
+            self._deadline_alarm_installed = True
 
         self._consumer.subscribe([config.kafka.raw_topic])
         self.running = True
@@ -210,6 +202,12 @@ class ProcessingWorker:
 
         try:
             while self.running:
+                if time.monotonic() - self._last_lag_time >= 5:
+                    try:
+                        self.update_consumer_lag()
+                    except Exception as exc:
+                        logger.warning(str(exc), extra={"event": "consumer_lag_unavailable"})
+                    self._last_lag_time = time.monotonic()
                 # Poll with short 100ms timeout
                 msg = self._consumer.poll(timeout=0.1)
 
@@ -217,7 +215,7 @@ class ProcessingWorker:
                     # Check flush timer on idle / interval
                     if (
                         self.buffer
-                        and (time.time() - self.last_flush_time) >= self.flush_interval
+                        and (time.monotonic() - self.last_flush_time) >= self.flush_interval
                     ):
                         self.flush_buffer()
                     continue
@@ -234,10 +232,13 @@ class ProcessingWorker:
                 # Check flush interval during message stream
                 if (
                     self.buffer
-                    and (time.time() - self.last_flush_time) >= self.flush_interval
+                    and (time.monotonic() - self.last_flush_time) >= self.flush_interval
                 ):
                     self.flush_buffer()
 
+        except Exception:
+            self.failed = True
+            raise
         finally:
             self.stop()
 
@@ -249,18 +250,23 @@ class ProcessingWorker:
             f"Stopping worker: flushing {len(self.buffer)} remaining in-memory records...",
             extra={"event": "graceful_shutdown_flushing"},
         )
-        if self.buffer:
-            self.flush_buffer()
-
-        if self._consumer and hasattr(self._consumer, "close"):
-            self._consumer.close()
-
-        self.dlq_producer.close()
-        self.clickhouse.close()
-        self.internal_server.stop()
-        logger.info("Processing Worker stopped cleanly", extra={"event": "worker_stopped"})
+        try:
+            if self.buffer and not self.failed:
+                self.flush_buffer()
+        finally:
+            try:
+                if self._consumer:
+                    self._consumer.close()
+            finally:
+                try:
+                    self.dlq_producer.close()
+                finally:
+                    self.clickhouse.close()
+                    self.internal_server.stop()
+                    if self._deadline_alarm_installed:
+                        signal.setitimer(signal.ITIMER_REAL, 0)
+        logger.info("Worker stopped", extra={"event": "worker_stopped"})
 
 
 if __name__ == "__main__":
-    worker = ProcessingWorker()
-    worker.start()
+    ProcessingWorker().start()

tests/test_delivery_regressions.py

@@ -0,0 +1,193 @@
+"""Durability regressions using broker acknowledgement and partition-aware doubles."""
+import json
+import time
+import unittest
+from concurrent.futures import Future
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+from app.worker.consumer import ProcessingWorker
+from app.worker.metrics import WorkerMetrics
+from app.worker.normalizer import normalize_cdr_payload
+
+
+def payload(**changes):
+    data = dict(raw_id="regression-1", device_id="310410998877665",
+                caller_num="0019875551234", svc="4G_DATA", rat=6,
+                up_bytes=100, down_bytes=200, lac="4021", ci="8812",
+                epoch_start=1726070000, epoch_end=1726070300, src_ip="10.0.0.1")
+    data.update(changes)
+    return data
+
+
+class Message:
+    def __init__(self, data, offset=0, partition=0):
+        self.data = data if isinstance(data, bytes) else json.dumps(data).encode()
+        self.position = offset
+        self.partition_id = partition
+
+    def value(self): return self.data
+    def topic(self): return "cdr.raw"
+    def partition(self): return self.partition_id
+    def offset(self): return self.position
+    def timestamp(self): return (1, int((time.time() - 2) * 1000))
+    def error(self): return None
+
+
+class Consumer:
+    def __init__(self, events, error=None, partition_error=False):
+        self.events = events
+        self.error = error
+        self.partition_error = partition_error
+        self.commits = []
+
+    def commit(self, offsets=None, message=None, asynchronous=False):
+        if self.error:
+            raise self.error
+        if offsets is not None:
+            positions = {(item.topic, item.partition): item.offset for item in offsets}
+        elif message is not None:
+            positions = {(message.topic(), message.partition()): message.offset() + 1}
+        else:
+            positions = "implicit-current-position"
+        self.events.append(("commit", positions))
+        self.commits.append(positions)
+        return [SimpleNamespace(error=RuntimeError("partition commit failed"))] if self.partition_error else []
+
+    def close(self): pass
+
+
+class Producer:
+    def __init__(self, events, failures=()):
+        self.events = events
+        self.failures = set(failures)
+        self.messages = []
+
+    def produce(self, topic, key, value):
+        self.messages.append(json.loads(value))
+        result = Future()
+        if len(self.messages) in self.failures:
+            result.set_exception(ConnectionError("broker rejected DLQ record"))
+        else:
+            self.events.append(("dlq_ack", key))
+            result.set_result(None)
+        return result
+
+    def close(self): pass
+    def flush(self, timeout=5): return 0
+
+
+class DeliverySafetyTests(unittest.TestCase):
+    def setUp(self):
+        self.events = []
+        self.consumer = Consumer(self.events)
+        self.producer = Producer(self.events)
+        self.ch = Mock()
+        self.ch.insert_batch.side_effect = lambda records: self.events.append(("insert", len(records))) or True
+        self.metrics = WorkerMetrics()
+        self.patches = [patch("app.worker.consumer.worker_metrics", self.metrics),
+                        patch("app.worker.dlq.worker_metrics", self.metrics)]
+        for item in self.patches:
+            item.start()
+            self.addCleanup(item.stop)
+        self.worker = ProcessingWorker(consumer=self.consumer, dlq_producer=self.producer,
+                                       clickhouse=self.ch, batch_size=100, internal_server=Mock())
+
+    def test_invalid_record_cannot_commit_past_buffered_valid_record(self):
+        self.worker.process_single_message(Message(payload(), offset=0))
+        self.worker.process_single_message(Message(b"{malformed", offset=1))
+        self.assertEqual(self.events[0], ("insert", 1))
+        self.assertEqual(self.consumer.commits[-1], {("cdr.raw", 0): 2})
+        self.assertEqual(len(self.worker.buffer), 0)
+
+    def test_batch_commits_only_its_explicit_partition_offsets(self):
+        self.worker.process_single_message(Message(payload(), offset=4, partition=0))
+        self.worker.process_single_message(Message(payload(raw_id="second"), offset=9, partition=1))
+        self.worker.flush_buffer()
+        self.assertEqual(self.consumer.commits, [{("cdr.raw", 0): 5, ("cdr.raw", 1): 10}])
+
+    def test_failed_dlq_ack_never_commits_or_counts_success(self):
+        self.producer.failures = {1}
+        with self.assertRaises(Exception):
+            self.worker.process_single_message(Message(b"{malformed"))
+        self.assertEqual(self.consumer.commits, [])
+        self.assertEqual(self.metrics.get_internal_stats()["pipeline"]["total_dlq"], 0)
+
+    def test_partial_batch_dlq_failure_keeps_all_offsets_uncommitted(self):
+        self.ch.insert_batch.side_effect = ConnectionError("ClickHouse unavailable")
+        self.producer.failures = {2}
+        self.worker.process_single_message(Message(payload(), offset=0))
+        self.worker.process_single_message(Message(payload(raw_id="second"), offset=1))
+        with self.assertRaises(Exception):
+            self.worker.flush_buffer()
+        self.assertEqual(self.consumer.commits, [])
+        self.assertEqual(self.metrics.get_internal_stats()["pipeline"]["total_dlq"], 1)
+
+    def test_commit_exception_propagates_without_routing_inserted_batch_to_dlq(self):
+        self.consumer.error = ConnectionError("commit unavailable")
+        self.worker.process_single_message(Message(payload()))
+        with self.assertRaises(Exception):
+            self.worker.flush_buffer()
+        self.ch.insert_batch.assert_called_once()
+        self.assertEqual(self.producer.messages, [])
+
+    def test_partition_commit_error_is_not_success(self):
+        self.consumer.partition_error = True
+        self.worker.process_single_message(Message(payload()))
+        with self.assertRaises(Exception):
+            self.worker.flush_buffer()
+        self.assertEqual(self.producer.messages, [])
+
+    def test_clickhouse_failure_dlq_retains_original_vendor_payload(self):
+        self.ch.insert_batch.side_effect = ConnectionError("ClickHouse unavailable")
+        original = payload(vendor_extension={"keep": True})
+        self.worker.process_single_message(Message(original))
+        self.worker.flush_buffer()
+        received = self.producer.messages[0]["original_payload"]
+        self.assertEqual(json.loads(received) if isinstance(received, str) else received, original)
+
+    def test_successful_insert_populates_kafka_latency(self):
+        self.worker.process_single_message(Message(payload()))
+        self.worker.flush_buffer()
+        self.assertIn("cdr_kafka_to_clickhouse_latency_seconds_count 1", self.metrics.get_prometheus_metrics())
+
+
+class InputBoundaryTests(unittest.TestCase):
+    def test_invalid_values_are_rejected_without_raising(self):
+        cases = [("epoch_start", 10**100), ("epoch_end", 10**100), ("epoch_end", 2**32),
+                 ("epoch_start", -1), ("epoch_start", 1726070000.5),
+                 ("rat", 6.9), ("rat", True), ("up_bytes", -0.5),
+                 ("up_bytes", True), ("up_bytes", 2**64),
+                 ("up_bytes", 2**64 - 1), ("src_ip", "999.1.2.3"),
+                 ("src_ip", "::1"), ("raw_id", 123)]
+        for field, value in cases:
+            with self.subTest(field=field, value=value):
+                record, reason, failed_field, _ = normalize_cdr_payload(json.dumps(payload(**{field: value})).encode())
+                self.assertIsNone(record)
+                self.assertIsNotNone(reason)
+                self.assertIsNotNone(failed_field)
+
+    def test_invalid_utf8_and_nonfinite_json_are_rejected(self):
+        for value in (b"\xff", json.dumps(payload(up_bytes=float("nan"))).encode()):
+            with self.subTest(value=value):
+                record, reason, _, _ = normalize_cdr_payload(value)
+                self.assertIsNone(record)
+                self.assertIsNotNone(reason)
+
+    def test_clickhouse_rows_use_utc_and_native_ipv4(self):
+        from datetime import timedelta
+        from ipaddress import IPv4Address
+        record, reason, _, _ = normalize_cdr_payload(json.dumps(payload()).encode())
+        self.assertIsNone(reason)
+        row = record.to_row()
+        self.assertEqual(row[11].utcoffset(), timedelta(0))
+        self.assertIsInstance(row[13], IPv4Address)
+
+    def test_prometheus_labels_escape_quotes_backslashes_and_newlines(self):
+        metrics = WorkerMetrics()
+        metrics.record_processed('svc"\\\nname', "LTE_4G")
+        self.assertIn('service_type="svc\\"\\\\\\nname"', metrics.get_prometheus_metrics())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_worker.py

@@ -43,7 +43,7 @@ class MockKafkaConsumer:
         self.committed = False
         self.closed = False
 
-    def commit(self, message=None, asynchronous=False):
+    def commit(self, message=None, offsets=None, asynchronous=False):
         self.committed = True
 
     def close(self):

tests/test_worker_lifecycle.py

@@ -0,0 +1,158 @@
+"""Worker lifecycle and retry behavior without broker or database processes."""
+import json
+import signal
+import unittest
+from types import SimpleNamespace
+from unittest.mock import Mock, call, patch
+
+from app.api.producer import MockMessageProducer
+from app.worker.clickhouse import ClickHouseService
+from app.worker.consumer import ProcessingWorker
+from app.worker.metrics import WorkerMetrics
+from app.worker.normalizer import normalize_cdr_payload
+from tests.test_delivery_regressions import Message, payload
+
+
+class WorkerLifecycleTests(unittest.TestCase):
+    def setUp(self):
+        self.consumer = Mock()
+        self.consumer.commit.return_value = []
+        self.consumer.assignment.return_value = []
+        self.producer = MockMessageProducer()
+        self.store = Mock()
+        self.server = Mock()
+        self.metrics = WorkerMetrics()
+        metric_patch = patch("app.worker.consumer.worker_metrics", self.metrics)
+        metric_patch.start()
+        self.addCleanup(metric_patch.stop)
+        self.worker = ProcessingWorker(consumer=self.consumer, dlq_producer=self.producer,
+                                       clickhouse=self.store, batch_size=2,
+                                       flush_interval=1, internal_server=self.server)
+
+    def test_count_trigger_flushes_exact_batch_before_shutdown(self):
+        self.worker.process_single_message(Message(payload(), offset=0))
+        self.store.insert_batch.assert_not_called()
+        self.worker.process_single_message(Message(payload(raw_id="second"), offset=1))
+        self.assertEqual(len(self.store.insert_batch.call_args.args[0]), 2)
+        self.assertEqual(self.worker.buffer, [])
+        self.assertEqual(self.consumer.commit.call_args.kwargs["offsets"][0].offset, 2)
+
+    def test_idle_poll_flushes_on_timer_without_waiting_for_shutdown(self):
+        now = [0.0]
+        self.worker.last_flush_time = 0.0
+        polled = [0]
+
+        def poll(timeout):
+            polled[0] += 1
+            if polled[0] == 1:
+                return Message(payload())
+            if polled[0] == 2:
+                now[0] = 1.1
+                return None
+            self.store.insert_batch.assert_called_once()
+            self.assertEqual(self.worker.buffer, [])
+            self.worker._handle_signal(signal.SIGTERM, None)
+            return None
+
+        self.consumer.poll.side_effect = poll
+        with patch("app.worker.consumer.time.monotonic", side_effect=lambda: now[0]), \
+                patch("app.worker.consumer.signal.signal"):
+            self.worker.start()
+        self.store.insert_batch.assert_called_once()
+        self.consumer.close.assert_called_once()
+        self.server.stop.assert_called_once()
+
+    def test_graceful_shutdown_flushes_partial_batch(self):
+        self.worker.process_single_message(Message(payload(), offset=3))
+        self.worker.stop()
+        self.store.insert_batch.assert_called_once()
+        self.assertEqual(self.consumer.commit.call_args.kwargs["offsets"][0].offset, 4)
+        self.consumer.close.assert_called_once()
+        self.store.close.assert_called_once()
+        self.server.stop.assert_called_once()
+
+    def test_failed_commit_cleanup_does_not_retry_write_or_commit(self):
+        self.consumer.commit.side_effect = ConnectionError("commit failed")
+        self.worker.process_single_message(Message(payload()))
+        with self.assertRaises(ConnectionError):
+            self.worker.flush_buffer()
+        self.worker.stop()
+        self.assertTrue(self.worker.failed)
+        self.store.insert_batch.assert_called_once()
+        self.consumer.commit.assert_called_once()
+        self.assertEqual(self.producer.messages, [])
+        self.consumer.close.assert_called_once()
+        self.store.close.assert_called_once()
+        self.server.stop.assert_called_once()
+
+    def test_lag_uses_committed_offsets_and_log_start_for_uncommitted_partition(self):
+        partitions = [SimpleNamespace(partition=0, offset=7),
+                      SimpleNamespace(partition=1, offset=-1001)]
+        self.consumer.assignment.return_value = partitions
+        self.consumer.committed.return_value = partitions
+        self.consumer.get_watermark_offsets.side_effect = [(0, 12), (5, 20)]
+        self.worker.update_consumer_lag()
+        output = self.metrics.get_prometheus_metrics()
+        self.assertIn('cdr_kafka_consumer_lag{partition="0"} 5', output)
+        self.assertIn('cdr_kafka_consumer_lag{partition="1"} 15', output)
+
+    def test_replay_after_write_before_commit_preserves_logical_identity(self):
+        # Models FINAL replacement by identity; real sorting-key verification is integration coverage.
+        logical_rows = {}
+        writes = []
+
+        def insert(records):
+            writes.extend(records)
+            logical_rows.update((record.cdr_id, record) for record in records)
+
+        self.store.insert_batch.side_effect = insert
+        self.consumer.commit.side_effect = ConnectionError("crash before commit")
+        source = Message(payload(), offset=8)
+        self.worker.process_single_message(source)
+        with self.assertRaises(ConnectionError):
+            self.worker.flush_buffer()
+        replay_consumer = Mock()
+        replay_consumer.commit.return_value = []
+        restarted = ProcessingWorker(consumer=replay_consumer, dlq_producer=self.producer,
+                                     clickhouse=self.store, batch_size=1, internal_server=Mock())
+        restarted.process_single_message(source)
+        self.assertEqual(len(writes), 2)
+        self.assertEqual(len(logical_rows), 1)
+        self.assertEqual(replay_consumer.commit.call_args.kwargs["offsets"][0].offset, 9)
+        self.assertEqual(self.producer.messages, [])
+
+
+class ClickHouseRetryTests(unittest.TestCase):
+    def make_service(self):
+        service = ClickHouseService.__new__(ClickHouseService)
+        service._client = Mock()
+        service.table = "cdr_records"
+        service.max_attempts = 3
+        service.initial_delay_ms = 100
+        service.backoff_factor = 2
+        record, reason, _, _ = normalize_cdr_payload(json.dumps(payload()).encode())
+        self.assertIsNone(reason)
+        return service, record
+
+    def test_transient_failures_retry_with_exponential_delays(self):
+        service, record = self.make_service()
+        service._client.insert.side_effect = [TimeoutError("first"), TimeoutError("second"), None]
+        with patch("app.worker.clickhouse.time.sleep") as sleep:
+            self.assertTrue(service.insert_batch([record]))
+        self.assertEqual(service._client.insert.call_count, 3)
+        self.assertEqual(sleep.call_args_list, [call(0.1), call(0.2)])
+
+    def test_exhausted_retries_raise_last_failure_without_final_sleep(self):
+        service, record = self.make_service()
+        failure = TimeoutError("persistent failure")
+        service._client.insert.side_effect = failure
+        with patch("app.worker.clickhouse.time.sleep") as sleep:
+            with self.assertRaises(TimeoutError) as caught:
+                service.insert_batch([record])
+        self.assertIs(caught.exception, failure)
+        self.assertEqual(service._client.insert.call_count, 3)
+        self.assertEqual(sleep.call_args_list, [call(0.1), call(0.2)])
+
+
+if __name__ == "__main__":
+    unittest.main()