Feat(#22): AI 분석 결과 이벤트 및 Worker 장애 처리 구현 - #26
Conversation
📝 WalkthroughWalkthroughChangesThe PR adds availability-aware scoring, centralized external API retries and timeouts, structured RabbitMQ result events, graceful consumer shutdown, retry/DLQ handling, Docker health checks, and CI validation. Analysis reliability
RabbitMQ processing
Build validation
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related issues
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 10
🧹 Nitpick comments (17)
Dockerfile (1)
17-18: 🩺 Stability & Availability | 🔵 TrivialMake the healthcheck reflect worker readiness, or document it as liveness-only.
The probed
/handler returns a static healthy payload and does not verify RabbitMQ or analysis-consumer state. A container with a live HTTP server but a disconnected consumer will therefore remain healthy. If this status drives readiness or traffic routing, add a dependency-aware readiness endpoint; otherwise explicitly treat this check as liveness-only.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@Dockerfile` around lines 17 - 18, Update the Dockerfile healthcheck to use a dependency-aware readiness endpoint that verifies RabbitMQ and analysis-consumer state, or explicitly document and configure the existing root check as liveness-only if it is not used for readiness or traffic routing.app/analysis/schemas.py (1)
40-44: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueStale rationale comment above the widened constraints.
The comment above (
상한이 두 시나리오 중 더 큰 쪽 기준으로 설정됨) explains a two-scenario (has-url / no-url) cap rationale, but the fields now use a flatge=0, le=100for all three, driven by dynamic availability-based weight renormalization inscoring.py(any single track can approach 100% weight when others are unavailable) — not just the two static scenarios described. Worth updating the comment so it doesn't mislead future readers about why the bound is 100.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/analysis/schemas.py` around lines 40 - 44, Update the Korean rationale comment above the llm, hybrid_url, and rules fields to explain that dynamic availability-based weight renormalization can make any remaining track approach 100% when others are unavailable. Remove the outdated two-scenario URL/LLM/rules redistribution explanation, while keeping the existing ge=0 and le=100 constraints unchanged.app/analysis/execution.py (1)
41-47: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winAsymmetric granularity: Gemini-only failure is indistinguishable from total text-track failure.
When Gemini fails but Naive Bayes already produced a trustworthy score,
service.pytreats the text track astext_available=Trueand keeps its full scoring weight (no fallback/derating) — the design comments explain this is intentionally low-risk (the NB score is still trusted). Yetclassify_executionmaps this case to the same generic"TEXT"tag (viagrade == "UNKNOWN") used when both engines fail and an arbitraryBOTH_ENGINES_UNAVAILABLE_FALLBACK_SCOREis substituted — a materially different reliability situation. The mirror case (NB fails, Gemini rescues) already gets a distinct"TEXT:NAIVE_BAYES"tag; there's no equivalent"TEXT:GEMINI"tag for symmetry.This doesn't change
final_score/risk_grade(that's computed correctly inscoring.py), but it degrades the precision of thePARTIALevent'sfailed_trackspayload for downstream consumers/observability trying to distinguish "one sub-engine down, still reliable" from "no text signal at all, using a constant fallback."♻️ Suggested distinction
if ( - text_result.get("grade") == "UNKNOWN" + text_result.get("grade") == "UNKNOWN" + and stage1_result.get("grade") in (None, "UNKNOWN") or text_result.get("error_message") ): failed_tracks.append("TEXT") elif stage1_result.get("error_message"): failed_tracks.append("TEXT:NAIVE_BAYES") + elif text_result.get("grade") == "UNKNOWN": + failed_tracks.append("TEXT:GEMINI")🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/analysis/execution.py` around lines 41 - 47, Update classify_execution’s text failure classification so a Gemini-only failure with a valid Naive Bayes score appends "TEXT:GEMINI" instead of the generic "TEXT" tag. Preserve "TEXT" for cases where both text engines fail and retain the existing "TEXT:NAIVE_BAYES" behavior when Naive Bayes fails.tests/analysis/test_execution.py (1)
1-108: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMissing coverage for TEXT / TEXT:NAIVE_BAYES / RULES failure branches.
Only URL-track failure paths and the terminal ERROR/PIPELINE path are tested.
classify_execution's text-track branch ("TEXT"vs"TEXT:NAIVE_BAYES") and the rules-failure branch ("RULES") are untested, despite being the core logic that determines theCOMPLETED/PARTIALevent classification for the majority of real-world failure modes (e.g., Gemini timeout, rule-engine crash).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/analysis/test_execution.py` around lines 1 - 108, Add focused tests for classify_execution covering text analysis failures that produce the expected TEXT or TEXT:NAIVE_BAYES failed-track identifiers, and rule_analysis failures that produce RULES. Assert each scenario’s COMPLETED/PARTIAL status and failed_tracks, while preserving the existing URL and pipeline error coverage.app/analysis/url/analyzer.py (2)
27-30: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueNit: annotate as
ClassVar[frozenset[str]].Silences Ruff RUF012 and makes the constant immutable.
- AVAILABLE_STATUSES = { - "safe", - "completed", - } + AVAILABLE_STATUSES: ClassVar[frozenset[str]] = frozenset( + {"safe", "completed"} + )Requires
from typing import ClassVar.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/analysis/url/analyzer.py` around lines 27 - 30, Update the class-level AVAILABLE_STATUSES constant to be annotated as ClassVar[frozenset[str]], import ClassVar from typing, and ensure its value remains an immutable frozenset to satisfy Ruff RUF012.Source: Linters/SAST tools
223-265: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider extracting the shared unavailable payload.
_scan_gsband_scan_virustotalare identical apart from the client and log message; a single helper (_call_provider(client, name)) removes the duplication and keeps the fallback payload in one place.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/analysis/url/analyzer.py` around lines 223 - 265, Extract the duplicated provider-call and unavailable-response logic from _scan_gsb and _scan_virustotal into a shared helper such as _call_provider(client, name). Have the helper invoke the supplied client, log the provider-specific failure, and return the common fallback payload; update both scan methods to delegate to it while preserving their existing client and provider names.app/analysis/url/tracker.py (1)
181-201: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winBind the loop variables in the retry lambdas (Ruff B023).
The lambdas are invoked within the same iteration, so behavior is correct today, but the closures still capture
current_url/headerslate and Ruff flags them. Binding them as default args silences the warning and keeps the call safe against future refactors.♻️ Suggested change
- response = await request_with_retry( - lambda: client.head( - current_url, - headers=headers, + response = await request_with_retry( + lambda url=current_url, hdrs=headers: client.head( + url, + headers=hdrs, timeout=timeout, ), max_retries=settings.EXTERNAL_API_MAX_RETRIES, operation_name="URL trace HEAD", ) # HEAD를 차단하거나 거부하는 서버(400, 404, 405)에 대응하기 위한 GET 폴백 if response.status_code in [400, 404, 405]: - response = await request_with_retry( - lambda: client.get( - current_url, - headers=headers, + response = await request_with_retry( + lambda url=current_url, hdrs=headers: client.get( + url, + headers=hdrs, timeout=timeout, ), max_retries=settings.EXTERNAL_API_MAX_RETRIES, operation_name="URL trace GET", )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/analysis/url/tracker.py` around lines 181 - 201, Update the retry lambdas in the URL trace flow to bind the current iteration’s current_url and headers as default arguments, for both the HEAD request and the GET fallback. Preserve the existing client.head/client.get calls and retry behavior while eliminating Ruff B023 late-binding warnings.Source: Linters/SAST tools
app/infrastructure/http_retry.py (1)
34-68: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winRetries fire immediately with no backoff.
On 429/5xx and network errors the next attempt is issued instantly, which amplifies load on an already-struggling upstream and makes rate-limit recovery unlikely. Add exponential backoff with jitter (and ideally honor
Retry-Afteron 429).♻️ Sketch
+import asyncio +import random + +_BASE_DELAY_SECONDS = 0.5 + + +async def _sleep_before_retry(attempt: int, retry_after: str | None = None) -> None: + if retry_after: + try: + await asyncio.sleep(min(float(retry_after), 10.0)) + return + except ValueError: + pass + delay = _BASE_DELAY_SECONDS * (2 ** attempt) + await asyncio.sleep(delay * (0.5 + random.random() / 2))Then
await _sleep_before_retry(attempt)before eachcontinue(passingresponse.headers.get("Retry-After")in the status-code branch).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/infrastructure/http_retry.py` around lines 34 - 68, Update the retry loop around the network-error and retryable-status branches to await a backoff helper before each retry continue. Implement or reuse _sleep_before_retry(attempt) with exponential backoff and jitter, and pass the 429 response’s Retry-After header in the status branch so that server guidance is honored when available.app/infrastructure/rabbitmq/result_factory.py (1)
187-190: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRedundant
int()cast flagged by Ruff.
round(float(value))withoutndigitsalready returns anint; the outerint(...)is unnecessary.🧹 Suggested fix
def _integer_score(value) -> int | None: if value is None: return None - return max(0, min(100, int(round(float(value))))) + return max(0, min(100, round(float(value))))🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/infrastructure/rabbitmq/result_factory.py` around lines 187 - 190, Update _integer_score to remove the redundant int() wrapper around round(float(value)), preserving the existing clamping behavior and return type.Source: Linters/SAST tools
app/infrastructure/rabbitmq/publisher.py (1)
61-71: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winNo failure classification on publish timeout/errors, unlike
DeadLetterPublisher.publish().
DeadLetterPublisher.publish()wraps itsasyncio.wait_forcall in try/except to raiseRetryableProcessingErrorwith a specificfailure_codeon timeout vs. generic failure.AnalysisResultPublisher.publish()has no equivalent handling, so a timeout/unroutable-message failure here propagates as a raw exception without a descriptive failure code for downstream retry/DLQ logging.♻️ Suggested fix mirroring dead_letter.py
- await asyncio.wait_for( - self.exchange.publish( - message, - routing_key=routing_key, - mandatory=True, - ), - timeout=( - self.settings - .RABBITMQ_PUBLISH_TIMEOUT_SECONDS - ), - ) + try: + await asyncio.wait_for( + self.exchange.publish( + message, + routing_key=routing_key, + mandatory=True, + ), + timeout=( + self.settings + .RABBITMQ_PUBLISH_TIMEOUT_SECONDS + ), + ) + except TimeoutError as exception: + raise RetryableProcessingError( + message="Result event publication timed out", + failure_code="RESULT_PUBLISH_TIMEOUT", + ) from exception + except Exception as exception: + raise RetryableProcessingError( + message="Failed to publish result event", + failure_code="RESULT_PUBLISH_FAILED", + ) from exception🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/infrastructure/rabbitmq/publisher.py` around lines 61 - 71, Update AnalysisResultPublisher.publish() around its asyncio.wait_for call to classify failures like DeadLetterPublisher.publish(): convert asyncio.TimeoutError into RetryableProcessingError with the timeout-specific failure_code, and convert other publish exceptions into RetryableProcessingError with the generic publish failure_code. Preserve the existing publish arguments and timeout configuration.app/infrastructure/rabbitmq/schemas.py (3)
121-127: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick win
riskGrade/phishingTypeare unconstrained strings instead of reusing the existingRiskGradeenum.
app.analysis.schemas.RiskGradealready exists (used byresult_factory.pyviagetattr(result.risk_grade, "value", ...)and imported directly in tests). TypingriskGradeasstr | Nonehere means any string passes validation except the single hard-coded"LOW"check on FAILED — typos or inconsistent casing wouldn't be caught.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/infrastructure/rabbitmq/schemas.py` around lines 121 - 127, Update AnalysisResultPayload to type riskGrade and phishingType with the existing RiskGrade enum from app.analysis.schemas instead of unconstrained strings, while preserving their optional nullable defaults and existing validation behavior.
121-138: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick win
AnalysisResultPayloaddoesn't enforce mutual exclusivity between success/failure fields.
finalScore/riskGradeandfailureCode/failedTrackscan all be populated simultaneously on the same payload; nothing in the model prevents a caller from setting both a completed score and a failure code.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/infrastructure/rabbitmq/schemas.py` around lines 121 - 138, Update AnalysisResultPayload with model-level validation that enforces mutually exclusive success and failure states: completed score fields (finalScore/riskGrade) must not coexist with failure indicators (failureCode/failedTracks). Preserve valid success-only, failure-only, and existing optional analysis-field payloads while rejecting mixed states.
154-187: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick win
validate_event_resultonly checks required-field presence, not forbidden combinations.For COMPLETED, nothing stops
failureCodebeing set orfailedTracksbeing non-empty; for PARTIAL/FAILED there's likewise no check that fields belonging to the "wrong" state are absent (the FAILED branch only special-casesriskGrade == "LOW"). Ifresult_factory.py(or a future caller) ever produces an inconsistent combination, this validator will still accept it and publish a self-contradictory event to Spring.♻️ Example hardening for the COMPLETED branch (apply similarly to PARTIAL/FAILED)
if self.eventType == AnalysisEventType.COMPLETED: if self.payload.finalScore is None: raise ValueError( "completed event requires finalScore" ) if self.payload.riskGrade is None: raise ValueError( "completed event requires riskGrade" ) + if self.payload.failureCode is not None: + raise ValueError( + "completed event must not set failureCode" + ) + if self.payload.failedTracks: + raise ValueError( + "completed event must not report failedTracks" + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/infrastructure/rabbitmq/schemas.py` around lines 154 - 187, Strengthen validate_event_result in AnalysisResultEvent to reject state-incompatible payload fields, not just missing required fields: COMPLETED must not include failureCode or non-empty failedTracks, and PARTIAL/FAILED must likewise reject fields belonging to other event states. Preserve the existing required-field checks and FAILED riskGrade restriction while enforcing mutually exclusive payload combinations for every AnalysisEventType.tests/infrastructure/rabbitmq/test_result_factory.py (1)
102-159: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider adding coverage for the untested branches flagged in
result_factory.py.No test currently exercises a
stage1_naive_bayes-absent + Gemini-only scenario (to pin down whetherTextAnalysisMethod.GEMINIshould ever be produced) or aurl_risk_scoreof exactly1(0-100 vs 0-1 scale ambiguity).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/infrastructure/rabbitmq/test_result_factory.py` around lines 102 - 159, Extend the result-factory tests around AnalysisResultEventFactory.create to cover a result with stage1_naive_bayes absent and Gemini-only text analysis, asserting the intended TextAnalysisMethod behavior, and add a case where url_risk_score is exactly 1, asserting the expected 0–100 score mapping. Reuse the existing request/result helpers and preserve current successful and failed execution coverage.app/infrastructure/rabbitmq/connection.py (1)
49-93: 🧹 Nitpick | 🔵 TrivialOperational note:
mandatory=Trueresult/DLQ publishes require the consumer-side queue binding to already exist.With
on_return_raises=Trueandmandatory=Trueon the result routing keys (bound only by the downstream Spring service, not declared here), publishing before Spring has declared its queue bindings will raise on the AI worker side. Worth confirming deployment/rollout ordering guarantees this, or that the retry path in the consumer tolerates it gracefully.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/infrastructure/rabbitmq/connection.py` around lines 49 - 93, The RabbitMQ setup in the connection initialization flow must ensure result/DLQ publishes are not attempted before the downstream Spring consumer has established its queue bindings. Confirm and enforce the deployment or rollout ordering around the existing exchange and queue declarations, or update the consumer retry path to gracefully handle mandatory publish failures raised by on_return_raises=True.tests/infrastructure/rabbitmq/test_consumer.py (1)
94-120: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winGraceful-shutdown drain is untested.
stop()gained non-trivial logic (subscription cancel, in-flight task snapshot,asyncio.waittimeout branch), but no test here or intest_lifecycle.pyexercises it. Worth adding cases for: pending task completes within the timeout, and timeout reached with tasks still pending.Want me to draft those tests?
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/infrastructure/rabbitmq/test_consumer.py` around lines 94 - 120, The graceful-shutdown behavior of AnalysisRequestConsumer.stop is untested. Add tests using create_consumer that cover both an in-flight task completing within the shutdown timeout and asyncio.wait timing out while tasks remain pending, asserting subscription cancellation and the corresponding drain behavior in each case.app/infrastructure/rabbitmq/consumer.py (1)
58-64: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win
retry_attemptscan grow unboundedly.Entries are only removed on success, DLQ routing, or exhaustion. Messages that are nacked with
requeue=Truebut never redelivered (connection drop, queue purge, broker-side loss) leave a permanent entry. For a long-lived consumer, consider a bounded structure (e.g.OrderedDictwith an LRU cap or a TTL sweep) instead of a plain dict.♻️ Sketch
- self.retry_attempts: dict[str, int] = {} + # bounded LRU so abandoned retries cannot accumulate + self.retry_attempts: OrderedDict[str, int] = OrderedDict() + self.max_tracked_retries = 10_000and when inserting:
self.retry_attempts[event_id] = retry_attempt + 1 self.retry_attempts.move_to_end(event_id) while len(self.retry_attempts) > self.max_tracked_retries: self.retry_attempts.popitem(last=False)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@app/infrastructure/rabbitmq/consumer.py` around lines 58 - 64, Bound retry_attempts growth in the consumer initialization and retry-tracking logic. Replace the unbounded dict with an OrderedDict or equivalent bounded structure, define a suitable max_tracked_retries limit, and when recording a retry update its recency and evict the oldest entries until the limit is respected. Preserve existing removal behavior for successful, DLQ-routed, and exhausted messages.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In @.github/workflows/ai-ci.yml:
- Around line 19-20: Disable persisted checkout credentials in both checkout
steps in .github/workflows/ai-ci.yml at lines 19-20 and 46-47 by setting
persist-credentials to false under each actions/checkout@v4 configuration.
In `@app/analysis/service.py`:
- Around line 133-140: Update the zero-score guard in the analysis service to
account for rules availability: a valid rules-only verdict with rules_available
true must not raise ValueError solely because rule_score is zero. Preserve the
existing failure behavior when no reliable signal is available, using the
availability flags consistently with the scoring engine.
In `@app/analysis/url/analyzer.py`:
- Around line 192-196: Update the is_final_malicious calculation to stop
treating vt_malicious_count >= 1 as independently malicious; rely on the
VirusTotalClient.scan_url result is_vt_malicious, which already applies the
configured threshold, while preserving the Google Safe Browsing and existing
VirusTotal checks.
- Around line 265-277: Remove the unreachable duplicate except Exception handler
from _scan_virustotal, retaining the first handler’s logging and fallback
response unchanged.
In `@app/analysis/url/tracker.py`:
- Around line 31-37: Move the triple-quoted documentation block to the first
statement inside _is_public_host, before the dns_timeout defaulting logic, so
Python recognizes it as the function docstring while preserving the existing
timeout behavior.
In `@app/infrastructure/google_safe_browsing/client.py`:
- Around line 13-16: Update the Safe Browsing client’s api_url construction and
request handling so API keys are not embedded in URLs that may appear in
retryable failure logs. Keep a key-free base URL in the client, pass the key
separately as a query parameter for requests, and ensure exceptions propagated
to http_retry do not contain the key.
In `@app/infrastructure/rabbitmq/consumer.py`:
- Around line 331-342: Update the exception path surrounding the sanitized DLQ
publish and message.nack in the consumer flow to await a short backoff before
requeueing. Keep the existing requeue=True behavior and error logging, ensuring
repeated DLQ failures are throttled while preserving retry handling.
- Around line 266-315: Replace the per-process retry_attempts logic in the
consumer retry/DLQ flow with broker-persisted delivery count state, using
x-death or x-delivery-count as the authoritative attempt count so retries remain
bounded across replicas and restarts. Preserve the existing
PROCESSING_RETRIES_EXHAUSTED DLQ routing and logging behavior, and change the
requeue path around message.nack to apply broker-supported delayed
redelivery/backoff instead of immediate requeueing.
- Around line 176-209: Update the success path in the consumer method containing
handler.handle, result_factory.create, result_publisher.publish, and message.ack
so redelivery cannot publish a duplicate result event for the same
request.eventId/analysisId. Prefer making the result event identity
deterministic or publishing idempotently using those request fields;
alternatively ensure ACK failure is treated as fatal and cannot trigger normal
redelivery. Preserve existing retry and dead-letter handling for processing
failures.
In `@app/infrastructure/rabbitmq/result_factory.py`:
- Around line 160-169: Update _url_detail so UrlAnalysisDetail.errorCode uses
the machine-readable provider error-code field rather than the human-readable
error_message text. Reuse the existing URL payload key containing the stable
provider code, and preserve error_message separately only if the model exposes a
dedicated message field.
---
Nitpick comments:
In `@app/analysis/execution.py`:
- Around line 41-47: Update classify_execution’s text failure classification so
a Gemini-only failure with a valid Naive Bayes score appends "TEXT:GEMINI"
instead of the generic "TEXT" tag. Preserve "TEXT" for cases where both text
engines fail and retain the existing "TEXT:NAIVE_BAYES" behavior when Naive
Bayes fails.
In `@app/analysis/schemas.py`:
- Around line 40-44: Update the Korean rationale comment above the llm,
hybrid_url, and rules fields to explain that dynamic availability-based weight
renormalization can make any remaining track approach 100% when others are
unavailable. Remove the outdated two-scenario URL/LLM/rules redistribution
explanation, while keeping the existing ge=0 and le=100 constraints unchanged.
In `@app/analysis/url/analyzer.py`:
- Around line 27-30: Update the class-level AVAILABLE_STATUSES constant to be
annotated as ClassVar[frozenset[str]], import ClassVar from typing, and ensure
its value remains an immutable frozenset to satisfy Ruff RUF012.
- Around line 223-265: Extract the duplicated provider-call and
unavailable-response logic from _scan_gsb and _scan_virustotal into a shared
helper such as _call_provider(client, name). Have the helper invoke the supplied
client, log the provider-specific failure, and return the common fallback
payload; update both scan methods to delegate to it while preserving their
existing client and provider names.
In `@app/analysis/url/tracker.py`:
- Around line 181-201: Update the retry lambdas in the URL trace flow to bind
the current iteration’s current_url and headers as default arguments, for both
the HEAD request and the GET fallback. Preserve the existing
client.head/client.get calls and retry behavior while eliminating Ruff B023
late-binding warnings.
In `@app/infrastructure/http_retry.py`:
- Around line 34-68: Update the retry loop around the network-error and
retryable-status branches to await a backoff helper before each retry continue.
Implement or reuse _sleep_before_retry(attempt) with exponential backoff and
jitter, and pass the 429 response’s Retry-After header in the status branch so
that server guidance is honored when available.
In `@app/infrastructure/rabbitmq/connection.py`:
- Around line 49-93: The RabbitMQ setup in the connection initialization flow
must ensure result/DLQ publishes are not attempted before the downstream Spring
consumer has established its queue bindings. Confirm and enforce the deployment
or rollout ordering around the existing exchange and queue declarations, or
update the consumer retry path to gracefully handle mandatory publish failures
raised by on_return_raises=True.
In `@app/infrastructure/rabbitmq/consumer.py`:
- Around line 58-64: Bound retry_attempts growth in the consumer initialization
and retry-tracking logic. Replace the unbounded dict with an OrderedDict or
equivalent bounded structure, define a suitable max_tracked_retries limit, and
when recording a retry update its recency and evict the oldest entries until the
limit is respected. Preserve existing removal behavior for successful,
DLQ-routed, and exhausted messages.
In `@app/infrastructure/rabbitmq/publisher.py`:
- Around line 61-71: Update AnalysisResultPublisher.publish() around its
asyncio.wait_for call to classify failures like DeadLetterPublisher.publish():
convert asyncio.TimeoutError into RetryableProcessingError with the
timeout-specific failure_code, and convert other publish exceptions into
RetryableProcessingError with the generic publish failure_code. Preserve the
existing publish arguments and timeout configuration.
In `@app/infrastructure/rabbitmq/result_factory.py`:
- Around line 187-190: Update _integer_score to remove the redundant int()
wrapper around round(float(value)), preserving the existing clamping behavior
and return type.
In `@app/infrastructure/rabbitmq/schemas.py`:
- Around line 121-127: Update AnalysisResultPayload to type riskGrade and
phishingType with the existing RiskGrade enum from app.analysis.schemas instead
of unconstrained strings, while preserving their optional nullable defaults and
existing validation behavior.
- Around line 121-138: Update AnalysisResultPayload with model-level validation
that enforces mutually exclusive success and failure states: completed score
fields (finalScore/riskGrade) must not coexist with failure indicators
(failureCode/failedTracks). Preserve valid success-only, failure-only, and
existing optional analysis-field payloads while rejecting mixed states.
- Around line 154-187: Strengthen validate_event_result in AnalysisResultEvent
to reject state-incompatible payload fields, not just missing required fields:
COMPLETED must not include failureCode or non-empty failedTracks, and
PARTIAL/FAILED must likewise reject fields belonging to other event states.
Preserve the existing required-field checks and FAILED riskGrade restriction
while enforcing mutually exclusive payload combinations for every
AnalysisEventType.
In `@Dockerfile`:
- Around line 17-18: Update the Dockerfile healthcheck to use a dependency-aware
readiness endpoint that verifies RabbitMQ and analysis-consumer state, or
explicitly document and configure the existing root check as liveness-only if it
is not used for readiness or traffic routing.
In `@tests/analysis/test_execution.py`:
- Around line 1-108: Add focused tests for classify_execution covering text
analysis failures that produce the expected TEXT or TEXT:NAIVE_BAYES
failed-track identifiers, and rule_analysis failures that produce RULES. Assert
each scenario’s COMPLETED/PARTIAL status and failed_tracks, while preserving the
existing URL and pipeline error coverage.
In `@tests/infrastructure/rabbitmq/test_consumer.py`:
- Around line 94-120: The graceful-shutdown behavior of
AnalysisRequestConsumer.stop is untested. Add tests using create_consumer that
cover both an in-flight task completing within the shutdown timeout and
asyncio.wait timing out while tasks remain pending, asserting subscription
cancellation and the corresponding drain behavior in each case.
In `@tests/infrastructure/rabbitmq/test_result_factory.py`:
- Around line 102-159: Extend the result-factory tests around
AnalysisResultEventFactory.create to cover a result with stage1_naive_bayes
absent and Gemini-only text analysis, asserting the intended TextAnalysisMethod
behavior, and add a case where url_risk_score is exactly 1, asserting the
expected 0–100 score mapping. Reuse the existing request/result helpers and
preserve current successful and failed execution coverage.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: a36d21f9-168a-430d-b67d-86cf911314b6
📒 Files selected for processing (34)
.dockerignore.github/workflows/ai-ci.ymlDockerfileapp/analysis/execution.pyapp/analysis/schemas.pyapp/analysis/scoring.pyapp/analysis/service.pyapp/analysis/text/gemini_analyzer.pyapp/analysis/url/analyzer.pyapp/analysis/url/tracker.pyapp/core/config.pyapp/infrastructure/errors.pyapp/infrastructure/gemini/client.pyapp/infrastructure/google_safe_browsing/client.pyapp/infrastructure/http_retry.pyapp/infrastructure/rabbitmq/connection.pyapp/infrastructure/rabbitmq/consumer.pyapp/infrastructure/rabbitmq/dead_letter.pyapp/infrastructure/rabbitmq/handler.pyapp/infrastructure/rabbitmq/publisher.pyapp/infrastructure/rabbitmq/result_factory.pyapp/infrastructure/rabbitmq/schemas.pyapp/infrastructure/virustotal/client.pyapp/main.pytests/analysis/test_execution.pytests/analysis/test_scoring.pytests/analysis/url/test_analyzer.pytests/infrastructure/rabbitmq/test_connection.pytests/infrastructure/rabbitmq/test_consumer.pytests/infrastructure/rabbitmq/test_dead_letter.pytests/infrastructure/rabbitmq/test_handler.pytests/infrastructure/rabbitmq/test_lifecycle.pytests/infrastructure/rabbitmq/test_result_factory.pytests/infrastructure/rabbitmq/test_result_schemas.py
💤 Files with no reviewable changes (1)
- app/analysis/text/gemini_analyzer.py
| try: | ||
| await self.handler.handle(event) | ||
| except Exception: | ||
| await self._handle_processing_failure( | ||
| result = await self.handler.handle(event) | ||
|
|
||
| execution = classify_execution(result) | ||
|
|
||
| result_event = self.result_factory.create( | ||
| request=event, | ||
| execution=execution, | ||
| ) | ||
|
|
||
| await self.result_publisher.publish( | ||
| result_event | ||
| ) | ||
| except NonRetryableProcessingError as exception: | ||
| await self._route_to_dead_letter( | ||
| message=message, | ||
| event=event, | ||
| failure_code=exception.failure_code, | ||
| ) | ||
| return | ||
| except Exception as exception: | ||
| await self._handle_retryable_failure( | ||
| message=message, | ||
| event=event, | ||
| exception=exception, | ||
| ) | ||
| return | ||
|
|
||
| self.retry_attempts.pop(str(event.eventId), None) | ||
| self.retry_attempts.pop( | ||
| str(event.eventId), | ||
| None, | ||
| ) | ||
|
|
||
| await message.ack() |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
fd -t f result_factory.py --exec cat -n {}
fd -t f publisher.py -p app/infrastructure/rabbitmq --exec cat -n {}Repository: SafeFam/SafeFam_AI
Length of output: 15255
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '--- consumer.py relevant sections ---\n'
sed -n '120,230p' app/infrastructure/rabbitmq/consumer.py | cat -n
printf '\n--- result schema fields ---\n'
fd -t f schemas.py -p app/infrastructure/rabbitmq --exec sed -n '1,220p' {} \;
printf '\n--- consumers/downstream references to result fields ---\n'
rg -n "AnalysisResultEvent|result.*Event|eventId|causationId|analysisId|correlation_id|message_id|RABBITMQ_ANALYSIS" app tests 2>/dev/null || trueRepository: SafeFam/SafeFam_AI
Length of output: 247
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- consumer.py relevant sections ---'
sed -n '120,230p' app/infrastructure/rabbitmq/consumer.py | cat -n
printf '%s\n' ''
printf '%s\n' '--- result schema fields ---'
fd -t f schemas.py -p app/infrastructure/rabbitmq --exec sed -n '1,240p' {} \;
printf '%s\n' ''
printf '%s\n' '--- consumers/downstream references to result fields ---'
rg -n "AnalysisResultEvent|result.*Event|eventId|causationId|analysisId|correlation_id|message_id|RABBITMQ_ANALYSIS" app tests 2>/dev/null || trueRepository: SafeFam/SafeFam_AI
Length of output: 23552
Avoid publishing duplicate result events after successful ACK.
result_publisher.publish(...) is sent before message.ack(), so if ACK delivery fails or the process dies after publish, the queue redelivers and AnalysisResultEventFactory.create(...) generates a fresh result eventId for the same request.eventId/analysisId payload. Either make ack failure fatal to avoid redelivery, or publish retry-safe/idempotent result events keyed by request fields downstream.
🧰 Tools
🪛 Ruff (0.16.0)
[warning] 196-196: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@app/infrastructure/rabbitmq/consumer.py` around lines 176 - 209, Update the
success path in the consumer method containing handler.handle,
result_factory.create, result_publisher.publish, and message.ack so redelivery
cannot publish a duplicate result event for the same request.eventId/analysisId.
Prefer making the result event identity deterministic or publishing idempotently
using those request fields; alternatively ensure ACK failure is treated as fatal
and cannot trigger normal redelivery. Preserve existing retry and dead-letter
handling for processing failures.
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@app/infrastructure/http_retry.py`:
- Around line 36-40: Update the retry-after parsing in the visible retry delay
logic to support HTTP-date strings using email.utils.parsedate_to_datetime,
converting the parsed date into a nonnegative delay relative to the current
time. Preserve the existing numeric-seconds handling and invalid-value fallback,
while ensuring both supported formats honor the provider’s retry window.
In `@tests/analysis/test_execution.py`:
- Around line 131-151: Extend classify_execution so a Naive Bayes result with
grade="UNKNOWN" is classified as failed even when error_message is None,
matching the existing stage1_failed behavior. Preserve the current PARTIAL
status and failed_tracks entry for valid Gemini results, and add coverage for
the missing-error-code case alongside
test_classifies_naive_bayes_failure_with_valid_gemini.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 325bf32e-713f-4d34-8c2f-2e92b2f30d4d
📒 Files selected for processing (17)
.github/workflows/ai-ci.ymlapp/analysis/execution.pyapp/analysis/service.pyapp/analysis/url/analyzer.pyapp/analysis/url/tracker.pyapp/core/config.pyapp/infrastructure/google_safe_browsing/client.pyapp/infrastructure/http_retry.pyapp/infrastructure/rabbitmq/consumer.pyapp/infrastructure/rabbitmq/publisher.pyapp/infrastructure/rabbitmq/result_factory.pyapp/infrastructure/rabbitmq/schemas.pytests/analysis/test_execution.pytests/analysis/url/test_analyzer.pytests/infrastructure/rabbitmq/test_consumer.pytests/infrastructure/rabbitmq/test_publisher.pytests/infrastructure/rabbitmq/test_result_factory.py
🚧 Files skipped from review as they are similar to previous changes (11)
- app/infrastructure/rabbitmq/publisher.py
- tests/analysis/url/test_analyzer.py
- app/core/config.py
- app/infrastructure/google_safe_browsing/client.py
- app/analysis/url/analyzer.py
- app/analysis/url/tracker.py
- app/infrastructure/rabbitmq/result_factory.py
- app/analysis/service.py
- .github/workflows/ai-ci.yml
- app/analysis/execution.py
- tests/infrastructure/rabbitmq/test_consumer.py
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@Dockerfile`:
- Around line 26-27: Fix the Dockerfile CMD declaration by using valid multiline
JSON-array syntax with backslash continuations, or place the entire array on one
line. Ensure all quoted uvicorn arguments remain part of the same CMD
instruction and the image builds successfully.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tests/analysis/test_service.py`:
- Line 144: Update the docstring of
test_analyze_pipeline_uses_available_zero_score_rules_when_text_engines_are_down
to describe the current contract: when the rules track is available but produces
a zero score, the analysis returns SUCCESS with LOW confidence and a score of 0;
do not describe LOW as a fail-open error.
In `@tests/infrastructure/test_http_retry.py`:
- Around line 26-46: Make test_sleep_honors_http_date_retry_after deterministic
by freezing or mocking the clock used by _sleep_before_retry in
app.infrastructure.http_retry, then construct retry_at from that fixed time.
Keep the assertion focused on the expected approximately 20-second delay without
relying on the real system clock.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: af9bd467-eb6e-427f-b76d-e62e298999a0
📒 Files selected for processing (7)
Dockerfileapp/analysis/execution.pyapp/analysis/service.pyapp/infrastructure/http_retry.pytests/analysis/test_execution.pytests/analysis/test_service.pytests/infrastructure/test_http_retry.py
🚧 Files skipped from review as they are similar to previous changes (5)
- app/analysis/execution.py
- app/infrastructure/http_retry.py
- app/analysis/service.py
- Dockerfile
- tests/analysis/test_execution.py
| @patch("app.analysis.service.analyze_text_with_gemini", new_callable=AsyncMock) | ||
| @patch("app.analysis.service.analyze_text_with_naive_bayes", new_callable=AsyncMock) | ||
| async def test_analyze_pipeline_does_not_fail_open_when_both_text_engines_are_down(mock_nb, mock_gemini): | ||
| async def test_analyze_pipeline_uses_available_zero_score_rules_when_text_engines_are_down(mock_nb, mock_gemini): |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Align the docstring with the new test contract.
The renamed test now intentionally expects SUCCESS/LOW/0 when the rules track is available but produces a zero score. The existing docstring still says that returning LOW is a fail-open error, contradicting the assertions and the scoring behavior.
Proposed docstring update
- URL/규칙 신호가 전혀 없는 문자라도 최종 등급이 조용히 LOW로 나와선 안 된다.
- 두 분류기가 동시에 다운됐다는 인프라 장애가 "안전 확인됨"으로 둔갑하면 안 됨 (fail-open 방지).
+ 텍스트 엔진이 다운되어도 규칙 트랙이 가용하고 점수가 0이면
+ 정상적으로 LOW 등급을 반환해야 한다.
+ 모든 분석 트랙이 unavailable한 경우에만 fail-closed 처리한다.📝 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 test_analyze_pipeline_uses_available_zero_score_rules_when_text_engines_are_down(mock_nb, mock_gemini): | |
| async def test_analyze_pipeline_uses_available_zero_score_rules_when_text_engines_are_down(mock_nb, mock_gemini): | |
| """ | |
| 텍스트 엔진이 다운되어도 규칙 트랙이 가용하고 점수가 0이면 | |
| 정상적으로 LOW 등급을 반환해야 한다. | |
| 모든 분석 트랙이 unavailable한 경우에만 fail-closed 처리한다. | |
| """ |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/analysis/test_service.py` at line 144, Update the docstring of
test_analyze_pipeline_uses_available_zero_score_rules_when_text_engines_are_down
to describe the current contract: when the rules track is available but produces
a zero score, the analysis returns SUCCESS with LOW confidence and a score of 0;
do not describe LOW as a fail-open error.
| @pytest.mark.asyncio | ||
| async def test_sleep_honors_http_date_retry_after() -> None: | ||
| retry_at = datetime.now(timezone.utc) + timedelta( | ||
| seconds=20 | ||
| ) | ||
| sleep = AsyncMock() | ||
|
|
||
| with patch( | ||
| "app.infrastructure.http_retry.asyncio.sleep", | ||
| sleep, | ||
| ): | ||
| await _sleep_before_retry( | ||
| attempt=0, | ||
| retry_after=format_datetime( | ||
| retry_at, | ||
| usegmt=True, | ||
| ), | ||
| ) | ||
|
|
||
| delay = sleep.await_args.args[0] | ||
| assert 18.0 <= delay <= 20.0 |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Make the HTTP-date test deterministic.
The test derives retry_at from the real clock and requires the calculated delay to remain within two seconds of 20. A slow CI runner or scheduling pause can make this test fail even when the retry logic is correct. Freeze or mock the clock used by app/infrastructure/http_retry.py.
Proposed fix
async def test_sleep_honors_http_date_retry_after() -> None:
- retry_at = datetime.now(timezone.utc) + timedelta(
- seconds=20
- )
+ now = datetime(2026, 1, 1, tzinfo=timezone.utc)
+ retry_at = now + timedelta(seconds=20)
sleep = AsyncMock()
with (
+ patch(
+ "app.infrastructure.http_retry.datetime"
+ ) as datetime_mock,
patch(
"app.infrastructure.http_retry.asyncio.sleep",
sleep,
),
):
+ datetime_mock.now.return_value = now
await _sleep_before_retry(
attempt=0,
retry_after=format_datetime(
@@
- assert 18.0 <= delay <= 20.0
+ assert delay == pytest.approx(20.0)📝 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.
| @pytest.mark.asyncio | |
| async def test_sleep_honors_http_date_retry_after() -> None: | |
| retry_at = datetime.now(timezone.utc) + timedelta( | |
| seconds=20 | |
| ) | |
| sleep = AsyncMock() | |
| with patch( | |
| "app.infrastructure.http_retry.asyncio.sleep", | |
| sleep, | |
| ): | |
| await _sleep_before_retry( | |
| attempt=0, | |
| retry_after=format_datetime( | |
| retry_at, | |
| usegmt=True, | |
| ), | |
| ) | |
| delay = sleep.await_args.args[0] | |
| assert 18.0 <= delay <= 20.0 | |
| `@pytest.mark.asyncio` | |
| async def test_sleep_honors_http_date_retry_after() -> None: | |
| now = datetime(2026, 1, 1, tzinfo=timezone.utc) | |
| retry_at = now + timedelta(seconds=20) | |
| sleep = AsyncMock() | |
| with ( | |
| patch( | |
| "app.infrastructure.http_retry.datetime" | |
| ) as datetime_mock, | |
| patch( | |
| "app.infrastructure.http_retry.asyncio.sleep", | |
| sleep, | |
| ), | |
| ): | |
| datetime_mock.now.return_value = now | |
| await _sleep_before_retry( | |
| attempt=0, | |
| retry_after=format_datetime( | |
| retry_at, | |
| usegmt=True, | |
| ), | |
| ) | |
| delay = sleep.await_args.args[0] | |
| assert delay == pytest.approx(20.0) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/infrastructure/test_http_retry.py` around lines 26 - 46, Make
test_sleep_honors_http_date_retry_after deterministic by freezing or mocking the
clock used by _sleep_before_retry in app.infrastructure.http_retry, then
construct retry_at from that fixed time. Keep the assertion focused on the
expected approximately 20-second delay without relying on the real system clock.
1 participant
📝 개요
FastAPI AI 서버가 RabbitMQ 분석 요청을 소비하고, 분석 결과를 COMPLETED, PARTIAL, FAILED 이벤트로 분류하여 발행하도록 AI Worker를 구현했습니다.
결과 발행 이후에만 요청을 ACK하며, 복구 불가능한 메시지는 원문을 제외한 정제된 DLQ 이벤트로 처리합니다.
외부 API 재시도와 Graceful Shutdown, Docker Healthcheck 및 GitHub Actions CI도 추가했습니다.
🔗 관련 이슈
🎯 주요 변경 사항
분석 결과 이벤트 계약 정의
결과 이벤트 발행 및 ACK 처리
재시도 및 DLQ 처리
외부 API 안정성 강화
Graceful Shutdown 구현
Docker 및 CI 구성
📸 사진
✅ PR 체크리스트
uvicorn구동 또는 테스트 코드)를 통과했습니다.Summary by CodeRabbit
.dockerignore.