ruff fixes, + extra doc about lock in producer

Author: HardMax71

backend/app/db/repositories/event_repository.py

@@ -125,18 +125,11 @@ async def get_events_by_aggregate(
         docs = await cursor.to_list(length=limit)
         return [self.mapper.from_mongo_document(doc) for doc in docs]
 
-    async def get_events_by_correlation(
-        self, correlation_id: str, limit: int = 100, skip: int = 0
-    ) -> EventListResult:
+    async def get_events_by_correlation(self, correlation_id: str, limit: int = 100, skip: int = 0) -> EventListResult:
         query: dict[str, Any] = {EventFields.METADATA_CORRELATION_ID: correlation_id}
         total_count = await self._collection.count_documents(query)
 
-        cursor = (
-            self._collection.find(query)
-            .sort(EventFields.TIMESTAMP, ASCENDING)
-            .skip(skip)
-            .limit(limit)
-        )
+        cursor = self._collection.find(query).sort(EventFields.TIMESTAMP, ASCENDING).skip(skip).limit(limit)
         docs = await cursor.to_list(length=limit)
         return EventListResult(
             events=[self.mapper.from_mongo_document(doc) for doc in docs],
@@ -166,18 +159,11 @@ async def get_events_by_user(
         docs = await cursor.to_list(length=limit)
         return [self.mapper.from_mongo_document(doc) for doc in docs]
 
-    async def get_execution_events(
-        self, execution_id: str, limit: int = 100, skip: int = 0
-    ) -> EventListResult:
+    async def get_execution_events(self, execution_id: str, limit: int = 100, skip: int = 0) -> EventListResult:
         query = {"$or": [{EventFields.PAYLOAD_EXECUTION_ID: execution_id}, {EventFields.AGGREGATE_ID: execution_id}]}
         total_count = await self._collection.count_documents(query)
 
-        cursor = (
-            self._collection.find(query)
-            .sort(EventFields.TIMESTAMP, ASCENDING)
-            .skip(skip)
-            .limit(limit)
-        )
+        cursor = self._collection.find(query).sort(EventFields.TIMESTAMP, ASCENDING).skip(skip).limit(limit)
         docs = await cursor.to_list(length=limit)
         return EventListResult(
             events=[self.mapper.from_mongo_document(doc) for doc in docs],

backend/app/events/core/producer.py

@@ -8,10 +8,6 @@
 from confluent_kafka import Message, Producer
 from confluent_kafka.error import KafkaError
 
-# Global lock to serialize Producer initialization (workaround for librdkafka race condition)
-# See: https://github.com/confluentinc/confluent-kafka-python/issues/1797
-_producer_init_lock = threading.Lock()
-
 from app.core.lifecycle import LifecycleEnabled
 from app.core.logging import logger
 from app.core.metrics.context import get_event_metrics
@@ -23,6 +19,10 @@
 
 from .types import ProducerConfig, ProducerMetrics, ProducerState
 
+# Global lock to serialize Producer initialization (workaround for librdkafka race condition)
+# See: https://github.com/confluentinc/confluent-kafka-python/issues/1797
+_producer_init_lock = threading.Lock()
+
 DeliveryCallback: TypeAlias = Callable[[KafkaError | None, Message], None]
 StatsCallback: TypeAlias = Callable[[dict[str, Any]], None]
 

backend/app/infrastructure/mappers/event_mapper.py

@@ -78,7 +78,6 @@ def from_mongo_document(document: dict[str, Any]) -> Event:
         )
 
 
-
 class EventSummaryMapper:
     """Handles EventSummary serialization."""
 
@@ -211,5 +210,3 @@ def from_admin_pydantic(pflt: AdminEventFilter) -> EventFilter:
             search_text=pflt.search_text,
             text_search=pflt.search_text,
         )
-
-

backend/app/services/event_service.py

@@ -37,9 +37,7 @@ async def get_execution_events(
         limit: int = 1000,
         skip: int = 0,
     ) -> EventListResult | None:
-        result = await self.repository.get_execution_events(
-            execution_id=execution_id, limit=limit, skip=skip
-        )
+        result = await self.repository.get_execution_events(execution_id=execution_id, limit=limit, skip=skip)
         if not result.events:
             return EventListResult(events=[], total=0, skip=skip, limit=limit, has_more=False)
 
@@ -132,9 +130,7 @@ async def get_events_by_correlation(
         limit: int = 100,
         skip: int = 0,
     ) -> EventListResult:
-        result = await self.repository.get_events_by_correlation(
-            correlation_id=correlation_id, limit=limit, skip=skip
-        )
+        result = await self.repository.get_events_by_correlation(correlation_id=correlation_id, limit=limit, skip=skip)
         if not include_all_users or user_role != UserRole.ADMIN:
             filtered = [e for e in result.events if (e.metadata and e.metadata.user_id == user_id)]
             return EventListResult(

docs/testing/kafka-test-stability.md

@@ -0,0 +1,86 @@
+# Kafka test stability
+
+## The problem
+
+When running tests in parallel (e.g., with `pytest-xdist`), you might encounter sporadic crashes with messages like:
+
+```
+Fatal Python error: Aborted
+```
+
+The stack trace typically points to `confluent_kafka` operations, often during producer initialization in fixtures or test setup. This isn't a bug in the application code - it's a known race condition in the underlying `librdkafka` C library.
+
+## Why it happens
+
+The `confluent-kafka-python` library is a thin wrapper around `librdkafka`, a high-performance C library. When multiple Python processes or threads try to create Kafka `Producer` instances simultaneously, they can trigger a race condition in `librdkafka`'s internal initialization routines.
+
+This manifests as:
+
+- Random `SIGABRT` signals during test runs
+- Crashes in `rd_kafka_broker_destroy_final` or similar internal functions
+- Flaky CI failures that pass on retry
+
+The issue is particularly common in CI environments where tests run in parallel across multiple workers.
+
+## The fix
+
+The solution is to serialize `Producer` initialization using a global threading lock. This prevents multiple threads from entering `librdkafka`'s initialization code simultaneously.
+
+In `app/events/core/producer.py`:
+
+```python
+import threading
+
+# Global lock to serialize Producer initialization (workaround for librdkafka race condition)
+# See: https://github.com/confluentinc/confluent-kafka-python/issues/1797
+_producer_init_lock = threading.Lock()
+
+class UnifiedProducer:
+    async def start(self) -> None:
+        # ... config setup ...
+
+        # Serialize Producer initialization to prevent librdkafka race condition
+        with _producer_init_lock:
+            self._producer = Producer(producer_config)
+
+        # ... rest of startup ...
+```
+
+The lock is process-global, so all `UnifiedProducer` instances in the same process will serialize their initialization. This adds negligible overhead in production (producers are typically created once at startup) while eliminating the race condition in tests.
+
+## Related issues
+
+These GitHub issues document the underlying problem:
+
+| Issue | Description |
+|-------|-------------|
+| [confluent-kafka-python#1797](https://github.com/confluentinc/confluent-kafka-python/issues/1797) | Segfaults in multithreaded/asyncio pytest environments |
+| [confluent-kafka-python#1761](https://github.com/confluentinc/confluent-kafka-python/issues/1761) | Segfault on garbage collection in multithreaded context |
+| [librdkafka#3608](https://github.com/confluentinc/librdkafka/issues/3608) | Crash in `rd_kafka_broker_destroy_final` |
+
+## Alternative approaches
+
+If you still encounter issues:
+
+1. **Reduce parallelism** - Run Kafka-dependent tests with fewer workers: `pytest -n 2` instead of `-n auto`
+
+2. **Isolate Kafka tests** - Mark Kafka tests and run them separately:
+   ```python
+   @pytest.mark.kafka
+   def test_producer_sends_message():
+       ...
+   ```
+   ```bash
+   pytest -m "not kafka" -n auto  # parallel
+   pytest -m kafka -n 1           # sequential
+   ```
+
+3. **Use fixtures carefully** - Ensure producer fixtures are properly scoped and cleaned up:
+   ```python
+   @pytest.fixture(scope="function")
+   async def producer():
+       p = UnifiedProducer(config, schema_registry)
+       await p.start()
+       yield p
+       await p.stop()  # Always clean up
+   ```

mkdocs.yml

@@ -139,3 +139,4 @@ nav:
   - Testing:
     - Load Testing: testing/load-testing.md
     - Frontend Testing: testing/frontend-testing.md
+    - Kafka Test Stability: testing/kafka-test-stability.md