chore: moved init stuff to DI #113
Conversation
📝 WalkthroughWalkthroughThis PR executes a large-scale architectural refactoring to eliminate lifecycle-managed stateful services in favor of a stateless, dependency-injection-driven event-handler pattern. Key changes include removing the LifecycleEnabled base class, converting services to handler-based event processors, introducing Redis-backed repositories for state persistence, and simplifying the DI provider structure. The refactor affects event processing, DLQ management, orchestration, SSE streaming, and worker entrypoints across the entire backend. Changes
Sequence Diagram(s)The scale and heterogeneity of this refactoring across disparate architectural layers (DI container, event processing, services, repositories, workers) with multiple independent transformations (lifecycle removal, state migration to Redis, handler-based patterns, type enum conversions) does not form a single coherent sequence. The changes span distinct concerns without a unified sequential flow that would benefit from visualization. Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 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
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 |
HardMax71
force-pushed
the
chore/complete-di
branch
from
3366a1f to
9271f12
Compare
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
14 issues found across 59 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/services/event_bus.py">
<violation number="1" location="backend/app/services/event_bus.py:138">
P2: Synchronous handlers run directly on the event loop, which can block async processing. Offload sync handlers to a thread executor to avoid stalling Kafka message handling.</violation>
</file>
<file name="backend/workers/run_pod_monitor.py">
<violation number="1" location="backend/workers/run_pod_monitor.py:41">
P2: Ensure cleanup runs even when the consumer loop is cancelled or errors. Wrap the loop in a try/finally so container.close executes on shutdown.</violation>
</file>
<file name="backend/app/db/repositories/resource_repository.py">
<violation number="1" location="backend/app/db/repositories/resource_repository.py:145">
P1: Validate resource requests are non-negative before allocation. Negative values would increase the available pool, allowing callers to mint resources.</violation>
</file>
<file name="backend/workers/run_coordinator.py">
<violation number="1" location="backend/workers/run_coordinator.py:42">
P2: Ensure container cleanup runs even if the consumer loop raises by wrapping the loop in a try/finally and moving `container.close()` into the finally block.</violation>
</file>
<file name="backend/workers/run_k8s_worker.py">
<violation number="1" location="backend/workers/run_k8s_worker.py:41">
P2: Wrap the consume loop in `try/finally` so the DI container is closed even when message handling raises, otherwise resources remain open and the worker can leak connections on failures.</violation>
</file>
<file name="backend/app/services/pod_monitor/monitor.py">
<violation number="1" location="backend/app/services/pod_monitor/monitor.py:187">
P2: Pod events are no longer persisted to the EventRepository before producing to Kafka. This removes the audit copy previously stored via KafkaEventService and can break consumers that rely on stored events.</violation>
</file>
<file name="backend/workers/run_event_replay.py">
<violation number="1" location="backend/workers/run_event_replay.py:33">
P2: On shutdown (cancellation), `await asyncio.Event().wait()` raises `CancelledError` and the container is never closed, leaking resources. Wrap the wait in a try/finally so `container.close()` always runs.</violation>
</file>
<file name="backend/app/events/core/consumer.py">
<violation number="1" location="backend/app/events/core/consumer.py:47">
P2: Guard against missing Kafka headers. This line will crash on messages produced without headers (producer sends `headers=None`). Use a fallback for `msg.headers` and decode only when a value is present.</violation>
</file>
<file name="backend/app/db/repositories/execution_queue_repository.py">
<violation number="1" location="backend/app/db/repositories/execution_queue_repository.py:72">
P2: Queue size and per-user limits are checked outside an atomic transaction, so concurrent enqueues can exceed the limits despite the checks.</violation>
<violation number="2" location="backend/app/db/repositories/execution_queue_repository.py:130">
P2: When queue data is missing, the user count is never decremented, which can permanently inflate per-user limits and block future enqueues.</violation>
<violation number="3" location="backend/app/db/repositories/execution_queue_repository.py:221">
P1: cleanup_stale will delete non-stale entries for priorities 0–9 because the threshold uses BACKGROUND’s priority offset, making the score range include all higher-priority items regardless of timestamp.</violation>
</file>
<file name="backend/workers/run_saga_orchestrator.py">
<violation number="1" location="backend/workers/run_saga_orchestrator.py:41">
P2: Cleanup is not guaranteed if message handling raises or the task is cancelled. Wrap the consumer loop in try/finally so container.close() always runs.</violation>
</file>
<file name="backend/app/services/coordinator/coordinator.py">
<violation number="1" location="backend/app/services/coordinator/coordinator.py:115">
P2: After cancellation, the coordinator doesn’t attempt to schedule the next queued execution. With the scheduling loop removed, this can leave the queue stalled until another event triggers scheduling. Consider calling `_try_schedule_next()` after cancellation cleanup.</violation>
</file>
<file name="backend/tests/e2e/test_k8s_worker_create_pod.py">
<violation number="1" location="backend/tests/e2e/test_k8s_worker_create_pod.py:23">
P2: Using the DI-managed worker ignores `test_settings.K8S_NAMESPACE`; the worker’s `K8sWorkerConfig` namespace comes from the environment, so the test may create resources in a different namespace than the one it asserts/cleans up. This can lead to flaky failures or leaked resources when test settings differ from the env.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
| if asyncio.iscoroutinefunction(handler): | ||
| await handler(event) | ||
| else: | ||
| handler(event) # type: ignore[operator] |
There was a problem hiding this comment.
P2: Synchronous handlers run directly on the event loop, which can block async processing. Offload sync handlers to a thread executor to avoid stalling Kafka message handling.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At backend/app/services/event_bus.py, line 138:
<comment>Synchronous handlers run directly on the event loop, which can block async processing. Offload sync handlers to a thread executor to avoid stalling Kafka message handling.</comment>
<file context>
@@ -35,313 +39,102 @@ class Subscription:
+ if asyncio.iscoroutinefunction(handler):
+ await handler(event)
+ else:
+ handler(event) # type: ignore[operator]
+ except Exception as e:
+ self._logger.error(f"Handler failed for {event.event_type}: {e}")
</file context>
| handler(event) # type: ignore[operator] | |
| await asyncio.to_thread(handler, event) |
| async for msg in kafka_consumer: | ||
| await handler.handle(msg) | ||
| await kafka_consumer.commit() | ||
|
|
||
| logger.info("PodMonitor shutdown complete") | ||
|
|
||
| await container.close() |
There was a problem hiding this comment.
P2: Ensure cleanup runs even when the consumer loop is cancelled or errors. Wrap the loop in a try/finally so container.close executes on shutdown.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At backend/workers/run_pod_monitor.py, line 41:
<comment>Ensure cleanup runs even when the consumer loop is cancelled or errors. Wrap the loop in a try/finally so container.close executes on shutdown.</comment>
<file context>
@@ -29,27 +33,18 @@ async def run_pod_monitor(settings: Settings) -> None:
+
+ logger.info("PodMonitor started, consuming events...")
+
+ async for msg in kafka_consumer:
+ await handler.handle(msg)
+ await kafka_consumer.commit()
</file context>
| async for msg in kafka_consumer: | |
| await handler.handle(msg) | |
| await kafka_consumer.commit() | |
| logger.info("PodMonitor shutdown complete") | |
| await container.close() | |
| try: | |
| async for msg in kafka_consumer: | |
| await handler.handle(msg) | |
| await kafka_consumer.commit() | |
| finally: | |
| logger.info("PodMonitor shutdown complete") | |
| await container.close() |
| async for msg in kafka_consumer: | ||
| await handler.handle(msg) | ||
| await kafka_consumer.commit() | ||
|
|
||
| logger.info("ExecutionCoordinator shutdown complete") | ||
|
|
||
| await container.close() |
There was a problem hiding this comment.
P2: Ensure container cleanup runs even if the consumer loop raises by wrapping the loop in a try/finally and moving container.close() into the finally block.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At backend/workers/run_coordinator.py, line 42:
<comment>Ensure container cleanup runs even if the consumer loop raises by wrapping the loop in a try/finally and moving `container.close()` into the finally block.</comment>
<file context>
@@ -27,27 +34,18 @@ async def run_coordinator(settings: Settings) -> None:
+
+ logger.info("ExecutionCoordinator started, consuming events...")
+
+ async for msg in kafka_consumer:
+ await handler.handle(msg)
+ await kafka_consumer.commit()
</file context>
| async for msg in kafka_consumer: | |
| await handler.handle(msg) | |
| await kafka_consumer.commit() | |
| logger.info("ExecutionCoordinator shutdown complete") | |
| await container.close() | |
| try: | |
| async for msg in kafka_consumer: | |
| await handler.handle(msg) | |
| await kafka_consumer.commit() | |
| finally: | |
| logger.info("ExecutionCoordinator shutdown complete") | |
| await container.close() |
| logger.info("SagaOrchestrator started, consuming events...") | ||
|
|
||
| logger.info("Saga orchestrator started and running") | ||
| async for msg in kafka_consumer: |
There was a problem hiding this comment.
P2: Cleanup is not guaranteed if message handling raises or the task is cancelled. Wrap the consumer loop in try/finally so container.close() always runs.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At backend/workers/run_saga_orchestrator.py, line 41:
<comment>Cleanup is not guaranteed if message handling raises or the task is cancelled. Wrap the consumer loop in try/finally so container.close() always runs.</comment>
<file context>
@@ -27,27 +33,18 @@ async def run_saga_orchestrator(settings: Settings) -> None:
+ logger.info("SagaOrchestrator started, consuming events...")
- logger.info("Saga orchestrator started and running")
+ async for msg in kafka_consumer:
+ await handler.handle(msg)
+ await kafka_consumer.commit()
</file context>
| idem: IdempotencyManager = await scope.get(IdempotencyManager) | ||
| event_metrics: EventMetrics = await scope.get(EventMetrics) | ||
| # Get worker from DI (already configured with dependencies) | ||
| worker: KubernetesWorker = await scope.get(KubernetesWorker) |
There was a problem hiding this comment.
P2: Using the DI-managed worker ignores test_settings.K8S_NAMESPACE; the worker’s K8sWorkerConfig namespace comes from the environment, so the test may create resources in a different namespace than the one it asserts/cleans up. This can lead to flaky failures or leaked resources when test settings differ from the env.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At backend/tests/e2e/test_k8s_worker_create_pod.py, line 23:
<comment>Using the DI-managed worker ignores `test_settings.K8S_NAMESPACE`; the worker’s `K8sWorkerConfig` namespace comes from the environment, so the test may create resources in a different namespace than the one it asserts/cleans up. This can lead to flaky failures or leaked resources when test settings differ from the env.</comment>
<file context>
@@ -25,27 +19,10 @@ async def test_worker_creates_configmap_and_pod(
- idem: IdempotencyManager = await scope.get(IdempotencyManager)
- event_metrics: EventMetrics = await scope.get(EventMetrics)
+ # Get worker from DI (already configured with dependencies)
+ worker: KubernetesWorker = await scope.get(KubernetesWorker)
- cfg = K8sWorkerConfig(namespace=ns, max_concurrent_pods=1)
</file context>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 17
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
backend/app/services/idempotency/middleware.py (1)
50-56: Type mismatch:key_strategyisstrbutIdempotencyManagerexpectsKeyStrategyenum.The pipeline failures confirm that
IdempotencyManager.check_and_reserve(),mark_completed(), andmark_failed()expectKeyStrategyenum, notstr. Either:
- Update
IdempotencyManagerto acceptstr, or- Keep
KeyStrategyenum in this module🔧 Option 1: Convert string to enum at call sites
+from app.domain.idempotency import KeyStrategy + # Check idempotency idempotency_result = await self.idempotency_manager.check_and_reserve( event=event, - key_strategy=self.key_strategy, + key_strategy=KeyStrategy(self.key_strategy), custom_key=custom_key, ttl_seconds=self.ttl_seconds, fields=self.fields, )Apply similar changes to
mark_completed()(line 81) andmark_failed()(line 87).backend/app/services/notification_service.py (1)
350-369: Throttle exemption is ineffective and non-exempt notifications get double-counted.
_create_system_for_userchecks throttling and then callscreate_notification, which throttles again. Also,throttle_exempt=Truestill gets throttled increate_notification.🛠️ Suggested fix
- async def create_notification( + async def create_notification( self, user_id: str, subject: str, body: str, tags: list[str], severity: NotificationSeverity = NotificationSeverity.MEDIUM, channel: NotificationChannel = NotificationChannel.IN_APP, scheduled_for: datetime | None = None, action_url: str | None = None, - metadata: dict[str, object] | None = None, + metadata: dict[str, object] | None = None, + skip_throttle: bool = False, ) -> DomainNotification: """Create a new notification.""" if not tags: raise NotificationValidationError("tags must be a non-empty list") @@ - if await self._throttle_cache.check_throttle( + if not skip_throttle and await self._throttle_cache.check_throttle( user_id, severity, window_hours=self._settings.NOTIF_THROTTLE_WINDOW_HOURS, max_per_hour=self._settings.NOTIF_THROTTLE_MAX_PER_HOUR, ): @@ - if not cfg.throttle_exempt: - throttled = await self._throttle_cache.check_throttle( - user_id, - cfg.severity, - window_hours=self._settings.NOTIF_THROTTLE_WINDOW_HOURS, - max_per_hour=self._settings.NOTIF_THROTTLE_MAX_PER_HOUR, - ) - if throttled: - return "throttled" - await self.create_notification( user_id=user_id, subject=title, body=str(base_context.get("message", "Alert")), severity=cfg.severity, tags=tags, channel=NotificationChannel.IN_APP, metadata=base_context, + skip_throttle=cfg.throttle_exempt, ) return "created" + except NotificationThrottledError: + return "throttled"backend/tests/e2e/test_k8s_worker_create_pod.py (1)
47-71: Ensure cleanup runs even if the test fails.If pod creation or assertions raise, cleanup won’t run and resources can leak.
🧹 Suggested fix
- pod = worker._pod_builder.build_pod_manifest(cmd) # noqa: SLF001 - await worker._create_pod(pod) # noqa: SLF001 - - # Verify resources exist - got_cm = worker._v1.read_namespaced_config_map(name=f"script-{exec_id}", namespace=ns) # noqa: SLF001 - assert got_cm is not None - got_pod = worker._v1.read_namespaced_pod(name=f"executor-{exec_id}", namespace=ns) # noqa: SLF001 - assert got_pod is not None - - # Cleanup - worker._v1.delete_namespaced_pod(name=f"executor-{exec_id}", namespace=ns) # noqa: SLF001 - worker._v1.delete_namespaced_config_map(name=f"script-{exec_id}", namespace=ns) # noqa: SLF001 + pod = worker._pod_builder.build_pod_manifest(cmd) # noqa: SLF001 + try: + await worker._create_pod(pod) # noqa: SLF001 + + # Verify resources exist + got_cm = worker._v1.read_namespaced_config_map(name=f"script-{exec_id}", namespace=ns) # noqa: SLF001 + assert got_cm is not None + got_pod = worker._v1.read_namespaced_pod(name=f"executor-{exec_id}", namespace=ns) # noqa: SLF001 + assert got_pod is not None + finally: + if worker._v1 is not None: # noqa: SLF001 + try: + worker._v1.delete_namespaced_pod(name=f"executor-{exec_id}", namespace=ns) # noqa: SLF001 + except ApiException as e: + if e.status != 404: + raise + try: + worker._v1.delete_namespaced_config_map(name=f"script-{exec_id}", namespace=ns) # noqa: SLF001 + except ApiException as e: + if e.status != 404: + raisebackend/app/services/result_processor/processor.py (1)
96-104: Fix ExecutionResultDomain metadata type (MyPy failure).
metadataexpectsEventMetadata | None, butmodel_dump()produces a dict. Either pass theEventMetadatainstance directly or widen the domain type.🔧 Suggested fix
- metadata=event.metadata.model_dump(), + metadata=event.metadata, @@ - metadata=event.metadata.model_dump(), + metadata=event.metadata, @@ - metadata=event.metadata.model_dump(), + metadata=event.metadata,Also applies to: 123-131, 154-162
🤖 Fix all issues with AI agents
In `@backend/app/core/providers.py`:
- Around line 131-170: The provider currently hardcodes limits in
RedisRepositoryProvider; update get_execution_queue_repository and
get_resource_repository to read max_queue_size, max_executions_per_user,
total_cpu_cores, and total_memory_mb from the Settings argument instead of
hardcoded literals; add those fields with sensible defaults to the Settings
class (e.g., max_queue_size, max_executions_per_user, total_cpu_cores,
total_memory_mb), then pass settings.max_queue_size and
settings.max_executions_per_user into ExecutionQueueRepository and
settings.total_cpu_cores and settings.total_memory_mb into ResourceRepository
and keep existing initialize() call.
In `@backend/app/db/repositories/execution_queue_repository.py`:
- Around line 218-229: The cleanup_stale method computes a single huge
threshold_score that causes all non-BACKGROUND priorities to be considered
stale; update cleanup_stale to iterate over each QueuePriority and for each
priority compute min_score = priority.value * 1e12 and max_score =
priority.value * 1e12 + (time.time() - self.stale_timeout_seconds) (or
alternatively fetch and check the stored _enqueue_timestamp in the entry hash)
then call zrangebyscore with that per-priority range, decode entries and call
self.remove(execution_id) as before; ensure you still limit batch size (num=100)
and aggregate removed count.
- Around line 71-101: The current enqueue logic performs zcard and hincrby(...,
0) checks outside the pipeline which allows race conditions; change the
implementation in the enqueue routine to perform the queue-size and
per-user-count checks and the zadd + data hset/hincrby/expire atomically by
moving them into a single Redis Lua script (or by using WATCH on QUEUE_KEY and
USER_COUNT_KEY then MULTI/EXEC). Specifically, include the zcard(QUEUE_KEY) and
hincrby(USER_COUNT_KEY, user_id, 0) checks, the zadd(QUEUE_KEY, execution_id,
score), the creation of data_key and its hset/expire, and the
hincrby(USER_COUNT_KEY, user_id, 1) increment inside the atomic operation so the
old pipeline usage (pipe.zadd, pipe.hset, pipe.expire, pipe.hincrby) is executed
only from within the Lua script or guarded MULTI after WATCH; ensure the script
returns appropriate failure/success values (matching the existing
False/None/"Queue is full" or limit-exceeded responses) so the calling code can
handle errors the same way.
In `@backend/app/db/repositories/resource_repository.py`:
- Around line 216-236: The release method has a race where it reads the
allocation then updates the pool separately; change release to perform the read,
increment and delete atomically via a Redis Lua script (similar to allocate()):
create a script that takes KEYS[1]=self.POOL_KEY and KEYS[2]=alloc_key, reads
'cpu','memory','gpu' from the alloc hash (coerce to numbers, default 0), returns
0 if all zero, otherwise calls HINCRBYFLOAT/HINCRBY on the pool fields and DEL
on the alloc key and returns 1; in release() call that script with the two keys
(using self._redis.eval or the client script API), interpret the numeric return
as the boolean result and remove the existing pipeline-based logic and prior
hgetall read. Ensure you reference ALLOC_KEY_PREFIX when building alloc_key and
keep behavior of logging/returning False when script returns 0.
- Around line 117-193: The allocate method currently does a non-atomic
Python-side existence check before running the Lua script; move that existence
check into the Lua script by testing if alloc_key exists (redis.call('EXISTS',
alloc_key)) at the top of lua_script and return a distinct code (e.g., 2) when
it does, keep returning 0 for insufficient resources and 1 for successful
allocation; then update the Python allocate method (the result handling after
self._redis.eval) to treat result == 2 by reading the existing allocation from
alloc_key and returning a ResourceAllocation built from those fields (same
parsing logic used earlier), otherwise proceed as before for success/failure.
- Around line 174-176: alloc_key entries are being set with a 7200s TTL but pool
resource counters (updated in the allocation code around alloc_key and the pool
keys) are never restored if release() isn't called, causing permanent capacity
leaks; add a reconciliation background task (e.g., reclaim_expired_allocations
or reconcile_expired_allocations) that periodically scans redis for keys
matching "alloc_key:*", detects expired or missing allocation metadata and
atomically restores the pool counters (cpu, memory, gpu) by incrementing the
corresponding pool hash fields, using Redis transactions or a Lua script to
avoid races, and mark or remove reclaimed alloc_key entries to make reclamation
idempotent; alternatively add TTLs to pool keys and implement a periodic
reconcile_pools task that verifies sums against active allocations and fixes
discrepancies—update resource_repository functions (alloc_key handling and
release()) to cooperate with the new reconciler.
In `@backend/app/services/coordinator/coordinator.py`:
- Around line 194-204: The current flow dequeues an execution and, on allocation
failure, removes its claim and requeues it via self._state_repo.remove and
self._queue_repo.enqueue, which loses original queue position and can cause
busy-loops; change this by checking resource availability before dequeuing (use
or add a can_allocate or resources_available check in the coordinator) or move
the execution into a "waiting for resources" state instead of re-enqueueing at
the tail, and if you must requeue implement exponential/backoff delay before
calling self._queue_repo.enqueue or mark the item with a waiting flag so dequeue
logic preserves position and avoids tight requeue cycles (update the code paths
around dequeue, allocation, allocation, _state_repo.remove, and
_queue_repo.enqueue accordingly).
- Around line 97-100: handle_execution_requested currently logs exceptions and
records metrics but does not notify downstream systems; update the except block
in handle_execution_requested to call self._publish_scheduling_failed(event,
str(e)) before/after recording the metric so an ExecutionFailedEvent is
published (mirror the behavior in _schedule_execution), keeping the existing
self._logger.error and
self._metrics.record_coordinator_execution_scheduled("error") calls.
In `@backend/app/services/event_bus.py`:
- Around line 74-82: The current try/except around self._producer.send_and_wait
in event_bus (the block catching Exception as e) only logs failures and swallows
the exception; update the error-handling so callers receive failure info —
either re-raise the caught exception after logging or change the method
signature to return a boolean and return False on failure (and True on success);
locate the send_and_wait call and its except Exception as e handler in the
publish/publish_event method, add the chosen behavior (re-raise e after
self._logger.error(...) or return False), and ensure successful paths return
True if using the success indicator approach.
In `@backend/app/services/notification_service.py`:
- Around line 170-174: The truncation code assumes event_data["stdout"] and
["stderr"] are strings but they can be None; update the logic around event_data
(produced by event.model_dump) to guard against None before slicing—e.g.,
replace direct slicing with a safe truncation that uses a default empty string
when the key is None or missing, and apply the same guard for both "stdout" and
"stderr" to prevent TypeError in NotificationService (look for event_data,
"stdout", and "stderr" usage in notification_service.py).
In `@backend/tests/unit/services/pod_monitor/test_monitor.py`:
- Line 15: Replace the incorrect import of ResourceUsageDomain in
test_monitor.py: locate the module that actually defines/exports
ResourceUsageDomain (instead of app.domain.execution.models) and import
ResourceUsageDomain from that module (update the import statement that currently
references ResourceUsageDomain). Ensure the import uses the exact exported
symbol name ResourceUsageDomain so MyPy recognizes it.
In `@backend/tests/unit/services/sse/test_sse_service.py`:
- Around line 13-14: The import for ResourceUsageDomain is incorrect: update the
import line that currently pulls DomainExecution and ResourceUsageDomain from
app.domain.execution so that DomainExecution remains imported from
app.domain.execution and ResourceUsageDomain is imported from
app.domain.events.typed; specifically adjust the import statement referencing
DomainExecution and ResourceUsageDomain in test_sse_service.py so
DomainExecution stays from app.domain.execution and ResourceUsageDomain is
imported from app.domain.events.typed (keeping SSEExecutionStatusDomain and
ShutdownStatus as-is).
In `@backend/workers/run_coordinator.py`:
- Around line 37-48: The async consumer loop in ExecutionCoordinator using
kafka_consumer and handler can raise exceptions which skip cleanup; wrap the
async for msg in kafka_consumer loop in a try/finally so that await
container.close() (and any kafka_consumer commit/close if needed) always runs,
move the "ExecutionCoordinator shutdown complete" log into the finally block,
and ensure resource-specific cleanup for AIOKafkaConsumer and UnifiedConsumer
happens before closing the container (referencing kafka_consumer, handler,
container, AIOKafkaConsumer, UnifiedConsumer).
In `@backend/workers/run_event_replay.py`:
- Around line 32-35: The forever-blocking await asyncio.Event().wait() prevents
container.close() from ever running; replace the anonymous Event with a named
asyncio.Event (e.g., shutdown_event), register signal handlers for
SIGINT/SIGTERM that set shutdown_event (using loop.add_signal_handler or
signal.signal fallback), await shutdown_event.wait() in run_event_replay.py,
then call await container.close() after the wait so the container is closed
during graceful shutdown; reference the existing asyncio.Event().wait() and
container.close() calls when making the change.
In `@backend/workers/run_k8s_worker.py`:
- Around line 36-47: The consumer loop can raise in handler.handle() or
kafka_consumer.commit() and skip cleanup; wrap the async for msg in
kafka_consumer loop in a try/finally so that await container.close() (and any
other shutdown steps) always run. Specifically, enclose the async for ...: await
handler.handle(msg); await kafka_consumer.commit() inside try: ... finally:
await container.close() and keep the logger.info("KubernetesWorker shutdown
complete") inside the finally; apply the same pattern to the other worker
modules (run_result_processor.py, run_saga_orchestrator.py, run_coordinator.py,
run_pod_monitor.py) referencing kafka_consumer, UnifiedConsumer.handle,
kafka_consumer.commit, and container.close where present.
In `@backend/workers/run_result_processor.py`:
- Around line 36-47: Wrap the consumer loop in a try/finally so cleanup always
runs: start by acquiring kafka_consumer (AIOKafkaConsumer) and handler
(UnifiedConsumer) as before, then run the async for msg in kafka_consumer /
await handler.handle(msg) loop inside a try block; in the finally block ensure
you stop/close the kafka_consumer (call kafka_consumer.stop() or equivalent) and
then await container.close(); re-raise cancellation errors if needed so signal
handling still works. This guarantees the consumer and container are closed even
if handler.handle() raises or the task is cancelled.
In `@backend/workers/run_saga_orchestrator.py`:
- Around line 41-47: Add graceful shutdown and per-message error resilience
around the kafka_consumer loop: install asyncio signal handlers for
SIGINT/SIGTERM that set an asyncio.Event (e.g., shutdown_event) and make the
async for msg in kafka_consumer loop respect that event so it stops cleanly;
wrap message processing so each iteration calls await handler.handle(msg) inside
a try/except that logs the exception via logger.error (including exception
details) and continues (optionally still commit offsets on successful processing
via kafka_consumer.commit), and wrap the whole consumer loop in try/finally to
ensure await container.close() and any kafka_consumer stop/close are always
awaited on shutdown. Reference symbols: kafka_consumer, handler.handle,
kafka_consumer.commit, container.close, logger.
🟡 Minor comments (9)
backend/tests/e2e/idempotency/test_consumer_idempotent.py-81-97 (1)
81-97: Test bypassesIdempotentConsumerWrapper- idempotency not exercised.The
IdempotentConsumerWrapperis instantiated but the consume loop callshandler.handle(msg)directly (line 94), bypassing the wrapper entirely. The test verifies thatseen["n"] >= 1but doesn't verify duplicate blocking.The wrapper's
_wrap_handlers()wraps handlers in the dispatcher, buthandler.handle()dispatches through the dispatcher, which should work. However, the assertionseen["n"] >= 1doesn't verify that duplicates were blocked (should be exactly 1 if idempotency works).💡 Suggested assertion fix to verify idempotency
- assert seen["n"] >= 1 + # Both messages produced were duplicates (same event_id), so idempotency should block one + assert seen["n"] == 1, f"Expected exactly 1 (duplicates blocked), got {seen['n']}"backend/workers/run_pod_monitor.py-36-47 (1)
36-47: Guarantee cleanup on handler failures.
Ifhandler.handle()orkafka_consumer.commit()raises an exception, execution skips thecontainer.close()call. Wrap the loop withtry/finallyto ensure shutdown always runs.🔧 Suggested fix
- async for msg in kafka_consumer: - await handler.handle(msg) - await kafka_consumer.commit() - - logger.info("PodMonitor shutdown complete") - - await container.close() + try: + async for msg in kafka_consumer: + await handler.handle(msg) + await kafka_consumer.commit() + finally: + logger.info("PodMonitor shutdown complete") + await container.close()backend/tests/e2e/events/test_consumer_lifecycle.py-21-45 (1)
21-45: Ensure partitions are assigned before calling seek operations.After
await kafka_consumer.start()(line 37),assignment()may return empty because partition assignment happens asynchronously via group coordination. This causes the test to become a no-op—theif assignment:guard prevents the seek calls from executing.Add a call to trigger assignment and assert it's non-empty:
🔧 Suggested fix
await kafka_consumer.start() try: # Exercise seek functions on AIOKafkaConsumer directly + await kafka_consumer.getmany(timeout_ms=1000) assignment = kafka_consumer.assignment() - if assignment: - await kafka_consumer.seek_to_beginning(*assignment) - await kafka_consumer.seek_to_end(*assignment) + assert assignment, "Expected partition assignment" + await kafka_consumer.seek_to_beginning(*assignment) + await kafka_consumer.seek_to_end(*assignment)backend/app/db/repositories/pod_state_repository.py-138-145 (1)
138-145: Add error handling for malformed datetime strings.If Redis contains corrupted data,
datetime.fromisoformat()will raise aValueError. Consider adding defensive handling.Proposed fix
+ try: + created_at = datetime.fromisoformat(get_str("created_at")) + updated_at = datetime.fromisoformat(get_str("updated_at")) + except ValueError as e: + self._logger.warning(f"Invalid datetime in pod state for {pod_name}: {e}") + return None + return PodState( pod_name=get_str("pod_name"), execution_id=get_str("execution_id"), status=get_str("status"), - created_at=datetime.fromisoformat(get_str("created_at")), - updated_at=datetime.fromisoformat(get_str("updated_at")), + created_at=created_at, + updated_at=updated_at, metadata=metadata, )backend/app/services/event_bus.py-111-120 (1)
111-120: Local subscribers don't receive events published by the same instance.The
handle_kafka_messagefilters out messages from its own instance (source_instance == self._instance_id), which prevents feedback loops in cross-instance pub/sub. However, this means thepublishmethod only distributes to Kafka—local subscribers never receive events published within the same instance.If local subscribers should receive locally-published events, add local distribution in the
publishmethod before sending to Kafka.backend/app/events/core/producer.py-54-54 (1)
54-54: Useget_topic_for_event()helper or handle missing mappings defensively.Direct dict access at line 54 will raise
KeyErrorfor unmapped EventType values.NOTIFICATION_ALL_READis currently unmapped and defined in the EventType enum but will fail if produced. A defensive helper functionget_topic_for_event()already exists in the mappings module with fallback toSYSTEM_EVENTStopic—use it instead of direct dict access, or add explicit error handling for missing types.Recommended fix
-from app.infrastructure.kafka.mappings import EVENT_TYPE_TO_TOPIC +from app.infrastructure.kafka.mappings import get_topic_for_eventThen at line 54:
- topic = f"{self._topic_prefix}{EVENT_TYPE_TO_TOPIC[event_to_produce.event_type]}" + topic = f"{self._topic_prefix}{get_topic_for_event(event_to_produce.event_type)}"backend/app/services/pod_monitor/monitor.py-94-115 (1)
94-115: Harden malformed event handling.
Ifraw_event["object"]isNoneor not pod-like,AttributeError/TypeErrorcan escape and break the worker loop. Consider widening the catch.🩹 Suggested tweak
- except (KeyError, ValueError) as e: + except (KeyError, ValueError, TypeError, AttributeError) as e: self._logger.error(f"Invalid event format: {e}")backend/app/services/saga/saga_orchestrator.py-391-402 (1)
391-402: Same silent exception swallowing issue.This has the same issue as in
_start_saga- the bareexcept Exception: passat lines 401-402 silently ignores errors frombind_dependencies. Apply the same fix to log the warning.🐛 Proposed fix
try: saga.bind_dependencies( producer=self._producer, alloc_repo=self._alloc_repo, publish_commands=bool(getattr(self._config, "publish_commands", False)), ) - except Exception: - pass + except Exception as e: + self._logger.warning(f"Failed to bind dependencies for saga cancellation {saga_instance.saga_name}: {e}")backend/app/services/saga/saga_orchestrator.py-204-212 (1)
204-212: Silent exception swallowing hides potential configuration errors.The bare
except Exception: passat lines 211-212 silently ignores all errors frombind_dependencies. This could mask important configuration issues or bugs. At minimum, log the exception for debugging purposes.🐛 Proposed fix
try: saga.bind_dependencies( producer=self._producer, alloc_repo=self._alloc_repo, publish_commands=bool(getattr(self._config, "publish_commands", False)), ) - except Exception: - pass + except Exception as e: + self._logger.warning(f"Failed to bind dependencies for saga {saga_name}: {e}")
🧹 Nitpick comments (14)
backend/tests/unit/services/sse/test_shutdown_manager.py (1)
67-80: Consider using small non-zero timeouts to reduce test flakiness.Setting
notification_timeout=0.0andforce_close_timeout=0.0can lead to unpredictable async behavior. The permissive assertion on line 80 (phase in ("draining", "complete", "closing", "notifying")) compensates for this but doesn't verify a specific transition path.If the intent is to test rapid transitions, consider using small non-zero values (e.g.,
0.001) to ensure the async machinery has minimal but consistent time to execute.♻️ Suggested improvement
m = SSEShutdownManager( drain_timeout=0.01, - notification_timeout=0.0, - force_close_timeout=0.0, + notification_timeout=0.001, + force_close_timeout=0.001, logger=_test_logger, connection_metrics=connection_metrics, )backend/app/services/kafka_event_service.py (1)
4-4: Prefer lowercasedictfor type hints in Python 3.9+.The codebase uses modern Python syntax (e.g.,
str | Noneon line 44), so importingDictfromtypingis unnecessary. Consider usingdict[str, Any]directly for consistency.♻️ Proposed fix
-from typing import Any, Dict +from typing import AnyThen update all
Dict[...]annotations todict[...]throughout the file.backend/app/services/idempotency/middleware.py (2)
21-23: Inconsistent typing: UsingSet[str]butstrfor key_strategy.If the intent is to accept string-based strategies, consider also using
KeyStrategyenum with string conversion, or document the valid string values as constants. The current approach loses type safety without clear documentation of valid values.
5-5: Prefer lowercasesetanddictfor type hints.Similar to the kafka_event_service.py feedback, Python 3.9+ supports lowercase generics directly.
♻️ Proposed fix
-from typing import Any, Awaitable, Callable, Dict, Set +from typing import Any, Awaitable, CallableThen use
set[str],dict[...]throughout.backend/tests/e2e/events/test_consume_roundtrip.py (1)
47-76: Align auto-commit with manual commits in the test loop.
Line 53 enables auto-commit while Line 76 commits manually; this can advance offsets beforehandler.handlecompletes. Consider disabling auto-commit when you explicitly commit.🔧 Suggested fix
- enable_auto_commit=True, + enable_auto_commit=False,backend/app/services/sse/kafka_redis_bridge.py (1)
51-84: Return a copy of relevant event types and keep status ordering deterministic.This prevents accidental mutation of the module-level set and avoids flaky status output ordering.
♻️ Proposed change
`@staticmethod` def get_relevant_event_types() -> set[EventType]: """Get event types that should be routed to SSE. Helper for worker entrypoint to know which topics to subscribe to. """ - return RELEVANT_EVENT_TYPES + return set(RELEVANT_EVENT_TYPES) @@ async def get_status(self) -> dict[str, list[str]]: """Get bridge status.""" return { - "relevant_event_types": [str(et) for et in RELEVANT_EVENT_TYPES], + "relevant_event_types": sorted(str(et) for et in RELEVANT_EVENT_TYPES), }backend/tests/e2e/events/test_event_dispatcher.py (1)
47-83: Disable auto-commit when manually committing offsets.With
enable_auto_commit=Trueand explicitcommit()calls, auto-commit runs in the background every ~5000ms and may commit offsets beforehandler.handle()completes. This creates a race condition where offset commits become non-deterministic, potentially introducing flakiness in this test. Official aiokafka documentation recommends disabling auto-commit when using manual commit.♻️ Proposed change
kafka_consumer = AIOKafkaConsumer( topic, bootstrap_servers=settings.KAFKA_BOOTSTRAP_SERVERS, group_id=group_id, - enable_auto_commit=True, + enable_auto_commit=False, auto_offset_reset="earliest", )backend/app/events/core/producer.py (1)
76-76: Consider lazy logging to avoid string formatting overhead.Using f-strings in logging statements causes the string to be formatted even when the log level is disabled. For debug/error logs in hot paths, prefer lazy formatting.
Example using lazy formatting
- self._logger.debug(f"Message [{event_to_produce}] sent to topic: {topic}") + self._logger.debug("Message [%s] sent to topic: %s", event_to_produce, topic)- self._logger.error(f"Failed to produce message: {e}") + self._logger.error("Failed to produce message: %s", e)Also applies to: 83-83
backend/app/db/repositories/pod_state_repository.py (1)
77-100:track_podoverwritescreated_aton re-tracking.When tracking a pod that was already tracked, this will overwrite the original
created_attimestamp. If preserving the original creation time is important, consider checking existence first or usingHSETNXfor thecreated_atfield.Alternative: preserve original created_at
async def track_pod( self, pod_name: str, execution_id: str, status: str, metadata: dict[str, object] | None = None, ttl_seconds: int = 7200, ) -> None: """Track a pod's state.""" key = f"{self.TRACKED_KEY_PREFIX}:{pod_name}" now = datetime.now(timezone.utc).isoformat() + # Check if already tracked to preserve original created_at + existing = await self._redis.hget(key, "created_at") + created_at = existing.decode() if existing else now + data = { "pod_name": pod_name, "execution_id": execution_id, "status": status, - "created_at": now, + "created_at": created_at, "updated_at": now, "metadata": json.dumps(metadata) if metadata else "{}", }backend/app/services/k8s_worker/worker.py (1)
283-283: Logging level should be DEBUG, not INFO.The log message showing image name sanitization (
before: {image_ref} -> {sanitized_image_ref}) is implementation detail useful for debugging, not operational info. Consider changing to DEBUG level.Suggested change
- self._logger.info(f"DAEMONSET: before: {image_ref} -> {sanitized_image_ref}") + self._logger.debug(f"Image name sanitized: {image_ref} -> {sanitized_image_ref}")backend/app/services/event_bus.py (1)
42-42: Handler type annotation is too loose.Using
objectfor the handler type loses all type safety. Consider using a properCallabletype hint for better IDE support and type checking.Suggested typing improvement
+from collections.abc import Awaitable, Callable +from typing import Union + +EventHandler = Callable[[EventBusEvent], Union[None, Awaitable[None]]] + `@dataclass` class Subscription: """Represents a single event subscription.""" id: str = field(default_factory=lambda: str(uuid4())) pattern: str = "" - handler: object = field(default=None) + handler: EventHandler | None = field(default=None)And update the subscribe signature:
- async def subscribe(self, pattern: str, handler: object) -> str: + async def subscribe(self, pattern: str, handler: EventHandler) -> str:Also applies to: 84-84
backend/app/services/saga/saga_orchestrator.py (1)
69-106: Consider extracting completion events to a class constant to avoid duplication.The
completion_eventsset is defined identically in bothget_trigger_event_types(lines 81-85) andhandle_event(lines 99-103). Extracting this to a class-level constant would prevent potential inconsistencies if one is updated without the other.♻️ Suggested refactor
class SagaOrchestrator: """Stateless saga orchestrator - pure event handler. No lifecycle methods (start/stop) - receives ready-to-use dependencies from DI. All state stored in SagaRepository. Worker entrypoint handles the consume loop. """ + + _COMPLETION_EVENTS: frozenset[EventType] = frozenset({ + EventType.EXECUTION_COMPLETED, + EventType.EXECUTION_FAILED, + EventType.EXECUTION_TIMEOUT, + })Then use
self._COMPLETION_EVENTSin both methods.backend/app/services/coordinator/coordinator.py (1)
279-292: Hardcodedestimated_wait_seconds=None- consider calculating or documenting.The
estimated_wait_secondsfield is set toNone(line 286). If this is intentional for now, consider adding a TODO comment. Otherwise, this could be calculated from queue stats and average execution time to provide useful feedback to users.backend/app/core/providers.py (1)
644-703: Config objects created inline with defaults - consider DI consistency.
K8sWorkerConfig(line 658) andPodMonitorConfig(line 694) are instantiated inline with default values. For consistency with other configuration patterns (e.g.,SagaConfigwhich uses a factory function), consider extracting these to factory functions or providing them via the Settings object if configuration is needed.This is a minor consistency suggestion - the current approach works if defaults are sufficient.
| class RedisRepositoryProvider(Provider): | ||
| """Provides Redis-backed state repositories for stateless services.""" | ||
|
|
||
| scope = Scope.APP | ||
|
|
||
| @provide | ||
| def get_execution_state_repository( | ||
| self, redis_client: redis.Redis, logger: logging.Logger | ||
| ) -> ExecutionStateRepository: | ||
| return ExecutionStateRepository(redis_client, logger) | ||
|
|
||
| @provide | ||
| def get_execution_queue_repository( | ||
| self, redis_client: redis.Redis, logger: logging.Logger, settings: Settings | ||
| ) -> ExecutionQueueRepository: | ||
| return ExecutionQueueRepository( | ||
| redis_client, | ||
| logger, | ||
| max_queue_size=10000, | ||
| max_executions_per_user=100, | ||
| ) | ||
|
|
||
| @provide | ||
| async def get_resource_repository( | ||
| self, redis_client: redis.Redis, logger: logging.Logger, settings: Settings | ||
| ) -> ResourceRepository: | ||
| repo = ResourceRepository( | ||
| redis_client, | ||
| logger, | ||
| total_cpu_cores=32.0, | ||
| total_memory_mb=65536, | ||
| ) | ||
| await repo.initialize() | ||
| return repo | ||
|
|
||
| @provide | ||
| def get_pod_state_repository( | ||
| self, redis_client: redis.Redis, logger: logging.Logger | ||
| ) -> PodStateRepository: | ||
| return PodStateRepository(redis_client, logger) |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Hardcoded resource limits should be configurable via Settings.
The RedisRepositoryProvider has hardcoded values for queue and resource limits:
max_queue_size=10000(line 149)max_executions_per_user=100(line 150)total_cpu_cores=32.0(line 160)total_memory_mb=65536(line 161)
These should be pulled from Settings to allow configuration per environment without code changes.
♻️ Suggested refactor
`@provide`
def get_execution_queue_repository(
self, redis_client: redis.Redis, logger: logging.Logger, settings: Settings
) -> ExecutionQueueRepository:
return ExecutionQueueRepository(
redis_client,
logger,
- max_queue_size=10000,
- max_executions_per_user=100,
+ max_queue_size=settings.QUEUE_MAX_SIZE,
+ max_executions_per_user=settings.QUEUE_MAX_PER_USER,
)
`@provide`
async def get_resource_repository(
self, redis_client: redis.Redis, logger: logging.Logger, settings: Settings
) -> ResourceRepository:
repo = ResourceRepository(
redis_client,
logger,
- total_cpu_cores=32.0,
- total_memory_mb=65536,
+ total_cpu_cores=settings.RESOURCE_TOTAL_CPU_CORES,
+ total_memory_mb=settings.RESOURCE_TOTAL_MEMORY_MB,
)
await repo.initialize()
return repoAdd corresponding fields to your Settings class with sensible defaults.
🤖 Prompt for AI Agents
In `@backend/app/core/providers.py` around lines 131 - 170, The provider currently
hardcodes limits in RedisRepositoryProvider; update
get_execution_queue_repository and get_resource_repository to read
max_queue_size, max_executions_per_user, total_cpu_cores, and total_memory_mb
from the Settings argument instead of hardcoded literals; add those fields with
sensible defaults to the Settings class (e.g., max_queue_size,
max_executions_per_user, total_cpu_cores, total_memory_mb), then pass
settings.max_queue_size and settings.max_executions_per_user into
ExecutionQueueRepository and settings.total_cpu_cores and
settings.total_memory_mb into ResourceRepository and keep existing initialize()
call.
| async def allocate( | ||
| self, | ||
| execution_id: str, | ||
| language: str, | ||
| requested_cpu: float | None = None, | ||
| requested_memory_mb: int | None = None, | ||
| requested_gpu: int = 0, | ||
| ) -> ResourceAllocation | None: | ||
| """Allocate resources for execution. Returns allocation or None if insufficient.""" | ||
| # Check if already allocated | ||
| alloc_key = f"{self.ALLOC_KEY_PREFIX}:{execution_id}" | ||
| existing = await self._redis.hgetall(alloc_key) # type: ignore[misc] | ||
| if existing: | ||
| self._logger.warning(f"Execution {execution_id} already has allocation") | ||
| return ResourceAllocation( | ||
| execution_id=execution_id, | ||
| cpu_cores=float(existing.get(b"cpu", existing.get("cpu", 0))), | ||
| memory_mb=int(existing.get(b"memory", existing.get("memory", 0))), | ||
| gpu_count=int(existing.get(b"gpu", existing.get("gpu", 0))), | ||
| ) | ||
|
|
||
| # Determine requested resources | ||
| if requested_cpu is None or requested_memory_mb is None: | ||
| default_cpu, default_memory = self.DEFAULT_ALLOCATIONS.get(language, (0.5, 512)) | ||
| requested_cpu = requested_cpu or default_cpu | ||
| requested_memory_mb = requested_memory_mb or default_memory | ||
|
|
||
| # Apply limits | ||
| requested_cpu = min(requested_cpu, self._max_cpu_per_exec) | ||
| requested_memory_mb = min(requested_memory_mb, self._max_memory_per_exec) | ||
|
|
||
| # Atomic allocation using Lua script | ||
| lua_script = """ | ||
| local pool_key = KEYS[1] | ||
| local alloc_key = KEYS[2] | ||
| local req_cpu = tonumber(ARGV[1]) | ||
| local req_memory = tonumber(ARGV[2]) | ||
| local req_gpu = tonumber(ARGV[3]) | ||
| local min_cpu = tonumber(ARGV[4]) | ||
| local min_memory = tonumber(ARGV[5]) | ||
|
|
||
| local avail_cpu = tonumber(redis.call('HGET', pool_key, 'available_cpu') or '0') | ||
| local avail_memory = tonumber(redis.call('HGET', pool_key, 'available_memory') or '0') | ||
| local avail_gpu = tonumber(redis.call('HGET', pool_key, 'available_gpu') or '0') | ||
|
|
||
| local cpu_after = avail_cpu - req_cpu | ||
| local memory_after = avail_memory - req_memory | ||
| local gpu_after = avail_gpu - req_gpu | ||
|
|
||
| if cpu_after < min_cpu or memory_after < min_memory or gpu_after < 0 then | ||
| return 0 | ||
| end | ||
|
|
||
| redis.call('HSET', pool_key, 'available_cpu', tostring(cpu_after)) | ||
| redis.call('HSET', pool_key, 'available_memory', tostring(memory_after)) | ||
| redis.call('HSET', pool_key, 'available_gpu', tostring(gpu_after)) | ||
|
|
||
| redis.call('HSET', alloc_key, 'cpu', tostring(req_cpu), 'memory', tostring(req_memory), | ||
| 'gpu', tostring(req_gpu)) | ||
| redis.call('EXPIRE', alloc_key, 7200) | ||
|
|
||
| return 1 | ||
| """ | ||
|
|
||
| result = await self._redis.eval( # type: ignore[misc] | ||
| lua_script, | ||
| 2, | ||
| self.POOL_KEY, | ||
| alloc_key, | ||
| str(requested_cpu), | ||
| str(requested_memory_mb), | ||
| str(requested_gpu), | ||
| str(self._min_reserve_cpu), | ||
| str(self._min_reserve_memory), | ||
| ) | ||
|
|
||
| if not result: |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cat -n backend/app/db/repositories/resource_repository.py | head -250Repository: HardMax71/Integr8sCode
Length of output: 10876
Move the allocation existence check inside the Lua script to prevent concurrent reservation for the same execution_id.
The current implementation checks for an existing allocation on line 128 before executing the Lua script, creating a race condition. Two concurrent callers can both pass the Python check (when no allocation exists yet) and then both execute the Lua script, resulting in duplicate allocations for the same execution_id. The Lua script at lines 149–179 lacks an existence check and only validates resource availability, making the check non-atomic.
Add an EXISTS check at the start of the Lua script and return a distinct code (e.g., 2) when allocation already exists. Then handle that result in Python to fetch and return the existing allocation:
🛠️ Suggested fix (atomic existence check in Lua)
@@
- local req_gpu = tonumber(ARGV[3])
+ local req_gpu = tonumber(ARGV[3])
+ if redis.call('EXISTS', alloc_key) == 1 then
+ return 2
+ end
@@
- if not result:
+ if result == 2:
+ existing = await self._redis.hgetall(alloc_key) # type: ignore[misc]
+ self._logger.warning(f"Execution {execution_id} already has allocation")
+ return ResourceAllocation(
+ execution_id=execution_id,
+ cpu_cores=float(existing.get(b"cpu", existing.get("cpu", 0))),
+ memory_mb=int(existing.get(b"memory", existing.get("memory", 0))),
+ gpu_count=int(existing.get(b"gpu", existing.get("gpu", 0))),
+ )
+ if not result:
pool = await self._redis.hgetall(self.POOL_KEY) # type: ignore[misc]🤖 Prompt for AI Agents
In `@backend/app/db/repositories/resource_repository.py` around lines 117 - 193,
The allocate method currently does a non-atomic Python-side existence check
before running the Lua script; move that existence check into the Lua script by
testing if alloc_key exists (redis.call('EXISTS', alloc_key)) at the top of
lua_script and return a distinct code (e.g., 2) when it does, keep returning 0
for insufficient resources and 1 for successful allocation; then update the
Python allocate method (the result handling after self._redis.eval) to treat
result == 2 by reading the existing allocation from alloc_key and returning a
ResourceAllocation built from those fields (same parsing logic used earlier),
otherwise proceed as before for success/failure.
| redis.call('HSET', alloc_key, 'cpu', tostring(req_cpu), 'memory', tostring(req_memory), | ||
| 'gpu', tostring(req_gpu)) | ||
| redis.call('EXPIRE', alloc_key, 7200) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, let's look at the file and context around the problematic lines
wc -l backend/app/db/repositories/resource_repository.pyRepository: HardMax71/Integr8sCode
Length of output: 120
🏁 Script executed:
# Read the file to understand the context
cat -n backend/app/db/repositories/resource_repository.py | sed -n '160,190p'Repository: HardMax71/Integr8sCode
Length of output: 1416
🏁 Script executed:
# Let's also check the broader scope to understand the allocation mechanism
cat -n backend/app/db/repositories/resource_repository.py | head -50Repository: HardMax71/Integr8sCode
Length of output: 1459
🏁 Script executed:
# Look for release or deallocate methods in the repository
rg -n "def.*release|def.*deallocate|def.*free|def.*cleanup" backend/app/db/repositories/resource_repository.pyRepository: HardMax71/Integr8sCode
Length of output: 125
🏁 Script executed:
# Search for where alloc_key might be deleted or cleaned up
rg -n "DEL|cleanup|reclaim|sweeper|reconcil" backend/app/db/repositories/resource_repository.pyRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Let's read more of the file to see all methods
cat -n backend/app/db/repositories/resource_repository.py | tail -150Repository: HardMax71/Integr8sCode
Length of output: 7171
🏁 Script executed:
# Search for any background jobs or tasks that might handle cleanup
fd -t f "\.py$" backend/app | xargs rg -l "sweeper|reconcil|cleanup.*alloc|expire.*alloc" 2>/dev/null | head -20Repository: HardMax71/Integr8sCode
Length of output: 191
🏁 Script executed:
# Let's check the pod_monitor files for reconciliation logic
cat -n backend/app/services/pod_monitor/monitor.py | head -100Repository: HardMax71/Integr8sCode
Length of output: 3583
🏁 Script executed:
# Search for any cleanup or reconciliation logic related to allocations
rg -n "reclaim|cleanup|orphan|leaked|expired" backend/app/services/pod_monitor/ -A 3 -B 1Repository: HardMax71/Integr8sCode
Length of output: 459
🏁 Script executed:
# Check the entire backend for sweeper/reconciler patterns
rg -l "sweeper|reconciler|cleanup.*alloc|scan.*alloc" backend/appRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Let's see how release() is called in the codebase
rg -n "\.release\(" backend/app -B 2 -A 2 | head -60Repository: HardMax71/Integr8sCode
Length of output: 1650
🏁 Script executed:
# Let's check the coordinator more closely to understand the full lifecycle
cat -n backend/app/services/coordinator/coordinator.py | sed -n '100,160p'Repository: HardMax71/Integr8sCode
Length of output: 2952
🏁 Script executed:
# Search for any background tasks or jobs that might handle cleanup
rg -l "task|schedule|background|cron|timer|interval" backend/app --type py | xargs rg -l "alloc" 2>/dev/nullRepository: HardMax71/Integr8sCode
Length of output: 268
🏁 Script executed:
# Check if there's any handler for expired allocations or reconciliation
rg -n "scan_iter.*ALLOC|expired|orphan" backend/app -A 3 -B 1Repository: HardMax71/Integr8sCode
Length of output: 7490
🏁 Script executed:
# Look for any initialization or startup logic that might launch cleanup jobs
rg -n "startup|shutdown|lifespan|background.*task" backend/app/main.py -A 5 -B 2Repository: HardMax71/Integr8sCode
Length of output: 618
🏁 Script executed:
# Let's check the lifespan for background tasks
cat -n backend/app/core/dishka_lifespan.pyRepository: HardMax71/Integr8sCode
Length of output: 4452
🏁 Script executed:
# Look for any result processor or background job that might handle this
cat -n backend/app/services/result_processor/resource_cleaner.py | head -80Repository: HardMax71/Integr8sCode
Length of output: 3516
🏁 Script executed:
# Search the entire codebase for any background task that might reconcile Redis allocations
rg -n "background_task|create_task|gather|Task\(" backend/app --type py | grep -i "alloc\|resource" -A 2 -B 2Repository: HardMax71/Integr8sCode
Length of output: 507
🏁 Script executed:
# Check if there's any retry/recovery for execution events
rg -n "ExecutionCompleted|ExecutionFailed|ExecutionCancelled" backend/app --type py -B 2 -A 2 | head -80Repository: HardMax71/Integr8sCode
Length of output: 6401
🏁 Script executed:
# Check if there's dead letter queue or event replay for failed events
rg -n "dead.*letter|dlq|replay|retry.*event" backend/app --type py | head -30Repository: HardMax71/Integr8sCode
Length of output: 3092
🏁 Script executed:
# Let's verify that pod monitor is the ONLY source of these events
rg -n "ExecutionCompletedEvent\|ExecutionFailedEvent\|ExecutionCancelledEvent" backend/app --type py -lRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Check if there's any monitoring/alerting for orphaned allocations
rg -n "allocation.*orphan\|orphan.*alloc\|leaked\|capacity.*leak" backend/app --type pyRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Check pod_monitor more carefully - it must be creating the events
rg -n "ExecutionCompletedEvent\|ExecutionFailedEvent" backend/app/services/pod_monitor/ --type pyRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Verify the complete flow - from pod monitor to coordinator
rg -n "class.*Event\|@dataclass.*Event" backend/app/domain/events/typed.py -A 3 | head -40Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Check if there's any timeout-based reconciliation
rg -n "timeout\|ttl\|7200" backend/app/db/repositories/resource_repository.py -B 2 -A 2Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Find all files that mention ExecutionCompletedEvent or similar
find backend/app -name "*.py" -type f | xargs grep -l "ExecutionCompletedEvent" 2>/dev/nullRepository: HardMax71/Integr8sCode
Length of output: 331
🏁 Script executed:
# Check the pod_monitor event_mapper more carefully
cat -n backend/app/services/pod_monitor/event_mapper.py | grep -A 10 "def _map_completed"Repository: HardMax71/Integr8sCode
Length of output: 635
🏁 Script executed:
# Look at the complete resource_repository to see ALL methods
cat -n backend/app/db/repositories/resource_repository.py | sed -n '130,180p'Repository: HardMax71/Integr8sCode
Length of output: 2679
Allocation TTL creates permanent pool capacity leak without recovery mechanism.
The 7200-second TTL on alloc_key (line 176) will automatically expire the allocation record, but the decremented pool resources (lines 170-172) are never restored if release() is not called. While normal completions are handled via pod monitor events → coordinator release, any failure path that prevents event publication (pod undetected crash, lost event, coordinator crash before release) will permanently leak capacity.
No sweeper or reconciliation job exists to reclaim expired allocations. Either implement a background task to scan for expired alloc_key:* entries and restore pool resources, or add a TTL to the pool keys themselves with periodic reconciliation.
🤖 Prompt for AI Agents
In `@backend/app/db/repositories/resource_repository.py` around lines 174 - 176,
alloc_key entries are being set with a 7200s TTL but pool resource counters
(updated in the allocation code around alloc_key and the pool keys) are never
restored if release() isn't called, causing permanent capacity leaks; add a
reconciliation background task (e.g., reclaim_expired_allocations or
reconcile_expired_allocations) that periodically scans redis for keys matching
"alloc_key:*", detects expired or missing allocation metadata and atomically
restores the pool counters (cpu, memory, gpu) by incrementing the corresponding
pool hash fields, using Redis transactions or a Lua script to avoid races, and
mark or remove reclaimed alloc_key entries to make reclamation idempotent;
alternatively add TTLs to pool keys and implement a periodic reconcile_pools
task that verifies sums against active allocations and fixes
discrepancies—update resource_repository functions (alloc_key handling and
release()) to cooperate with the new reconciler.
| kafka_consumer = await container.get(AIOKafkaConsumer) | ||
| handler = await container.get(UnifiedConsumer) | ||
|
|
||
| logger.info("ExecutionCoordinator started, consuming events...") | ||
|
|
||
| async for msg in kafka_consumer: | ||
| await handler.handle(msg) | ||
| await kafka_consumer.commit() | ||
|
|
||
| logger.info("ExecutionCoordinator shutdown complete") | ||
|
|
||
| await container.close() |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
find . -type f -name "run_coordinator.py" | head -5Repository: HardMax71/Integr8sCode
Length of output: 102
🏁 Script executed:
cat -n backend/workers/run_coordinator.pyRepository: HardMax71/Integr8sCode
Length of output: 3053
🏁 Script executed:
find . -type f -name "container.py" | grep -i coreRepository: HardMax71/Integr8sCode
Length of output: 97
🏁 Script executed:
find . -type f -name "*container*" | head -10Repository: HardMax71/Integr8sCode
Length of output: 140
🏁 Script executed:
rg "class.*container|def create_coordinator_container" -t py | head -20Repository: HardMax71/Integr8sCode
Length of output: 167
🏁 Script executed:
cat -n backend/app/core/container.py | head -150Repository: HardMax71/Integr8sCode
Length of output: 5288
🏁 Script executed:
rg "create_coordinator_container" -A 30 backend/app/core/container.pyRepository: HardMax71/Integr8sCode
Length of output: 1035
🏁 Script executed:
rg "class.*Provider|AIOKafkaConsumer" backend/app/core/providers.py -A 5 | head -100Repository: HardMax71/Integr8sCode
Length of output: 2866
🏁 Script executed:
find . -type f -name "providers.py" | xargs wc -l | sort -nRepository: HardMax71/Integr8sCode
Length of output: 101
🏁 Script executed:
rg "CoordinatorProvider|class.*Provider" backend/app/core/providers.py | grep -A 15 "CoordinatorProvider"Repository: HardMax71/Integr8sCode
Length of output: 294
🏁 Script executed:
rg "AIOKafkaConsumer" backend/app/core/providers.py -B 5 -A 15Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
rg "class CoordinatorProvider" backend/app/core/providers.py -A 30Repository: HardMax71/Integr8sCode
Length of output: 1043
🏁 Script executed:
rg "KafkaServicesProvider" backend/app/core/providers.py -A 30Repository: HardMax71/Integr8sCode
Length of output: 957
🏁 Script executed:
rg "AIOKafkaConsumer" backend/app/core/providers.py -B 10 -A 10Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
rg "AIOKafkaConsumer" backend/ --type pyRepository: HardMax71/Integr8sCode
Length of output: 2707
🏁 Script executed:
rg "provide.*AIOKafkaConsumer|async def.*kafka.*consumer" backend/app/core/providers.py -B 5 -A 20Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
rg "MessagingProvider\|EventProvider" backend/app/core/providers.py -A 50 | head -150Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
rg "def.*kafka" backend/app/core/providers.py -B 3 -A 15Repository: HardMax71/Integr8sCode
Length of output: 1852
🏁 Script executed:
sed -n '1,818p' backend/app/core/providers.py | grep -n "kafka\|Kafka" -iRepository: HardMax71/Integr8sCode
Length of output: 2108
🏁 Script executed:
rg "class EventProvider" backend/app/core/providers.py -A 100 | head -120Repository: HardMax71/Integr8sCode
Length of output: 3239
🏁 Script executed:
rg "UnifiedConsumer\|AIOKafkaConsumer" backend/app/core/providers.py -B 5 -A 15Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
rg "class MessagingProvider" backend/app/core/providers.py -A 150 | head -180Repository: HardMax71/Integr8sCode
Length of output: 4874
🏁 Script executed:
rg "def.*consumer|async def.*consumer" backend/app/core/providers.py -B 3 -A 15Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
rg "AIOKafkaConsumer\|UnifiedConsumer" backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
rg "provide.*Consumer|@provide" backend/app/core/providers.py | grep -i consumerRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
find . -type f -name "*.py" -path "*/app/core/*" | xargs grep -l "AIOKafkaConsumer\|provide.*consumer" -iRepository: HardMax71/Integr8sCode
Length of output: 50
🏁 Script executed:
rg "UnifiedConsumer" backend/ -lRepository: HardMax71/Integr8sCode
Length of output: 618
🏁 Script executed:
rg "def.*UnifiedConsumer\|class UnifiedConsumer" backend/ -B 2 -A 15Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
cat -n backend/app/events/core/consumer.py | head -200Repository: HardMax71/Integr8sCode
Length of output: 3047
🏁 Script executed:
rg "provide" backend/app/events/core/consumer.pyRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
rg "@provide" backend/app/core/providers.py | grep -i "consumer\|unified"Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
rg "UnifiedConsumer" backend/app/core/providers.py -B 5 -A 5Repository: HardMax71/Integr8sCode
Length of output: 48
Wrap the consumer loop in try/finally to ensure cleanup on exceptions.
Any exception in the async for loop (lines 42-44) skips await container.close(), leaving the Kafka consumer and other resources open. The codebase establishes a clear pattern—resources like AIOKafkaProducer, K8sClients, and IdempotencyManager all use try/finally for cleanup. Apply the same pattern here:
Suggested fix
- async for msg in kafka_consumer:
- await handler.handle(msg)
- await kafka_consumer.commit()
-
- logger.info("ExecutionCoordinator shutdown complete")
-
- await container.close()
+ try:
+ async for msg in kafka_consumer:
+ await handler.handle(msg)
+ await kafka_consumer.commit()
+ finally:
+ await kafka_consumer.stop()
+ await container.close()
+ logger.info("ExecutionCoordinator shutdown complete")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| kafka_consumer = await container.get(AIOKafkaConsumer) | |
| handler = await container.get(UnifiedConsumer) | |
| logger.info("ExecutionCoordinator started, consuming events...") | |
| async for msg in kafka_consumer: | |
| await handler.handle(msg) | |
| await kafka_consumer.commit() | |
| logger.info("ExecutionCoordinator shutdown complete") | |
| await container.close() | |
| kafka_consumer = await container.get(AIOKafkaConsumer) | |
| handler = await container.get(UnifiedConsumer) | |
| logger.info("ExecutionCoordinator started, consuming events...") | |
| try: | |
| async for msg in kafka_consumer: | |
| await handler.handle(msg) | |
| await kafka_consumer.commit() | |
| finally: | |
| await kafka_consumer.stop() | |
| await container.close() | |
| logger.info("ExecutionCoordinator shutdown complete") |
🤖 Prompt for AI Agents
In `@backend/workers/run_coordinator.py` around lines 37 - 48, The async consumer
loop in ExecutionCoordinator using kafka_consumer and handler can raise
exceptions which skip cleanup; wrap the async for msg in kafka_consumer loop in
a try/finally so that await container.close() (and any kafka_consumer
commit/close if needed) always runs, move the "ExecutionCoordinator shutdown
complete" log into the finally block, and ensure resource-specific cleanup for
AIOKafkaConsumer and UnifiedConsumer happens before closing the container
(referencing kafka_consumer, handler, container, AIOKafkaConsumer,
UnifiedConsumer).
| # Service is HTTP-driven, wait for external shutdown | ||
| await asyncio.Event().wait() | ||
|
|
||
| await asyncio.Event().wait() | ||
| await container.close() |
There was a problem hiding this comment.
Unreachable container.close() - missing signal handling.
asyncio.Event().wait() blocks forever since the event is never set. Line 35 await container.close() will never execute, causing potential resource leaks when the process terminates.
Consider adding signal handling for graceful shutdown:
🔧 Proposed fix with signal handling
+import signal
+
async def run_replay_service(settings: Settings) -> None:
"""Run the event replay service."""
container = create_event_replay_container(settings)
-
logger = await container.get(logging.Logger)
logger.info("Starting EventReplayService with DI container...")
db = await container.get(Database)
await init_beanie(database=db, document_models=ALL_DOCUMENTS)
logger.info("Event replay service initialized and ready")
- # Service is HTTP-driven, wait for external shutdown
- await asyncio.Event().wait()
-
- await container.close()
+ # Wait for shutdown signal
+ shutdown_event = asyncio.Event()
+ loop = asyncio.get_running_loop()
+ for sig in (signal.SIGTERM, signal.SIGINT):
+ loop.add_signal_handler(sig, shutdown_event.set)
+
+ await shutdown_event.wait()
+ logger.info("Shutdown signal received, closing container...")
+ await container.close()🤖 Prompt for AI Agents
In `@backend/workers/run_event_replay.py` around lines 32 - 35, The
forever-blocking await asyncio.Event().wait() prevents container.close() from
ever running; replace the anonymous Event with a named asyncio.Event (e.g.,
shutdown_event), register signal handlers for SIGINT/SIGTERM that set
shutdown_event (using loop.add_signal_handler or signal.signal fallback), await
shutdown_event.wait() in run_event_replay.py, then call await container.close()
after the wait so the container is closed during graceful shutdown; reference
the existing asyncio.Event().wait() and container.close() calls when making the
change.
| kafka_consumer = await container.get(AIOKafkaConsumer) | ||
| handler = await container.get(UnifiedConsumer) | ||
|
|
||
| logger.info("KubernetesWorker started, consuming events...") | ||
|
|
||
| async for msg in kafka_consumer: | ||
| await handler.handle(msg) | ||
| await kafka_consumer.commit() | ||
|
|
||
| logger.info("KubernetesWorker shutdown complete") | ||
|
|
||
| await container.close() |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, let's look at the actual file around the specified lines
cd /tmp && find . -name "run_k8s_worker.py" -type f 2>/dev/nullRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Search for the file in the repository
git ls-files | grep -i "run_k8s_worker"Repository: HardMax71/Integr8sCode
Length of output: 99
🏁 Script executed:
# Look for worker-related files
git ls-files | grep -i worker | head -20Repository: HardMax71/Integr8sCode
Length of output: 856
🏁 Script executed:
# List backend directory structure
git ls-files | grep "^backend/" | head -30Repository: HardMax71/Integr8sCode
Length of output: 999
🏁 Script executed:
# Read the actual file
cat -n backend/workers/run_k8s_worker.pyRepository: HardMax71/Integr8sCode
Length of output: 3010
🏁 Script executed:
# Check AIOKafkaConsumer API and usage patterns in codebase
rg "AIOKafkaConsumer" -A 5 -B 5 | head -100Repository: HardMax71/Integr8sCode
Length of output: 6496
🏁 Script executed:
# Check container.close() implementation
rg "def close" backend/app/core/container.py -A 15Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Look at other worker files to see if they handle cleanup similarly
ls -la backend/workers/run_*.pyRepository: HardMax71/Integr8sCode
Length of output: 580
🏁 Script executed:
# Check one of the other worker files for comparison
cat -n backend/workers/run_result_processor.pyRepository: HardMax71/Integr8sCode
Length of output: 3054
🏁 Script executed:
# Find the container.close() implementation
find . -name "container.py" -type f | xargs cat -n | head -200Repository: HardMax71/Integr8sCode
Length of output: 6533
🏁 Script executed:
# Search for container close implementation more specifically
rg "async def close" backend/app/core/ -A 20Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Check how container manages Kafka consumer lifecycle
rg "AIOKafkaConsumer" backend/app/core/container.py -B 5 -A 10Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Find providers that register Kafka consumer
rg "K8sWorkerProvider|KafkaServicesProvider|EventProvider" backend/app/core/providers.py -A 20 | head -150Repository: HardMax71/Integr8sCode
Length of output: 2219
🏁 Script executed:
# Search for how AIOKafkaConsumer is provided/registered
rg "AIOKafkaConsumer" backend/app/core/providers.py -B 5 -A 15Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Look at all providers
ls backend/app/core/Repository: HardMax71/Integr8sCode
Length of output: 275
🏁 Script executed:
# Search for AIOKafkaConsumer registration in providers
rg "AIOKafkaConsumer" backend/app/core/providers.py -B 10 -A 10Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Look at the full providers.py file to understand Kafka consumer registration
wc -l backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 99
🏁 Script executed:
# Read the providers file to find how Kafka consumer is set up
head -300 backend/app/core/providers.py | tail -100Repository: HardMax71/Integr8sCode
Length of output: 3289
🏁 Script executed:
# Search for provide decorator and Kafka consumer
rg "@provide" backend/app/core/providers.py | grep -i kafkaRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Search more broadly for get_aiokafka_consumer or consumer registration
rg "get_aiokafka_consumer|AIOKafkaConsumer" backend/app/core/providers.py -B 5 -A 15Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Read the entire providers.py to find consumer registration
cat backend/app/core/providers.py | grep -A 20 -B 5 "AIOKafkaConsumer"Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Search for UnifiedConsumer which is what gets the kafka_consumer
rg "UnifiedConsumer" backend/app/core/providers.py -B 5 -A 15Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Check where AIOKafkaConsumer is imported from and how it's used
rg "from.*AIOKafkaConsumer|import.*AIOKafkaConsumer" backend/app/ -B 2 -A 8Repository: HardMax71/Integr8sCode
Length of output: 902
🏁 Script executed:
# Search for where UnifiedConsumer is defined and how it gets the kafka_consumer
rg "class UnifiedConsumer" backend/app/ -A 30Repository: HardMax71/Integr8sCode
Length of output: 2307
🏁 Script executed:
# Look for how kafka_consumer is provided - check EventProvider more carefully
cat backend/app/core/providers.py | grep -A 50 "class EventProvider"Repository: HardMax71/Integr8sCode
Length of output: 1662
🏁 Script executed:
# Search for async generator patterns with kafka
rg "AsyncIterator\[AIOKafkaConsumer\]" backend/app/ -B 10 -A 10Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Search all of providers.py for `@provide` methods that mention consumer
cat backend/app/core/providers.py | grep -B 2 -A 20 "@provide" | grep -B 2 -A 20 -i "consumer"Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Look for where workers get their kafka_consumer from - check the actual container.get call
rg "container.get.*AIOKafkaConsumer" backend/workers/ -B 5 -A 5Repository: HardMax71/Integr8sCode
Length of output: 4104
🏁 Script executed:
# Check if kafka consumer is provided as part of EventProvider or elsewhere
grep -n "kafka_consumer\|AIOKafkaConsumer" backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Let's search the entire providers.py file line by line for kafka
cat backend/app/core/providers.py | grep -i kafkaRepository: HardMax71/Integr8sCode
Length of output: 1948
🏁 Script executed:
# Check if there are other provider files that might define it
find backend/app -name "*provider*" -type fRepository: HardMax71/Integr8sCode
Length of output: 95
🏁 Script executed:
# Search entire app for where AIOKafkaConsumer is provided/created
rg "get_aiokafka_consumer|AsyncIterator.*Kafka" backend/app/ -B 5 -A 15Repository: HardMax71/Integr8sCode
Length of output: 1340
🏁 Script executed:
# Let's check the dishka_lifespan.py which might have setup
cat backend/app/core/dishka_lifespan.pyRepository: HardMax71/Integr8sCode
Length of output: 3773
🏁 Script executed:
# Search for where AIOKafkaConsumer is provided as a dependency with proper async cleanup
rg "get_.*consumer|AIOKafkaConsumer" backend/app/core/providers.py -B 8 -A 15Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Let's look at the entire EventProvider class in providers.py
sed -n '/class EventProvider/,/^class /p' backend/app/core/providers.py | head -150Repository: HardMax71/Integr8sCode
Length of output: 1071
🏁 Script executed:
# Check if there's a consumer provider we missed
grep -n "class.*Provider" backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 1016
🏁 Script executed:
# Check CoordinatorProvider for consumer setup
sed -n '/^class CoordinatorProvider/,/^class /p' backend/app/core/providers.py | head -50Repository: HardMax71/Integr8sCode
Length of output: 1007
🏁 Script executed:
# Check K8sWorkerProvider for consumer setup
sed -n '/^class K8sWorkerProvider/,/^class /p' backend/app/core/providers.py | head -50Repository: HardMax71/Integr8sCode
Length of output: 1056
🏁 Script executed:
# Search across all providers for any method that returns AIOKafkaConsumer
rg "@provide" backend/app/core/providers.py -A 20 | grep -B 5 -A 15 "AIOKafkaConsumer"Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Search the entire providers file for anything related to "consumer"
grep -n "consumer\|Consumer" backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Let's check the EventProvider more thoroughly - maybe it provides the consumer
sed -n '283,313p' backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 1035
🏁 Script executed:
# Search for UnifiedConsumer provider
grep -n "UnifiedConsumer" backend/app/core/providers.py -B 5 -A 10Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Maybe consumer is created somewhere else - search all event files
rg "get_aiokafka_consumer|@provide.*consumer" backend/app/events/ -B 3 -A 15Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Let's check if there's a consumer factory or if it's in __init__ files
rg "AIOKafkaConsumer" backend/app/ --type py -B 3 -A 3 | head -200Repository: HardMax71/Integr8sCode
Length of output: 1775
🏁 Script executed:
# Check if EventProvider does more than what we saw
cat backend/app/core/providers.py | sed -n '283,312p'Repository: HardMax71/Integr8sCode
Length of output: 1034
🏁 Script executed:
# Search for any file that creates or provides AIOKafkaConsumer
find backend/app -name "*.py" -exec grep -l "AIOKafkaConsumer" {} \;Repository: HardMax71/Integr8sCode
Length of output: 146
🏁 Script executed:
# Check startup.py which might initialize the consumer
cat backend/app/core/startup.pyRepository: HardMax71/Integr8sCode
Length of output: 1589
🏁 Script executed:
# Look at the full consumer.py to understand how it's meant to be used
cat backend/app/events/core/consumer.pyRepository: HardMax71/Integr8sCode
Length of output: 2564
🏁 Script executed:
# Check if there's a consumer factory
rg "def.*consumer|class.*Consumer" backend/app/events/core/ -B 2 -A 15Repository: HardMax71/Integr8sCode
Length of output: 5289
🏁 Script executed:
# Search for UnifiedConsumer provider - it must be provided somewhere
rg "@provide.*UnifiedConsumer|def.*unified_consumer" backend/app/core/providers.py -B 5 -A 20Repository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
# Let's check the entire EventReplayProvider to see if it provides consumer
sed -n '800,830p' backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 637
🏁 Script executed:
# Search for all `@provide` methods in EventProvider
sed -n '283,313p' backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 1035
🏁 Script executed:
# Maybe there's a consumer factory in the __init__ of events
cat backend/app/events/__init__.pyRepository: HardMax71/Integr8sCode
Length of output: 48
Wrap the consumer loop in try/finally to guarantee cleanup on exceptions.
An exception in handle() or commit() will skip await container.close(), leaving resources open. The async for loop (lines 41-43) should be wrapped to ensure cleanup always executes.
🔧 Suggested fix
- async for msg in kafka_consumer:
- await handler.handle(msg)
- await kafka_consumer.commit()
-
- logger.info("KubernetesWorker shutdown complete")
-
- await container.close()
+ try:
+ async for msg in kafka_consumer:
+ await handler.handle(msg)
+ await kafka_consumer.commit()
+ finally:
+ await kafka_consumer.stop()
+ await container.close()
+ logger.info("KubernetesWorker shutdown complete")This pattern affects all worker files (run_result_processor.py, run_saga_orchestrator.py, run_coordinator.py, run_pod_monitor.py).
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| kafka_consumer = await container.get(AIOKafkaConsumer) | |
| handler = await container.get(UnifiedConsumer) | |
| logger.info("KubernetesWorker started, consuming events...") | |
| async for msg in kafka_consumer: | |
| await handler.handle(msg) | |
| await kafka_consumer.commit() | |
| logger.info("KubernetesWorker shutdown complete") | |
| await container.close() | |
| kafka_consumer = await container.get(AIOKafkaConsumer) | |
| handler = await container.get(UnifiedConsumer) | |
| logger.info("KubernetesWorker started, consuming events...") | |
| try: | |
| async for msg in kafka_consumer: | |
| await handler.handle(msg) | |
| await kafka_consumer.commit() | |
| finally: | |
| await kafka_consumer.stop() | |
| await container.close() | |
| logger.info("KubernetesWorker shutdown complete") |
🤖 Prompt for AI Agents
In `@backend/workers/run_k8s_worker.py` around lines 36 - 47, The consumer loop
can raise in handler.handle() or kafka_consumer.commit() and skip cleanup; wrap
the async for msg in kafka_consumer loop in a try/finally so that await
container.close() (and any other shutdown steps) always run. Specifically,
enclose the async for ...: await handler.handle(msg); await
kafka_consumer.commit() inside try: ... finally: await container.close() and
keep the logger.info("KubernetesWorker shutdown complete") inside the finally;
apply the same pattern to the other worker modules (run_result_processor.py,
run_saga_orchestrator.py, run_coordinator.py, run_pod_monitor.py) referencing
kafka_consumer, UnifiedConsumer.handle, kafka_consumer.commit, and
container.close where present.
| kafka_consumer = await container.get(AIOKafkaConsumer) | ||
| handler = await container.get(UnifiedConsumer) | ||
|
|
||
| logger.info("ResultProcessor started, consuming events...") | ||
|
|
||
| async for msg in kafka_consumer: | ||
| await handler.handle(msg) | ||
| await kafka_consumer.commit() | ||
|
|
||
| logger.info("ResultProcessor shutdown complete") | ||
|
|
||
| await container.close() |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
find . -name "run_result_processor.py" -type fRepository: HardMax71/Integr8sCode
Length of output: 107
🏁 Script executed:
cat -n ./backend/workers/run_result_processor.pyRepository: HardMax71/Integr8sCode
Length of output: 3054
🏁 Script executed:
# Check the container implementation to understand how it manages lifecycle
find . -name "container.py" -path "*/core/*" -type fRepository: HardMax71/Integr8sCode
Length of output: 97
🏁 Script executed:
cat -n ./backend/app/core/container.pyRepository: HardMax71/Integr8sCode
Length of output: 6533
🏁 Script executed:
# Check the MessagingProvider to see how AIOKafkaConsumer is configured
find . -path "*/core/providers.py" -o -path "*/core/providers/*" -type f | head -20Repository: HardMax71/Integr8sCode
Length of output: 97
🏁 Script executed:
cat -n ./backend/app/core/providers.py | head -150Repository: HardMax71/Integr8sCode
Length of output: 7420
🏁 Script executed:
grep -n "AIOKafkaConsumer" ./backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
wc -l ./backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 101
🏁 Script executed:
grep -n "class.*Provider" ./backend/app/core/providers.py | grep -i "messaging\|kafka"Repository: HardMax71/Integr8sCode
Length of output: 182
🏁 Script executed:
sed -n '224,270p' ./backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 1536
🏁 Script executed:
sed -n '201,223p' ./backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 778
🏁 Script executed:
grep -n "AIOKafkaConsumer\|get.*consumer" ./backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 48
🏁 Script executed:
rg "AIOKafkaConsumer" --type py -A 5 -B 5Repository: HardMax71/Integr8sCode
Length of output: 21734
🏁 Script executed:
sed -n '280,400p' ./backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 3802
🏁 Script executed:
grep -n "provide.*consumer\|provide.*kafka" ./backend/app/core/providers.py -iRepository: HardMax71/Integr8sCode
Length of output: 227
🏁 Script executed:
sed -n '508,650p' ./backend/app/core/providers.pyRepository: HardMax71/Integr8sCode
Length of output: 4603
Wrap the consumer loop in try/finally to guarantee cleanup on failure and cancellation.
The consumer loop lacks exception/cancellation handling. If handle() raises or the consumer task is cancelled (e.g., via SIGTERM), the async for loop exits abnormally and await container.close() is never executed, leaving the consumer open. This pattern affects all worker files (run_pod_monitor.py, run_saga_orchestrator.py, run_k8s_worker.py, run_coordinator.py).
The producer already demonstrates correct cleanup with explicit stop() in a finally block. Apply the same pattern to the consumer:
🔧 Suggested fix
- async for msg in kafka_consumer:
- await handler.handle(msg)
- await kafka_consumer.commit()
-
- logger.info("ResultProcessor shutdown complete")
-
- await container.close()
+ try:
+ async for msg in kafka_consumer:
+ await handler.handle(msg)
+ await kafka_consumer.commit()
+ finally:
+ await kafka_consumer.stop()
+ await container.close()
+ logger.info("ResultProcessor shutdown complete")🤖 Prompt for AI Agents
In `@backend/workers/run_result_processor.py` around lines 36 - 47, Wrap the
consumer loop in a try/finally so cleanup always runs: start by acquiring
kafka_consumer (AIOKafkaConsumer) and handler (UnifiedConsumer) as before, then
run the async for msg in kafka_consumer / await handler.handle(msg) loop inside
a try block; in the finally block ensure you stop/close the kafka_consumer (call
kafka_consumer.stop() or equivalent) and then await container.close(); re-raise
cancellation errors if needed so signal handling still works. This guarantees
the consumer and container are closed even if handler.handle() raises or the
task is cancelled.
| async for msg in kafka_consumer: | ||
| await handler.handle(msg) | ||
| await kafka_consumer.commit() | ||
|
|
||
| try: | ||
| # Wait for shutdown signal or service to stop | ||
| while orchestrator.is_running and not shutdown_event.is_set(): | ||
| await asyncio.sleep(1) | ||
| finally: | ||
| # Container cleanup stops everything | ||
| logger.info("Initiating graceful shutdown...") | ||
| await container.close() | ||
| logger.info("SagaOrchestrator shutdown complete") | ||
|
|
||
| logger.warning("Saga orchestrator stopped") | ||
| await container.close() |
There was a problem hiding this comment.
Missing graceful shutdown handling and error resilience.
Several issues with the main loop:
-
No graceful shutdown: Signal handling (SIGTERM/SIGINT) was removed. In containerized environments, workers need to handle shutdown signals to stop consuming and exit cleanly.
-
No error handling: If
handler.handle(msg)raises an exception (whichUnifiedConsumer.handledoes after logging), the worker crashes. A single poison message could take down the worker. -
Container cleanup not guaranteed:
container.close()won't be called if an exception occurs in the loop.
Proposed fix with shutdown handling and error resilience
+import signal
+from contextlib import suppress
+
async def run_saga_orchestrator(settings: Settings) -> None:
"""Run the saga orchestrator service."""
container = create_saga_orchestrator_container(settings)
logger = await container.get(logging.Logger)
logger.info("Starting SagaOrchestrator with DI container...")
db = await container.get(Database)
await init_beanie(database=db, document_models=ALL_DOCUMENTS)
schema_registry = await container.get(SchemaRegistryManager)
await initialize_event_schemas(schema_registry)
kafka_consumer = await container.get(AIOKafkaConsumer)
handler = await container.get(UnifiedConsumer)
+ shutdown_event = asyncio.Event()
+ loop = asyncio.get_running_loop()
+ for sig in (signal.SIGTERM, signal.SIGINT):
+ loop.add_signal_handler(sig, shutdown_event.set)
+
logger.info("SagaOrchestrator started, consuming events...")
- async for msg in kafka_consumer:
- await handler.handle(msg)
- await kafka_consumer.commit()
-
- logger.info("SagaOrchestrator shutdown complete")
-
- await container.close()
+ try:
+ async for msg in kafka_consumer:
+ if shutdown_event.is_set():
+ break
+ try:
+ await handler.handle(msg)
+ await kafka_consumer.commit()
+ except Exception as e:
+ logger.exception(f"Error processing message: {e}")
+ # Message not committed - will be redelivered
+ finally:
+ logger.info("SagaOrchestrator shutting down...")
+ with suppress(Exception):
+ await kafka_consumer.stop()
+ await container.close()
+ logger.info("SagaOrchestrator shutdown complete")🤖 Prompt for AI Agents
In `@backend/workers/run_saga_orchestrator.py` around lines 41 - 47, Add graceful
shutdown and per-message error resilience around the kafka_consumer loop:
install asyncio signal handlers for SIGINT/SIGTERM that set an asyncio.Event
(e.g., shutdown_event) and make the async for msg in kafka_consumer loop respect
that event so it stops cleanly; wrap message processing so each iteration calls
await handler.handle(msg) inside a try/except that logs the exception via
logger.error (including exception details) and continues (optionally still
commit offsets on successful processing via kafka_consumer.commit), and wrap the
whole consumer loop in try/finally to ensure await container.close() and any
kafka_consumer stop/close are always awaited on shutdown. Reference symbols:
kafka_consumer, handler.handle, kafka_consumer.commit, container.close, logger.
There was a problem hiding this comment.
Actionable comments posted: 0
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (4)
backend/app/services/idempotency/middleware.py (2)
197-241: Dead code:async_handleris defined but never used.The
async_handlerfunction is defined but never referenced. The dispatcher path (line 249) registersdispatch_handlerinstead, and the else branch (line 251-252) only logs an error. This entire function is orphaned code.🐛 Suggested fix: remove dead code
- # Create an async handler that processes the message - async def async_handler(message: Any) -> Any: - self.logger.info(f"IDEMPOTENT HANDLER CALLED for {event_type}") - - # Extract event from confluent-kafka Message - if not hasattr(message, "value"): - self.logger.error(f"Received non-Message object for {event_type}: {type(message)}") - return None - - # Debug log to check message details - self.logger.info( - f"Handler for {event_type} - Message type: {type(message)}, " - f"has key: {hasattr(message, 'key')}, " - f"has topic: {hasattr(message, 'topic')}" - ) - - raw_value = message.value - - # Debug the raw value - self.logger.info(f"Raw value extracted: {raw_value[:100] if raw_value else 'None or empty'}") - - # Handle tombstone messages (null value for log compaction) - if raw_value is None: - self.logger.warning(f"Received empty message for {event_type} - tombstone or consumed value") - return None - - # Handle empty messages - if not raw_value: - self.logger.warning(f"Received empty message for {event_type} - empty bytes") - return None - - try: - # Deserialize using schema registry if available - event = await self.consumer._schema_registry.deserialize_event(raw_value, message.topic) - if not event: - self.logger.error(f"Failed to deserialize event for {event_type}") - return None - - # Call the idempotent wrapper directly in async context - await idempotent_wrapper(event) - - self.logger.debug(f"Successfully processed {event_type} event: {event.event_id}") - return None - except Exception as e: - self.logger.error(f"Failed to process message for {event_type}: {e}", exc_info=True) - raise - # Register with the dispatcher if availableIf this code was intended to be used for direct consumer registration (the else branch), it should be wired up there. Otherwise, remove it to avoid confusion.
243-252: Incomplete fallback: else branch logs error but doesn't register the handler.When no dispatcher is available, the code logs an error but doesn't actually register the handler with the consumer. If direct consumer registration is intended as a fallback, the previously defined
async_handler(or equivalent) should be registered here. If direct registration isn't supported, consider raising an exception instead of silently failing.♻️ Suggested fix: raise exception or implement fallback
Option 1 - Raise an exception if dispatcher is required:
else: - # Fallback to direct consumer registration if no dispatcher - self.logger.error(f"No EventDispatcher available for registering idempotent handler for {event_type}") + raise RuntimeError(f"No EventDispatcher available for registering idempotent handler for {event_type}")Option 2 - If direct consumer registration should work, implement it properly (requires understanding of
UnifiedConsumerAPI).backend/app/services/result_processor/processor.py (2)
88-94: Fragile memory limit parsing will fail for non-Mi suffixes.
rstrip("Mi")removes individual characters 'M' and 'i', not the literal string "Mi". While this works for "128Mi", it will:
- Fail with
ValueErrorfor "1Gi" → "1G" (not a valid int)- Give incorrect results for "1G" → "1" (should be 1024 MiB)
- Fail for "256Ki" or other Kubernetes resource quantity formats
Consider a proper parser:
Proposed fix
+def _parse_memory_to_mib(self, memory_str: str) -> float: + """Parse Kubernetes memory quantity to MiB.""" + import re + match = re.match(r'^(\d+(?:\.\d+)?)\s*([KMGTPE]i?)?[bB]?$', memory_str) + if not match: + raise ValueError(f"Invalid memory format: {memory_str}") + value = float(match.group(1)) + unit = match.group(2) or '' + multipliers = { + '': 1 / (1024 * 1024), 'K': 1 / 1024, 'Ki': 1 / 1024, + 'M': 1, 'Mi': 1, 'G': 1024, 'Gi': 1024, + } + return value * multipliers.get(unit, 1)Then use:
- settings_limit = self._settings.K8S_POD_MEMORY_LIMIT - memory_limit_mib = int(settings_limit.rstrip("Mi")) + memory_limit_mib = self._parse_memory_to_mib(self._settings.K8S_POD_MEMORY_LIMIT)
123-132: Exit code 0 incorrectly becomes -1.Line 126:
exit_code=event.exit_code or -1treats0as falsy. If an execution fails withexit_code=0(possible in some edge cases), this will incorrectly record-1.Since
ExecutionFailedEvent.exit_codeis typed asint(notOptional[int]), the fallback may be unnecessary. If it's defensive against potentialNonevalues, use an explicit check:Proposed fix
- exit_code=event.exit_code or -1, + exit_code=event.exit_code if event.exit_code is not None else -1,
🧹 Nitpick comments (4)
backend/app/services/idempotency/middleware.py (2)
5-5: Inconsistent type hint style: mixingDict/Setwithlist.The file uses
DictandSetfromtyping(line 5) but also uses the builtinlistfor generic type hints (lines 141, 157). For consistency, either use all typing imports (List) or all builtins (dict,set) throughout.♻️ Suggested fix for consistency
If targeting Python 3.9+, use builtin generics throughout:
-from typing import Any, Awaitable, Callable, Dict, Set +from typing import Any, Awaitable, CallableThen update:
- fields: Set[str] | None = None, + fields: set[str] | None = None,- self._original_handlers: Dict[EventType, list[Callable[[DomainEvent], Awaitable[None]]]] = {} + self._original_handlers: dict[EventType, list[Callable[[DomainEvent], Awaitable[None]]]] = {}Also applies to: 141-141, 157-157
146-150: Defensive check on non-optional parameter.The guard
if not self.dispatcherat line 148 checks a parameter that is typed as required (dispatcher: EventDispatcher). While defensive coding can be helpful, this creates inconsistency—ifdispatchercan legitimately beNone, the type hint should reflect that (dispatcher: EventDispatcher | None). Otherwise, this guard is misleading.Either update the type hint to be explicit about allowing
None:- dispatcher: EventDispatcher, + dispatcher: EventDispatcher | None,Or trust the type system and remove the guard (relying on type checkers to catch invalid calls).
backend/app/services/result_processor/processor.py (2)
31-43: Unused configuration fields.
ResultProcessorConfigdefinesbatch_size,processing_timeout,topics, andresult_topic, but none of these are accessed anywhere inResultProcessor. The storedself._configis never read after assignment.If these are planned for future use, consider removing them until needed (YAGNI). If they should be used, wire them into the handler logic.
180-196: Consider extracting hardcodedservice_version.
"1.0.0"is duplicated on lines 182 and 195. Extracting this to a constant or config would simplify version bumps and ensure consistency.
|
There was a problem hiding this comment.
3 issues found across 69 files (changes from recent commits).
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/services/coordinator/coordinator.py">
<violation number="1" location="backend/app/services/coordinator/coordinator.py:65">
P2: Avoid decrementing the per-user limit counter on exceptions unless the increment succeeded; otherwise a failed increment can push the Redis count negative and effectively disable the limit.</violation>
</file>
<file name="backend/app/db/repositories/redis/user_limit_repository.py">
<violation number="1" location="backend/app/db/repositories/redis/user_limit_repository.py:17">
P2: Limit enforcement is non-atomic; concurrent increments can exceed max_per_user. Use a Lua script/transaction to increment and validate in one atomic operation.</violation>
</file>
<file name="backend/app/api/routes/execution.py">
<violation number="1" location="backend/app/api/routes/execution.py:82">
P2: Duplicate requests can hit the reservation marker value ("1") and trigger model_validate_json errors, causing a 500 when a request races with the original. Guard against the reservation sentinel or return a 409/202 until a real cached JSON result is available.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
| """Send CreatePodCommandEvent to k8s-worker via SAGA_COMMANDS topic""" | ||
| metadata = await self._build_command_metadata(request) | ||
|
|
||
| await self._user_limit.decrement(user_id) |
There was a problem hiding this comment.
P2: Avoid decrementing the per-user limit counter on exceptions unless the increment succeeded; otherwise a failed increment can push the Redis count negative and effectively disable the limit.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At backend/app/services/coordinator/coordinator.py, line 65:
<comment>Avoid decrementing the per-user limit counter on exceptions unless the increment succeeded; otherwise a failed increment can push the Redis count negative and effectively disable the limit.</comment>
<file context>
@@ -33,228 +20,73 @@
+ self._metrics.record_coordinator_execution_scheduled("scheduled")
except Exception as e:
+ await self._user_limit.decrement(user_id)
self._logger.error(f"Failed to handle execution request {event.execution_id}: {e}", exc_info=True)
self._metrics.record_coordinator_execution_scheduled("error")
</file context>
|
|
||
| async def try_increment(self, user_id: str) -> bool: | ||
| """Increment user count. Returns True if under limit, False if limit exceeded.""" | ||
| count = await self._redis.hincrby(self.KEY, user_id, 1) # type: ignore[misc] |
There was a problem hiding this comment.
P2: Limit enforcement is non-atomic; concurrent increments can exceed max_per_user. Use a Lua script/transaction to increment and validate in one atomic operation.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At backend/app/db/repositories/redis/user_limit_repository.py, line 17:
<comment>Limit enforcement is non-atomic; concurrent increments can exceed max_per_user. Use a Lua script/transaction to increment and validate in one atomic operation.</comment>
<file context>
@@ -0,0 +1,25 @@
+
+ async def try_increment(self, user_id: str) -> bool:
+ """Increment user count. Returns True if under limit, False if limit exceeded."""
+ count = await self._redis.hincrby(self.KEY, user_id, 1) # type: ignore[misc]
+ if count > self._max_per_user:
+ await self._redis.hincrby(self.KEY, user_id, -1) # type: ignore[misc]
</file context>
| if not is_new: | ||
| cached = await idempotency_repo.get_result(idem_key) | ||
| if cached: | ||
| return ExecutionResponse.model_validate_json(cached) |
There was a problem hiding this comment.
P2: Duplicate requests can hit the reservation marker value ("1") and trigger model_validate_json errors, causing a 500 when a request races with the original. Guard against the reservation sentinel or return a 409/202 until a real cached JSON result is available.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At backend/app/api/routes/execution.py, line 82:
<comment>Duplicate requests can hit the reservation marker value ("1") and trigger model_validate_json errors, causing a 500 when a request races with the original. Guard against the reservation sentinel or return a 409/202 until a real cached JSON result is available.</comment>
<file context>
@@ -74,33 +72,14 @@ async def create_execution(
+ if not is_new:
+ cached = await idempotency_repo.get_result(idem_key)
+ if cached:
+ return ExecutionResponse.model_validate_json(cached)
try:
</file context>
There was a problem hiding this comment.
Actionable comments posted: 11
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
backend/app/services/execution_service.py (1)
125-146: Update priority docstring for QueuePriority enum.The signature now expects
QueuePriority, but the docstring still describes numeric priority.🔧 Suggested doc update
- priority: Execution priority (1-10, lower is higher priority). + priority: Execution priority (QueuePriority enum, e.g., QueuePriority.NORMAL).backend/app/services/result_processor/processor.py (1)
126-126: Bug:exit_code or -1returns-1whenexit_codeis0.The expression
event.exit_code or -1evaluates to-1whenexit_codeis0(a valid, successful exit code). This masks legitimate zero exit codes for failed executions.🐛 Proposed fix
- exit_code=event.exit_code or -1, + exit_code=event.exit_code if event.exit_code is not None else -1,backend/app/events/core/producer.py (1)
44-76: Missing error handling for unmapped event types.If
event_to_produce.event_typeis not present inEVENT_TYPE_TO_TOPIC, line 48 will raise aKeyError. Consider adding defensive handling or a clear error message.🛠️ Suggested fix
async def produce( self, event_to_produce: DomainEvent, key: str | None = None, headers: dict[str, str] | None = None ) -> None: """Produce a message to Kafka.""" + topic_name = EVENT_TYPE_TO_TOPIC.get(event_to_produce.event_type) + if topic_name is None: + raise ValueError(f"No topic mapping for event type: {event_to_produce.event_type}") - topic = f"{self._topic_prefix}{EVENT_TYPE_TO_TOPIC[event_to_produce.event_type]}" + topic = f"{self._topic_prefix}{topic_name}"
🤖 Fix all issues with AI agents
In `@backend/app/api/routes/execution.py`:
- Around line 77-82: When idem_key is present and
idempotency_repo.try_reserve(idem_key, ttl=86400) returns False but
idempotency_repo.get_result(idem_key) returns None (meaning the original request
is still processing), avoid re-executing and instead short-circuit with an
appropriate HTTP response (e.g., raise HTTPException(status_code=202) or 409)
indicating the request is in-flight; update the block around idem_key /
try_reserve / get_result to check for cached is None and return a 202 Accepted
(or 409 Conflict) with a clear body or headers rather than falling through to
re-execution so idempotency is preserved.
In `@backend/app/core/providers.py`:
- Around line 207-222: The factory function get_unified_producer currently
constructs UnifiedProducer without passing the topic_prefix setting; update
get_unified_producer to read the configured prefix (e.g., from
settings.KAFKA_TOPIC_PREFIX or the app Settings instance) and pass it into the
UnifiedProducer(...) call via the topic_prefix parameter so created topics
include the configured prefix; ensure you reference get_unified_producer and
UnifiedProducer when making the change and use the existing settings symbol
(KAFKA_TOPIC_PREFIX) used elsewhere in the codebase.
In `@backend/app/db/repositories/redis/idempotency_repository.py`:
- Around line 32-36: In get_result, the code converts Redis return values with
str(result) which yields "b'...'" for bytes; change get_result (referencing
get_result, KEY_PREFIX and _redis.get) to detect bytes and decode using
result.decode() (or return result directly if already a str) so the function
returns a proper string or None suitable for JSON parsing.
- Around line 21-25: try_reserve currently sets the same KEY_PREFIX key to the
placeholder "1", so callers that see False may read that placeholder via
get_result and fail JSON deserialization; change to a two-key pattern: use a
lock key (e.g., f"{self.KEY_PREFIX}:lock:{key}") set with NX and TTL inside
try_reserve and return True/False based on that, and write actual results to a
separate result key (e.g., f"{self.KEY_PREFIX}:result:{key}") in store_result;
update get_result to read the result key and either return None / a sentinel or
retry/wait if the lock key exists (indicating processing in progress) so callers
never try to deserialize the "in-progress" placeholder. Ensure references to
try_reserve, store_result, get_result, KEY_PREFIX and _default_ttl are updated
accordingly.
In `@backend/app/db/repositories/redis/pod_state_repository.py`:
- Around line 92-101: The update_pod_status function has a TOCTOU race between
the exists check and the hset call; replace the two-step logic in
update_pod_status with an atomic Redis operation (e.g., a Lua EVAL script or a
WATCH/MULTI transaction) so the update only occurs if the key still exists and
return a boolean based on the atomic result; locate update_pod_status in
PodStateRepository (method update_pod_status) and implement an EVAL that checks
redis.call('exists', key) and performs hset inside the script, returning
success/failure, or use WATCH on the key then MULTI/HSET/EXEC and handle EXEC
returning None as failure.
In `@backend/app/dlq/manager.py`:
- Around line 150-158: In _update_message_status, replace using vars(update)
with Pydantic's serialization: call update.model_dump(exclude_unset=True) (or
model_dump() with appropriate exclude flags) to build the updates dict so
internal attrs aren't included; then add updates["last_updated"] =
datetime.now(timezone.utc) and pass that to doc.set as before (target symbols:
_update_message_status and DLQMessageUpdate).
- Around line 79-109: handle_dlq_message currently calls json.loads(raw_message)
and DLQMessage(**data) without catching parsing/validation failures, so
JSONDecodeError or pydantic.ValidationError can bubble up and crash the
consumer; wrap the json.loads and DLQMessage(**data, headers=...) construction
in a try/except that catches json.JSONDecodeError and pydantic.ValidationError
(add "from pydantic import ValidationError" to imports), log the error including
the raw_message and exception via the existing logger, increment/record an
appropriate metric on self._metrics (e.g., a DLQ parse/validation failure
counter or record_dlq_processing_duration with zero/failed tag), and return
early so the worker continues processing subsequent messages instead of raising.
In `@backend/app/domain/enums/kafka.py`:
- Around line 35-37: The RESOURCE_ALLOCATION topic constant is defined but
unused; either remove RESOURCE_ALLOCATION from backend/app/domain/enums/kafka.py
or wire it up: add it to the appropriate consumer subscription set
(CONSUMER_GROUP_SUBSCRIPTIONS) and ensure a producer sends messages to it (or
update the DLQ processor to stop relying on it). Locate the constant
RESOURCE_ALLOCATION, the consumer map CONSUMER_GROUP_SUBSCRIPTIONS, and the DLQ
processor code that references RESOURCE_ALLOCATION, then either delete the
constant and any DLQ references or add a subscription entry and implement/verify
producer publishing so the topic is actually consumed and produced to.
In `@backend/app/services/k8s_worker/worker.py`:
- Around line 72-104: The current handle_delete_pod_command lumps pod and
ConfigMap deletion in one try/except so a 404 from _v1.delete_namespaced_pod
prevents cleanup of the ConfigMap; split into two independent delete attempts
using asyncio.to_thread for _v1.delete_namespaced_pod (pod_name =
f"executor-{execution_id}") and _v1.delete_namespaced_config_map (configmap_name
= f"script-{execution_id}") each wrapped in its own try/except that treats
ApiException with status == 404 as a warning and continues, while logging
successes and non-404 errors separately so ConfigMap deletion always runs even
if pod deletion returns 404.
In `@backend/app/services/pod_monitor/monitor.py`:
- Around line 94-114: The current handle_raw_event implementation only catches
KeyError/ValueError and can crash on AttributeError/TypeError when
raw_event["object"] is not the expected Pod-like object; update handle_raw_event
to defensively validate raw_event["object"] (e.g., ensure it is not None and has
a metadata attribute with resource_version) before constructing PodEvent or
broaden the except clause to also catch AttributeError and TypeError, log the
error via self._logger.error with the exception details, and call
self._metrics.record_pod_monitor_watch_error(ErrorType.PROCESSING_ERROR) the
same way as other parse errors; ensure references to PodEvent, WatchEventType,
_process_pod_event, _logger, and _metrics.record_pod_monitor_watch_error are
used so the change is localized to this method.
In `@backend/app/services/result_processor/processor.py`:
- Around line 89-91: The current parsing of K8S_POD_MEMORY_LIMIT via
settings_limit.rstrip("Mi") is brittle and will mis-handle values like "1Gi",
"512M", "128Ki" or plain numbers; replace the inline parsing where
settings_limit is used to compute memory_limit_mib and memory_percent with a
small robust parser (e.g., a new helper function parse_memory_to_mib or inline
logic) that: trims whitespace, lowercases the string, extracts the numeric part
and unit via regex, supports units Ki/K/KiB, Mi/M/MiB, Gi/G/GiB (and no unit
treated as bytes or Mi depending on your convention), converts those units to
MiB consistently, and raises or logs a clear error on invalid formats; update
the assignment to memory_limit_mib to call this parser and ensure memory_percent
calculation stays the same and handles division by zero safely.
🧹 Nitpick comments (14)
backend/tests/unit/events/test_schema_registry_manager.py (1)
28-34: Assertuser_idis preserved in metadata.The test adds
user_idbut doesn’t verify it’s retained post-deserialization. Add an assertion to lock in the new behavior.♻️ Proposed test assertion
assert isinstance(ev, ExecutionRequestedEvent) assert ev.execution_id == "e1" assert ev.language == "python" + assert ev.metadata["user_id"] == "test"backend/app/db/repositories/redis/user_limit_repository.py (2)
23-25: Counter can go negative ifdecrementis called without a matching increment.If
decrementis called erroneously (e.g., double-decrement due to a bug or retry) or for a user who never incremented, the counter becomes negative. This could allow that user to exceed the limit on subsequent calls.Consider adding a floor check or using a Lua script to ensure the counter doesn't drop below zero:
🛠️ Proposed fix
async def decrement(self, user_id: str) -> None: """Decrement user count.""" - await self._redis.hincrby(self.KEY, user_id, -1) # type: ignore[misc] + # Use Lua script to ensure counter doesn't go below 0 + lua_script = """ + local current = redis.call('HGET', KEYS[1], ARGV[1]) + if current and tonumber(current) > 0 then + return redis.call('HINCRBY', KEYS[1], ARGV[1], -1) + end + return 0 + """ + await self._redis.eval(lua_script, 1, self.KEY, user_id) # type: ignore[misc]
6-9: No TTL or cleanup for stale user entries in the hash.The
exec:user_counthash accumulates entries for all users indefinitely. Over time, this could grow unbounded if users are deleted or become inactive.Consider periodically pruning zero-valued entries or implementing a sliding window approach with expiring keys per user.
backend/app/events/core/dlq_handler.py (1)
14-24:retry_countsdict accumulates indefinitely for successfully retried events.When an event succeeds on retry (before exceeding
max_retries), its entry inretry_countsis never removed. Over time, this could lead to unbounded memory growth in long-running workers.♻️ Suggested improvement: clean up on success or use TTL-based cache
Consider adding cleanup logic when retry succeeds, or using an LRU cache with a max size:
from functools import lru_cache # Or simply track via Redis with TTL instead of in-memory dictbackend/app/services/coordinator/coordinator.py (1)
81-86: Inconsistent user_id retrieval across handlers.
handle_execution_cancelledfetchesuser_idvia a repository lookup, whilehandle_execution_completedandhandle_execution_failedaccessevent.metadata.user_iddirectly. Given thatEventMetadata.user_idis a required field (pertyped.py), consider usingevent.metadata.user_idconsistently:♻️ Suggested simplification
async def handle_execution_cancelled(self, event: ExecutionCancelledEvent) -> None: """Handle execution cancelled - decrement user counter.""" self._logger.info(f"Handling ExecutionCancelledEvent: {event.execution_id}") - exec_rec = await self._execution_repository.get_execution(event.execution_id) - if exec_rec and exec_rec.user_id: - await self._user_limit.decrement(exec_rec.user_id) + if event.metadata.user_id: + await self._user_limit.decrement(event.metadata.user_id)This removes the DB lookup overhead and aligns with the other handlers.
backend/tests/unit/services/pod_monitor/test_monitor.py (1)
49-82: FakePodStateRepository is missing several methods from the real implementation.The fake is missing methods that exist in
PodStateRepository:try_claim_creation,release_creation,get_active_creations_count,is_creation_active,update_pod_status, andget_pod_state. While current tests may not exercise these paths, incomplete fakes can lead toAttributeErrorif the code under test evolves.Consider adding stub implementations for completeness:
♻️ Suggested additions
async def try_claim_creation(self, execution_id: str, ttl_seconds: int = 300) -> bool: return True async def release_creation(self, execution_id: str) -> bool: return True async def get_active_creations_count(self) -> int: return 0 async def is_creation_active(self, execution_id: str) -> bool: return False async def update_pod_status(self, pod_name: str, status: str) -> bool: return pod_name in self._tracked async def get_pod_state(self, pod_name: str) -> None: return Nonebackend/app/db/repositories/redis/pod_state_repository.py (2)
88-89: Non-atomic HSET + EXPIRE could leave orphaned keys.If the process crashes after
hsetbut beforeexpire, the key will persist indefinitely. Consider using a pipeline or Lua script for atomicity:♻️ Proposed fix using pipeline
- await self._redis.hset(key, mapping=data) # type: ignore[misc] - await self._redis.expire(key, ttl_seconds) + async with self._redis.pipeline(transaction=True) as pipe: + await pipe.hset(key, mapping=data) # type: ignore[misc] + await pipe.expire(key, ttl_seconds) + await pipe.execute()
54-59:scan_itercounts are O(N) - consider alternatives at scale.Both
get_active_creations_countandget_tracked_pods_countiterate all matching keys. This is fine for typical pod counts but may become slow with thousands of pods. If high-frequency counting is needed at scale, consider maintaining atomic counters withINCR/DECRduringtrack_pod/untrack_podoperations.Also applies to: 136-141
backend/tests/e2e/services/coordinator/test_execution_coordinator.py (1)
14-29: Consider asserting observable side effects in handler tests.These tests now only check that no exception is raised. Adding a minimal assertion (e.g., producer calls or repo updates) would reduce false positives.
backend/app/dlq/manager.py (2)
75-77: Improve filter function type annotation.The
filter_funcparameter is typed asobject, which provides no type safety. This also causes thetype: ignoreon line 114.♻️ Suggested fix
+from typing import Callable + +# Type alias at module level +DLQFilter = Callable[[DLQMessage], bool] + class DLQManager: - _filters: list[object] = [] + _filters: list[DLQFilter] = [] - def add_filter(self, filter_func: object) -> None: + def add_filter(self, filter_func: DLQFilter) -> None: """Add a message filter.""" self._filters.append(filter_func)Then on line 114:
for filter_func in self._filters: - if not filter_func(message): # type: ignore[operator] + if not filter_func(message):
160-186: Document the intentional double-publish pattern in_retry_message.The method sends each retry to two topics:
retry_topic(lines 174–179) andoriginal_topic(lines 181–186). This appears intentional for audit tracking and reprocessing, but the rationale should be documented. Consider adding a docstring or inline comment explaining why both publishes are necessary.The topic prefix concern in the original comment is incorrect—
original_topicalready includesKAFKA_TOPIC_PREFIXwhen stored in the database.backend/app/core/providers.py (1)
600-659: Consider injecting settings into worker configs.
K8sWorkerConfig()andPodMonitorConfig()are created with defaults. For production deployments, these may need to be configurable viaSettingsto tune timeouts, batch sizes, or other parameters.backend/app/services/notification_service.py (1)
117-121: Improve channel handler type annotation.The
_channel_handlersis typed asdict[NotificationChannel, object], which loses type safety and causes thetype: ignoreon line 538.♻️ Suggested fix
+from typing import Callable, Awaitable + +# Type alias +ChannelHandler = Callable[[DomainNotification, DomainNotificationSubscription], Awaitable[None]] + class NotificationService: - _channel_handlers: dict[NotificationChannel, object] = { + _channel_handlers: dict[NotificationChannel, ChannelHandler] + + def __init__(self, ...): + ... + self._channel_handlers = { NotificationChannel.IN_APP: self._send_in_app, NotificationChannel.WEBHOOK: self._send_webhook, NotificationChannel.SLACK: self._send_slack, }Then on line 538:
- await handler(notification, subscription) # type: ignore + await handler(notification, subscription)backend/app/api/routes/dlq.py (1)
36-45: Consider wrappingtopicparameter withQuery()for consistency.The
topicparameter on line 39 uses a bareNonedefault, whilestatusandevent_typeon lines 38 and 40 useQuery(None). While FastAPI handles both, usingQuery(None)consistently ensures uniform OpenAPI documentation and explicit query parameter behavior.♻️ Suggested consistency fix
async def get_dlq_messages( repository: FromDishka[DLQRepository], status: DLQMessageStatus | None = Query(None), - topic: KafkaTopic | None = None, + topic: KafkaTopic | None = Query(None), event_type: EventType | None = Query(None), limit: int = Query(50, ge=1, le=1000), offset: int = Query(0, ge=0), ) -> DLQMessagesResponse:
| if idem_key: | ||
| is_new = await idempotency_repo.try_reserve(idem_key, ttl=86400) | ||
| if not is_new: | ||
| cached = await idempotency_repo.get_result(idem_key) | ||
| if cached: | ||
| return ExecutionResponse.model_validate_json(cached) |
There was a problem hiding this comment.
Missing handling when duplicate request finds no cached result.
If try_reserve returns False (duplicate) but get_result returns None (original request still processing), the code falls through and re-executes the request, defeating idempotency.
Consider returning a 409 Conflict or 202 Accepted to indicate the request is still being processed:
🐛 Proposed fix
if idem_key:
is_new = await idempotency_repo.try_reserve(idem_key, ttl=86400)
if not is_new:
cached = await idempotency_repo.get_result(idem_key)
if cached:
return ExecutionResponse.model_validate_json(cached)
+ # Original request still processing - return conflict
+ raise HTTPException(
+ status_code=409,
+ detail="Request with this idempotency key is currently being processed"
+ )🤖 Prompt for AI Agents
In `@backend/app/api/routes/execution.py` around lines 77 - 82, When idem_key is
present and idempotency_repo.try_reserve(idem_key, ttl=86400) returns False but
idempotency_repo.get_result(idem_key) returns None (meaning the original request
is still processing), avoid re-executing and instead short-circuit with an
appropriate HTTP response (e.g., raise HTTPException(status_code=202) or 409)
indicating the request is in-flight; update the block around idem_key /
try_reserve / get_result to check for cached is None and return a 202 Accepted
(or 409 Conflict) with a clear body or headers rather than falling through to
re-execution so idempotency is preserved.
| @provide | ||
| def get_unified_producer( | ||
| self, | ||
| settings: Settings, | ||
| aiokafka_producer: AIOKafkaProducer, | ||
| schema_registry: SchemaRegistryManager, | ||
| logger: logging.Logger, | ||
| dlq_metrics: DLQMetrics, | ||
| ) -> AsyncIterator[DLQManager]: | ||
| async with create_dlq_manager(settings, schema_registry, logger, dlq_metrics) as manager: | ||
| yield manager | ||
|
|
||
| @provide | ||
| def get_idempotency_repository(self, redis_client: redis.Redis) -> RedisIdempotencyRepository: | ||
| return RedisIdempotencyRepository(redis_client, key_prefix="idempotency") | ||
| event_metrics: EventMetrics, | ||
| producer_metrics: ProducerMetrics, | ||
| ) -> UnifiedProducer: | ||
| return UnifiedProducer( | ||
| producer=aiokafka_producer, | ||
| schema_registry_manager=schema_registry, | ||
| logger=logger, | ||
| event_metrics=event_metrics, | ||
| producer_metrics=producer_metrics, | ||
| ) |
There was a problem hiding this comment.
Missing topic_prefix parameter for UnifiedProducer.
The UnifiedProducer constructor accepts a topic_prefix parameter (defaulting to ""), but it's not being passed from settings here. This may cause topics to be created without the expected prefix if KAFKA_TOPIC_PREFIX is configured.
🛠️ Suggested fix
`@provide`
def get_unified_producer(
self,
aiokafka_producer: AIOKafkaProducer,
schema_registry: SchemaRegistryManager,
logger: logging.Logger,
event_metrics: EventMetrics,
producer_metrics: ProducerMetrics,
+ settings: Settings,
) -> UnifiedProducer:
return UnifiedProducer(
producer=aiokafka_producer,
schema_registry_manager=schema_registry,
logger=logger,
event_metrics=event_metrics,
producer_metrics=producer_metrics,
+ topic_prefix=settings.KAFKA_TOPIC_PREFIX,
)🤖 Prompt for AI Agents
In `@backend/app/core/providers.py` around lines 207 - 222, The factory function
get_unified_producer currently constructs UnifiedProducer without passing the
topic_prefix setting; update get_unified_producer to read the configured prefix
(e.g., from settings.KAFKA_TOPIC_PREFIX or the app Settings instance) and pass
it into the UnifiedProducer(...) call via the topic_prefix parameter so created
topics include the configured prefix; ensure you reference get_unified_producer
and UnifiedProducer when making the change and use the existing settings symbol
(KAFKA_TOPIC_PREFIX) used elsewhere in the codebase.
| async def try_reserve(self, key: str, ttl: int | None = None) -> bool: | ||
| """Reserve key atomically. Returns True if new (should process), False if duplicate.""" | ||
| full_key = f"{self.KEY_PREFIX}:{key}" | ||
| result = await self._redis.set(full_key, "1", nx=True, ex=ttl or self._default_ttl) | ||
| return result is not None |
There was a problem hiding this comment.
Race condition: duplicate request may receive placeholder "1" instead of cached result.
When try_reserve returns False, the caller retrieves the cached result via get_result. However, if the original request hasn't called store_result yet, get_result returns "1" (the placeholder from the initial SET NX), which will fail JSON deserialization upstream.
Consider using a two-key pattern (lock key + result key) or returning a sentinel indicating "processing in progress" so the caller can retry or wait.
🛠️ Suggested approach: separate lock and result keys
+ LOCK_PREFIX = "idempotent:lock"
+ RESULT_PREFIX = "idempotent:result"
async def try_reserve(self, key: str, ttl: int | None = None) -> bool:
"""Reserve key atomically. Returns True if new (should process), False if duplicate."""
- full_key = f"{self.KEY_PREFIX}:{key}"
- result = await self._redis.set(full_key, "1", nx=True, ex=ttl or self._default_ttl)
+ lock_key = f"{self.LOCK_PREFIX}:{key}"
+ result = await self._redis.set(lock_key, "processing", nx=True, ex=ttl or self._default_ttl)
return result is not None
async def store_result(self, key: str, result_json: str, ttl: int | None = None) -> None:
"""Store result for duplicate requests to retrieve."""
- full_key = f"{self.KEY_PREFIX}:{key}"
- await self._redis.set(full_key, result_json, ex=ttl or self._default_ttl)
+ result_key = f"{self.RESULT_PREFIX}:{key}"
+ await self._redis.set(result_key, result_json, ex=ttl or self._default_ttl)
async def get_result(self, key: str) -> str | None:
- """Get cached result if exists."""
- full_key = f"{self.KEY_PREFIX}:{key}"
- result = await self._redis.get(full_key)
- return str(result) if result is not None else None
+ """Get cached result if exists. Returns None if still processing."""
+ result_key = f"{self.RESULT_PREFIX}:{key}"
+ result = await self._redis.get(result_key)
+ if result is None:
+ return None
+ return result.decode() if isinstance(result, bytes) else str(result)🤖 Prompt for AI Agents
In `@backend/app/db/repositories/redis/idempotency_repository.py` around lines 21
- 25, try_reserve currently sets the same KEY_PREFIX key to the placeholder "1",
so callers that see False may read that placeholder via get_result and fail JSON
deserialization; change to a two-key pattern: use a lock key (e.g.,
f"{self.KEY_PREFIX}:lock:{key}") set with NX and TTL inside try_reserve and
return True/False based on that, and write actual results to a separate result
key (e.g., f"{self.KEY_PREFIX}:result:{key}") in store_result; update get_result
to read the result key and either return None / a sentinel or retry/wait if the
lock key exists (indicating processing in progress) so callers never try to
deserialize the "in-progress" placeholder. Ensure references to try_reserve,
store_result, get_result, KEY_PREFIX and _default_ttl are updated accordingly.
| async def get_result(self, key: str) -> str | None: | ||
| """Get cached result if exists.""" | ||
| full_key = f"{self.KEY_PREFIX}:{key}" | ||
| result = await self._redis.get(full_key) | ||
| return str(result) if result is not None else None |
There was a problem hiding this comment.
str(result) may produce incorrect output for bytes.
If the Redis client returns bytes, str(result) produces "b'...'" rather than the decoded string. This would break JSON parsing downstream.
🐛 Proposed fix
async def get_result(self, key: str) -> str | None:
"""Get cached result if exists."""
full_key = f"{self.KEY_PREFIX}:{key}"
result = await self._redis.get(full_key)
- return str(result) if result is not None else None
+ if result is None:
+ return None
+ return result.decode("utf-8") if isinstance(result, bytes) else str(result)🤖 Prompt for AI Agents
In `@backend/app/db/repositories/redis/idempotency_repository.py` around lines 32
- 36, In get_result, the code converts Redis return values with str(result)
which yields "b'...'" for bytes; change get_result (referencing get_result,
KEY_PREFIX and _redis.get) to detect bytes and decode using result.decode() (or
return result directly if already a str) so the function returns a proper string
or None suitable for JSON parsing.
| async def update_pod_status(self, pod_name: str, status: str) -> bool: | ||
| """Update a tracked pod's status. Returns True if updated.""" | ||
| key = f"{self.TRACKED_KEY_PREFIX}:{pod_name}" | ||
| exists = await self._redis.exists(key) | ||
| if not exists: | ||
| return False | ||
|
|
||
| now = datetime.now(timezone.utc).isoformat() | ||
| await self._redis.hset(key, mapping={"status": status, "updated_at": now}) # type: ignore[misc] | ||
| return True |
There was a problem hiding this comment.
TOCTOU race in update_pod_status.
Between the exists check and hset, another process could delete the key. The subsequent hset would create a malformed entry with only status and updated_at fields. Consider using a Lua script or WATCH/MULTI for atomicity, or accept the race and verify all required fields exist after update.
🤖 Prompt for AI Agents
In `@backend/app/db/repositories/redis/pod_state_repository.py` around lines 92 -
101, The update_pod_status function has a TOCTOU race between the exists check
and the hset call; replace the two-step logic in update_pod_status with an
atomic Redis operation (e.g., a Lua EVAL script or a WATCH/MULTI transaction) so
the update only occurs if the key still exists and return a boolean based on the
atomic result; locate update_pod_status in PodStateRepository (method
update_pod_status) and implement an EVAL that checks redis.call('exists', key)
and performs hset inside the script, returning success/failure, or use WATCH on
the key then MULTI/HSET/EXEC and handle EXEC returning None as failure.
| async def _update_message_status(self, event_id: str, update: DLQMessageUpdate) -> None: | ||
| """Update DLQ message status.""" | ||
| doc = await DLQMessageDocument.find_one({"event.event_id": event_id}) | ||
| if not doc: | ||
| return | ||
|
|
||
| update_dict: dict[str, Any] = {"status": update.status, "last_updated": datetime.now(timezone.utc)} | ||
| if update.next_retry_at is not None: | ||
| update_dict["next_retry_at"] = update.next_retry_at | ||
| if update.retried_at is not None: | ||
| update_dict["retried_at"] = update.retried_at | ||
| if update.discarded_at is not None: | ||
| update_dict["discarded_at"] = update.discarded_at | ||
| if update.retry_count is not None: | ||
| update_dict["retry_count"] = update.retry_count | ||
| if update.discard_reason is not None: | ||
| update_dict["discard_reason"] = update.discard_reason | ||
| if update.last_error is not None: | ||
| update_dict["last_error"] = update.last_error | ||
|
|
||
| await doc.set(update_dict) | ||
| updates = {k: v for k, v in vars(update).items() if v is not None} | ||
| updates["last_updated"] = datetime.now(timezone.utc) | ||
| await doc.set(updates) |
There was a problem hiding this comment.
Use model_dump() instead of vars() for Pydantic models.
Using vars() on a Pydantic model may include internal attributes and doesn't respect Pydantic's serialization logic. Use model_dump(exclude_unset=True) for correct behavior.
🐛 Proposed fix
async def _update_message_status(self, event_id: str, update: DLQMessageUpdate) -> None:
"""Update DLQ message status."""
doc = await DLQMessageDocument.find_one({"event.event_id": event_id})
if not doc:
return
- updates = {k: v for k, v in vars(update).items() if v is not None}
+ updates = {k: v for k, v in update.model_dump(exclude_unset=True).items() if v is not None}
updates["last_updated"] = datetime.now(timezone.utc)
await doc.set(updates)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async def _update_message_status(self, event_id: str, update: DLQMessageUpdate) -> None: | |
| """Update DLQ message status.""" | |
| doc = await DLQMessageDocument.find_one({"event.event_id": event_id}) | |
| if not doc: | |
| return | |
| update_dict: dict[str, Any] = {"status": update.status, "last_updated": datetime.now(timezone.utc)} | |
| if update.next_retry_at is not None: | |
| update_dict["next_retry_at"] = update.next_retry_at | |
| if update.retried_at is not None: | |
| update_dict["retried_at"] = update.retried_at | |
| if update.discarded_at is not None: | |
| update_dict["discarded_at"] = update.discarded_at | |
| if update.retry_count is not None: | |
| update_dict["retry_count"] = update.retry_count | |
| if update.discard_reason is not None: | |
| update_dict["discard_reason"] = update.discard_reason | |
| if update.last_error is not None: | |
| update_dict["last_error"] = update.last_error | |
| await doc.set(update_dict) | |
| updates = {k: v for k, v in vars(update).items() if v is not None} | |
| updates["last_updated"] = datetime.now(timezone.utc) | |
| await doc.set(updates) | |
| async def _update_message_status(self, event_id: str, update: DLQMessageUpdate) -> None: | |
| """Update DLQ message status.""" | |
| doc = await DLQMessageDocument.find_one({"event.event_id": event_id}) | |
| if not doc: | |
| return | |
| updates = {k: v for k, v in update.model_dump(exclude_unset=True).items() if v is not None} | |
| updates["last_updated"] = datetime.now(timezone.utc) | |
| await doc.set(updates) |
🤖 Prompt for AI Agents
In `@backend/app/dlq/manager.py` around lines 150 - 158, In
_update_message_status, replace using vars(update) with Pydantic's
serialization: call update.model_dump(exclude_unset=True) (or model_dump() with
appropriate exclude flags) to build the updates dict so internal attrs aren't
included; then add updates["last_updated"] = datetime.now(timezone.utc) and pass
that to doc.set as before (target symbols: _update_message_status and
DLQMessageUpdate).
| # Resource topics | ||
| RESOURCE_EVENTS = "resource_events" | ||
| RESOURCE_ALLOCATION = "resource_allocation" |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
rg -n 'RESOURCE_ALLOCATION' --type=pyRepository: HardMax71/Integr8sCode
Length of output: 465
🏁 Script executed:
# Check for producers of RESOURCE_ALLOCATION
rg -n 'RESOURCE_ALLOCATION|resource_allocation' --type=py -B 3 -A 3
# Also check the CONSUMER_GROUP_SUBSCRIPTIONS structure
rg -n 'CONSUMER_GROUP_SUBSCRIPTIONS' --type=py -A 20Repository: HardMax71/Integr8sCode
Length of output: 23150
🏁 Script executed:
# Search for producers or sends to RESOURCE_ALLOCATION
rg -n 'producer.*RESOURCE_ALLOCATION|send.*RESOURCE_ALLOCATION|produce.*resource_allocation' --type=py
# Check dlq_processor context more thoroughly
cat -n backend/workers/dlq_processor.py | head -60Repository: HardMax71/Integr8sCode
Length of output: 2439
🏁 Script executed:
# Search for any producer.send or .publish patterns with topics
rg -n 'producer\.send|producer\.publish|\.send\(|KafkaTopic\.' backend/app/services/saga --type=py -A 2 | grep -E '(RESOURCE_ALLOCATION|producer|send|publish)' -B 1 -A 1
# Also check if RESOURCE_ALLOCATION is produced in any event handler
rg -n 'KafkaTopic\.RESOURCE_ALLOCATION' --type=py -B 5 -A 5Repository: HardMax71/Integr8sCode
Length of output: 902
Remove unused RESOURCE_ALLOCATION topic or add consumer subscription.
The RESOURCE_ALLOCATION topic is defined and configured in the DLQ processor, but no consumer group in CONSUMER_GROUP_SUBSCRIPTIONS subscribes to it, and no producer sends messages to it. Either implement the consumer subscription and producer logic for this topic, or remove it if it's no longer needed.
🤖 Prompt for AI Agents
In `@backend/app/domain/enums/kafka.py` around lines 35 - 37, The
RESOURCE_ALLOCATION topic constant is defined but unused; either remove
RESOURCE_ALLOCATION from backend/app/domain/enums/kafka.py or wire it up: add it
to the appropriate consumer subscription set (CONSUMER_GROUP_SUBSCRIPTIONS) and
ensure a producer sends messages to it (or update the DLQ processor to stop
relying on it). Locate the constant RESOURCE_ALLOCATION, the consumer map
CONSUMER_GROUP_SUBSCRIPTIONS, and the DLQ processor code that references
RESOURCE_ALLOCATION, then either delete the constant and any DLQ references or
add a subscription entry and implement/verify producer publishing so the topic
is actually consumed and produced to.
| async def handle_delete_pod_command(self, command: DeletePodCommandEvent) -> None: | ||
| """Handle delete pod command from saga orchestrator (compensation).""" | ||
| execution_id = command.execution_id | ||
| self.logger.info(f"Deleting pod for execution {execution_id} due to: {command.reason}") | ||
| self._logger.info(f"Deleting pod for execution {execution_id} due to: {command.reason}") | ||
|
|
||
| try: | ||
| # Delete the pod | ||
| pod_name = f"executor-{execution_id}" | ||
| if self.v1: | ||
| await asyncio.to_thread( | ||
| self.v1.delete_namespaced_pod, | ||
| name=pod_name, | ||
| namespace=self.config.namespace, | ||
| grace_period_seconds=30, | ||
| ) | ||
| self.logger.info(f"Successfully deleted pod {pod_name}") | ||
| await asyncio.to_thread( | ||
| self._v1.delete_namespaced_pod, | ||
| name=pod_name, | ||
| namespace=self._config.namespace, | ||
| grace_period_seconds=30, | ||
| ) | ||
| self._logger.info(f"Successfully deleted pod {pod_name}") | ||
|
|
||
| # Delete associated ConfigMap | ||
| configmap_name = f"script-{execution_id}" | ||
| if self.v1: | ||
| await asyncio.to_thread( | ||
| self.v1.delete_namespaced_config_map, name=configmap_name, namespace=self.config.namespace | ||
| ) | ||
| self.logger.info(f"Successfully deleted ConfigMap {configmap_name}") | ||
|
|
||
| # NetworkPolicy cleanup is managed via a static cluster policy; no per-execution NP deletion | ||
| await asyncio.to_thread( | ||
| self._v1.delete_namespaced_config_map, | ||
| name=configmap_name, | ||
| namespace=self._config.namespace, | ||
| ) | ||
| self._logger.info(f"Successfully deleted ConfigMap {configmap_name}") | ||
|
|
||
| except ApiException as e: | ||
| if e.status == 404: | ||
| self.logger.warning(f"Resources for execution {execution_id} not found (may have already been deleted)") | ||
| self._logger.warning( | ||
| f"Resources for execution {execution_id} not found (may have already been deleted)" | ||
| ) | ||
| else: | ||
| self.logger.error(f"Failed to delete resources for execution {execution_id}: {e}") | ||
| self._logger.error(f"Failed to delete resources for execution {execution_id}: {e}") | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cat -n backend/app/services/k8s_worker/worker.py | sed -n '65,115p'Repository: HardMax71/Integr8sCode
Length of output: 2583
🏁 Script executed:
# Check imports and exception handling patterns
head -50 backend/app/services/k8s_worker/worker.py | grep -E "^import|^from"Repository: HardMax71/Integr8sCode
Length of output: 722
🏁 Script executed:
# Search for other usage patterns of delete_namespaced_pod in the codebase
rg "delete_namespaced_pod" --type py -B 3 -A 3Repository: HardMax71/Integr8sCode
Length of output: 1853
🏁 Script executed:
# Verify Kubernetes API exception behavior for 404 scenarios
web_search "kubernetes python client ApiException 404 delete_namespaced_pod behavior"Repository: HardMax71/Integr8sCode
Length of output: 152
🌐 Web query:
kubernetes python client delete_namespaced_pod ApiException 404
💡 Result:
CoreV1Api.delete_namespaced_pod(name, namespace, ...) raises ApiException(status=404) when the Kubernetes API server can’t find a Pod with that exact {namespace}/{name} at the moment of the DELETE request (the REST endpoint is DELETE /api/v1/namespaces/{namespace}/pods/{name}) [1]. A very common cause is simply that the pod was already deleted (by a previous call, by a controller recreating/replacing pods, TTL/cleanup, etc.), so your delete is racing something else.
Because that “already gone” case is usually fine, the common pattern is to treat 404 as success and re-raise anything else (Airflow does exactly this around delete_namespaced_pod) [2]:
from kubernetes import client
from kubernetes.client.exceptions import ApiException
v1 = client.CoreV1Api()
try:
v1.delete_namespaced_pod(
name=pod_name,
namespace=namespace,
body=client.V1DeleteOptions(),
)
except ApiException as e:
if e.status != 404:
raiseIf you didn’t expect 404, double-check you’re deleting the pod name (not a Deployment/Job name), and the namespace matches; pod names often include a generated suffix (e.g., myapp-7c9f8d6b6f-abcde), so deleting by a “base name” will 404 [1][3].
Sources: Kubernetes API endpoint for deleting pods [1]; kubernetes-python-client method signatures [3]; Airflow’s real-world handling of 404-on-delete [2].
Separate pod and ConfigMap deletion to ensure ConfigMap cleanup despite pod 404 errors.
When pod deletion raises a 404, the current single try-catch block skips ConfigMap cleanup entirely, leaving stale ConfigMaps behind. As a compensation operation, both resources should be deleted independently—treat 404 as non-fatal and continue cleanup.
🛠️ Suggested fix
- try:
- # Delete the pod
- pod_name = f"executor-{execution_id}"
- await asyncio.to_thread(
- self._v1.delete_namespaced_pod,
- name=pod_name,
- namespace=self._config.namespace,
- grace_period_seconds=30,
- )
- self._logger.info(f"Successfully deleted pod {pod_name}")
-
- # Delete associated ConfigMap
- configmap_name = f"script-{execution_id}"
- await asyncio.to_thread(
- self._v1.delete_namespaced_config_map,
- name=configmap_name,
- namespace=self._config.namespace,
- )
- self._logger.info(f"Successfully deleted ConfigMap {configmap_name}")
-
- except ApiException as e:
- if e.status == 404:
- self._logger.warning(
- f"Resources for execution {execution_id} not found (may have already been deleted)"
- )
- else:
- self._logger.error(f"Failed to delete resources for execution {execution_id}: {e}")
+ # Delete pod
+ try:
+ pod_name = f"executor-{execution_id}"
+ await asyncio.to_thread(
+ self._v1.delete_namespaced_pod,
+ name=pod_name,
+ namespace=self._config.namespace,
+ grace_period_seconds=30,
+ )
+ self._logger.info(f"Successfully deleted pod {pod_name}")
+ except ApiException as e:
+ if e.status == 404:
+ self._logger.warning(
+ f"Pod for execution {execution_id} not found (may have already been deleted)"
+ )
+ else:
+ self._logger.error(f"Failed to delete pod for execution {execution_id}: {e}")
+
+ # Delete ConfigMap (attempt regardless of pod deletion)
+ try:
+ configmap_name = f"script-{execution_id}"
+ await asyncio.to_thread(
+ self._v1.delete_namespaced_config_map,
+ name=configmap_name,
+ namespace=self._config.namespace,
+ )
+ self._logger.info(f"Successfully deleted ConfigMap {configmap_name}")
+ except ApiException as e:
+ if e.status == 404:
+ self._logger.warning(
+ f"ConfigMap for execution {execution_id} not found (may have already been deleted)"
+ )
+ else:
+ self._logger.error(f"Failed to delete ConfigMap for execution {execution_id}: {e}")🤖 Prompt for AI Agents
In `@backend/app/services/k8s_worker/worker.py` around lines 72 - 104, The current
handle_delete_pod_command lumps pod and ConfigMap deletion in one try/except so
a 404 from _v1.delete_namespaced_pod prevents cleanup of the ConfigMap; split
into two independent delete attempts using asyncio.to_thread for
_v1.delete_namespaced_pod (pod_name = f"executor-{execution_id}") and
_v1.delete_namespaced_config_map (configmap_name = f"script-{execution_id}")
each wrapped in its own try/except that treats ApiException with status == 404
as a warning and continues, while logging successes and non-404 errors
separately so ConfigMap deletion always runs even if pod deletion returns 404.
| async def handle_raw_event(self, raw_event: dict[str, Any]) -> None: | ||
| """Process a raw Kubernetes watch event. | ||
|
|
||
| Called by worker entrypoint for each event from watch stream. | ||
| """ | ||
| try: | ||
| for event in stream: | ||
| if self._state != MonitorState.RUNNING: | ||
| break | ||
|
|
||
| await self._process_raw_event(event) | ||
|
|
||
| finally: | ||
| # Store resource version for next watch | ||
| self._update_resource_version(stream) | ||
|
|
||
| def _update_resource_version(self, stream: Any) -> None: | ||
| """Update last resource version from stream.""" | ||
| try: | ||
| if stream._stop_event and stream._stop_event.resource_version: | ||
| self._last_resource_version = stream._stop_event.resource_version | ||
| except AttributeError: | ||
| pass | ||
|
|
||
| async def _process_raw_event(self, raw_event: KubeEvent) -> None: | ||
| """Process a raw Kubernetes watch event.""" | ||
| try: | ||
| # Parse event | ||
| event = PodEvent( | ||
| event_type=WatchEventType(raw_event["type"].upper()), | ||
| pod=raw_event["object"], | ||
| resource_version=( | ||
| raw_event["object"].metadata.resource_version if raw_event["object"].metadata else None | ||
| raw_event["object"].metadata.resource_version | ||
| if raw_event["object"].metadata | ||
| else None | ||
| ), | ||
| ) | ||
|
|
||
| await self._process_pod_event(event) | ||
|
|
||
| except (KeyError, ValueError) as e: | ||
| self.logger.error(f"Invalid event format: {e}") | ||
| self._logger.error(f"Invalid event format: {e}") | ||
| self._metrics.record_pod_monitor_watch_error(ErrorType.PROCESSING_ERROR) |
There was a problem hiding this comment.
Harden raw event parsing to avoid unexpected crash paths.
handle_raw_event only catches KeyError/ValueError, but attribute access on raw_event["object"] can raise AttributeError/TypeError for non‑Pod watch events (e.g., ERROR/BOOKMARK). That would bubble out and potentially kill the watch loop. Add safer access/validation or broaden the exception handling.
🔧 Suggested defensive handling
- try:
- event = PodEvent(
- event_type=WatchEventType(raw_event["type"].upper()),
- pod=raw_event["object"],
- resource_version=(
- raw_event["object"].metadata.resource_version
- if raw_event["object"].metadata
- else None
- ),
- )
+ try:
+ obj = raw_event["object"]
+ event = PodEvent(
+ event_type=WatchEventType(raw_event["type"].upper()),
+ pod=obj,
+ resource_version=(
+ obj.metadata.resource_version if getattr(obj, "metadata", None) else None
+ ),
+ )
await self._process_pod_event(event)
- except (KeyError, ValueError) as e:
+ except (KeyError, ValueError, AttributeError, TypeError) as e:
self._logger.error(f"Invalid event format: {e}")
self._metrics.record_pod_monitor_watch_error(ErrorType.PROCESSING_ERROR)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| async def handle_raw_event(self, raw_event: dict[str, Any]) -> None: | |
| """Process a raw Kubernetes watch event. | |
| Called by worker entrypoint for each event from watch stream. | |
| """ | |
| try: | |
| for event in stream: | |
| if self._state != MonitorState.RUNNING: | |
| break | |
| await self._process_raw_event(event) | |
| finally: | |
| # Store resource version for next watch | |
| self._update_resource_version(stream) | |
| def _update_resource_version(self, stream: Any) -> None: | |
| """Update last resource version from stream.""" | |
| try: | |
| if stream._stop_event and stream._stop_event.resource_version: | |
| self._last_resource_version = stream._stop_event.resource_version | |
| except AttributeError: | |
| pass | |
| async def _process_raw_event(self, raw_event: KubeEvent) -> None: | |
| """Process a raw Kubernetes watch event.""" | |
| try: | |
| # Parse event | |
| event = PodEvent( | |
| event_type=WatchEventType(raw_event["type"].upper()), | |
| pod=raw_event["object"], | |
| resource_version=( | |
| raw_event["object"].metadata.resource_version if raw_event["object"].metadata else None | |
| raw_event["object"].metadata.resource_version | |
| if raw_event["object"].metadata | |
| else None | |
| ), | |
| ) | |
| await self._process_pod_event(event) | |
| except (KeyError, ValueError) as e: | |
| self.logger.error(f"Invalid event format: {e}") | |
| self._logger.error(f"Invalid event format: {e}") | |
| self._metrics.record_pod_monitor_watch_error(ErrorType.PROCESSING_ERROR) | |
| async def handle_raw_event(self, raw_event: dict[str, Any]) -> None: | |
| """Process a raw Kubernetes watch event. | |
| Called by worker entrypoint for each event from watch stream. | |
| """ | |
| try: | |
| obj = raw_event["object"] | |
| event = PodEvent( | |
| event_type=WatchEventType(raw_event["type"].upper()), | |
| pod=obj, | |
| resource_version=( | |
| obj.metadata.resource_version if getattr(obj, "metadata", None) else None | |
| ), | |
| ) | |
| await self._process_pod_event(event) | |
| except (KeyError, ValueError, AttributeError, TypeError) as e: | |
| self._logger.error(f"Invalid event format: {e}") | |
| self._metrics.record_pod_monitor_watch_error(ErrorType.PROCESSING_ERROR) |
🤖 Prompt for AI Agents
In `@backend/app/services/pod_monitor/monitor.py` around lines 94 - 114, The
current handle_raw_event implementation only catches KeyError/ValueError and can
crash on AttributeError/TypeError when raw_event["object"] is not the expected
Pod-like object; update handle_raw_event to defensively validate
raw_event["object"] (e.g., ensure it is not None and has a metadata attribute
with resource_version) before constructing PodEvent or broaden the except clause
to also catch AttributeError and TypeError, log the error via self._logger.error
with the exception details, and call
self._metrics.record_pod_monitor_watch_error(ErrorType.PROCESSING_ERROR) the
same way as other parse errors; ensure references to PodEvent, WatchEventType,
_process_pod_event, _logger, and _metrics.record_pod_monitor_watch_error are
used so the change is localized to this method.
| settings_limit = self._settings.K8S_POD_MEMORY_LIMIT | ||
| memory_limit_mib = int(settings_limit.rstrip("Mi")) # TODO: Less brittle acquisition of limit | ||
| memory_limit_mib = int(settings_limit.rstrip("Mi")) | ||
| memory_percent = (memory_mib / memory_limit_mib) * 100 |
There was a problem hiding this comment.
Memory limit parsing assumes "Mi" suffix; may fail for other formats.
settings_limit.rstrip("Mi") will fail or produce incorrect results if K8S_POD_MEMORY_LIMIT uses a different format (e.g., "1Gi", "512M", or a raw number).
Consider a more robust parsing approach:
🐛 Proposed fix
- settings_limit = self._settings.K8S_POD_MEMORY_LIMIT
- memory_limit_mib = int(settings_limit.rstrip("Mi"))
+ settings_limit = self._settings.K8S_POD_MEMORY_LIMIT
+ # Parse Kubernetes memory format (e.g., "256Mi", "1Gi", "512M")
+ if settings_limit.endswith("Gi"):
+ memory_limit_mib = int(settings_limit[:-2]) * 1024
+ elif settings_limit.endswith("Mi"):
+ memory_limit_mib = int(settings_limit[:-2])
+ elif settings_limit.endswith("M"):
+ memory_limit_mib = int(settings_limit[:-1])
+ else:
+ # Assume bytes, convert to MiB
+ memory_limit_mib = int(settings_limit) // (1024 * 1024)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| settings_limit = self._settings.K8S_POD_MEMORY_LIMIT | |
| memory_limit_mib = int(settings_limit.rstrip("Mi")) # TODO: Less brittle acquisition of limit | |
| memory_limit_mib = int(settings_limit.rstrip("Mi")) | |
| memory_percent = (memory_mib / memory_limit_mib) * 100 | |
| settings_limit = self._settings.K8S_POD_MEMORY_LIMIT | |
| # Parse Kubernetes memory format (e.g., "256Mi", "1Gi", "512M") | |
| if settings_limit.endswith("Gi"): | |
| memory_limit_mib = int(settings_limit[:-2]) * 1024 | |
| elif settings_limit.endswith("Mi"): | |
| memory_limit_mib = int(settings_limit[:-2]) | |
| elif settings_limit.endswith("M"): | |
| memory_limit_mib = int(settings_limit[:-1]) | |
| else: | |
| # Assume bytes, convert to MiB | |
| memory_limit_mib = int(settings_limit) // (1024 * 1024) | |
| memory_percent = (memory_mib / memory_limit_mib) * 100 |
🤖 Prompt for AI Agents
In `@backend/app/services/result_processor/processor.py` around lines 89 - 91, The
current parsing of K8S_POD_MEMORY_LIMIT via settings_limit.rstrip("Mi") is
brittle and will mis-handle values like "1Gi", "512M", "128Ki" or plain numbers;
replace the inline parsing where settings_limit is used to compute
memory_limit_mib and memory_percent with a small robust parser (e.g., a new
helper function parse_memory_to_mib or inline logic) that: trims whitespace,
lowercases the string, extracts the numeric part and unit via regex, supports
units Ki/K/KiB, Mi/M/MiB, Gi/G/GiB (and no unit treated as bytes or Mi depending
on your convention), converts those units to MiB consistently, and raises or
logs a clear error on invalid formats; update the assignment to memory_limit_mib
to call this parser and ensure memory_percent calculation stays the same and
handles division by zero safely.
Summary by cubic
Moved service startup and lifecycle management into DI and refactored workers/services into stateless handlers. Added Redis-backed repositories and stronger typing to improve safety, scalability, and multi-instance behavior.
Refactors
New Features
Written for commit e12a973. Summary will update on new commits.
Summary by CodeRabbit
Bug Fixes
Chores
✏️ Tip: You can customize this high-level summary in your review settings.