diff --git a/app/worker/clickhouse.py b/app/worker/clickhouse.py index 5980d18..cd059d0 100644 --- a/app/worker/clickhouse.py +++ b/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}", diff --git a/app/worker/consumer.py b/app/worker/consumer.py index 7433f83..f43f9d2 100644 --- a/app/worker/consumer.py +++ b/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() diff --git a/tests/test_delivery_regressions.py b/tests/test_delivery_regressions.py new file mode 100644 index 0000000..76b8cdb --- /dev/null +++ b/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() diff --git a/tests/test_worker.py b/tests/test_worker.py index 09e94af..964c87a 100644 --- a/tests/test_worker.py +++ b/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): diff --git a/tests/test_worker_lifecycle.py b/tests/test_worker_lifecycle.py new file mode 100644 index 0000000..888afb6 --- /dev/null +++ b/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()