Conversation
Contributor
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthroughIntroduce per-bot queue routing by threading Changes
Sequence Diagram(s)sequenceDiagram
participant App as Bot App
participant QC as queue_client
participant Redis as Redis
participant Task as worker task
participant Outbox as Outbox service
participant OC as outbox_consumer
participant Sender as message_sender
App->>QC: init(bot_id)
QC->>Redis: create pool
QC->>QC: store _bot_id
App->>QC: enqueue_scrape(url, chat_id, message_id?)
QC->>Redis: enqueue_job("scrape_and_enrich", bot_id=_bot_id, ...)
Redis-->>Task: deliver job
Task->>Outbox: push(job_id, chat_id, metadata..., message_id?, bot_id=...)
Outbox->>Redis: LPUSH to f"outbox:{bot_id}" or base key
OC->>OC: start(bot_id)
OC->>Redis: BRPOP from f"outbox:{bot_id}"
Redis-->>OC: payload
OC->>Sender: send_item_message(data, chat_id, message_id=payload.message_id)
Sender->>Sender: resolve reply_to from message or message_id
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 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 |
Codecov Report❌ Patch coverage is
@@ Coverage Diff @@
## main #70 +/- ##
=======================================
Coverage ? 71.77%
=======================================
Files ? 61
Lines ? 3082
Branches ? 0
=======================================
Hits ? 2212
Misses ? 870
Partials ? 0
🚀 New features to boost your workflow:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
apps/telegram-bot/core/services/message_sender.py (1)
55-56: Use explicit| Nonetype hints per PEP 484.Static analysis flags implicit
Optionalusage. Update the type hints to be explicit:♻️ Proposed fix
async def send_item_message( - data: dict, chat_id: Union[int, str] = None, message: Message = None, - message_id: int = None, + data: dict, chat_id: Union[int, str] | None = None, message: Message | None = None, + message_id: int | None = None, ) -> None:🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@apps/telegram-bot/core/services/message_sender.py` around lines 55 - 56, Update the function signature in message_sender.py to use explicit union-with-None type hints per PEP 484: change chat_id: Union[int, str] = None to chat_id: Union[int, str] | None = None, change message: Message = None to message: Message | None = None, and change message_id: int = None to message_id: int | None = None so the parameters explicitly allow None; keep data: dict unchanged unless it should also be optional.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@apps/async-worker/async_worker/main.py`:
- Around line 50-52: The current config enables retry_jobs=True with max_tries=3
which can produce duplicate outbox entries because outbox.push() (in
services/outbox.py) does a plain lpush and the outbox consumer
(apps/telegram-bot/core/services/outbox_consumer.py) only logs job_id without
deduplication; fix by adding idempotency: modify outbox.push() to atomically
check-and-set a Redis SET (or Redis SETNX key per job_id) before performing
lpush so the same job_id is not pushed twice, or alternatively set
retry_jobs=False in main.py (disable retry_jobs / adjust max_tries) and
implement manual retries that use the same idempotency guard, or implement
deduplication in outbox_consumer.py by ignoring payloads whose job_id is already
recorded in a Redis set; update references to retry_jobs, max_tries,
outbox.push(), and the consumer's job_id handling accordingly.
In `@apps/telegram-bot/core/handlers/url_process.py`:
- Line 70: The current expression `message_id or message.message_id if message
else None` suffers from operator precedence and will yield None when `message`
is None even if `message_id` is provided; update it to explicitly prefer an
explicit `message_id` and fall back to `message.message_id` only when `message`
exists, e.g. replace with `message_id if message_id is not None else
(message.message_id if message else None)` so `message_id` is preserved when
set; adjust the assignment around the existing `message_id`/`message.message_id`
usage in url_process.py.
---
Nitpick comments:
In `@apps/telegram-bot/core/services/message_sender.py`:
- Around line 55-56: Update the function signature in message_sender.py to use
explicit union-with-None type hints per PEP 484: change chat_id: Union[int, str]
= None to chat_id: Union[int, str] | None = None, change message: Message = None
to message: Message | None = None, and change message_id: int = None to
message_id: int | None = None so the parameters explicitly allow None; keep
data: dict unchanged unless it should also be optional.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: b0e8aad0-514f-4cb3-a1cd-1d4719e640d8
📒 Files selected for processing (14)
apps/async-worker/async_worker/main.pyapps/async-worker/async_worker/services/outbox.pyapps/async-worker/async_worker/tasks/scrape.pyapps/telegram-bot/core/handlers/url_process.pyapps/telegram-bot/core/queue_client.pyapps/telegram-bot/core/services/bot_app.pyapps/telegram-bot/core/services/message_sender.pyapps/telegram-bot/core/services/outbox_consumer.pytests/unit/async_worker/conftest.pytests/unit/async_worker/test_enrichment.pytests/unit/async_worker/test_outbox.pytests/unit/async_worker/test_scrape_task.pytests/unit/telegram_bot/test_outbox_consumer.pytests/unit/telegram_bot/test_queue_client.py
Comment on lines
+50
to
+52
| # Retry jobs on transient failures (e.g. Redis timeout after task completes) | ||
| retry_jobs = True | ||
| max_tries = 3 |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Search for deduplication or job_id tracking in outbox consumer
rg -n -C3 'job_id|dedupe|duplicate|idempoten' apps/telegram-bot/core/services/outbox_consumer.pyRepository: aturret/FastFetchBot
Length of output: 1128
🏁 Script executed:
#!/bin/bash
# 1. Read the outbox.push() function to confirm it uses simple lpush
echo "=== Checking outbox.push() implementation ==="
sed -n '19,42p' apps/async-worker/async_worker/services/outbox.py
# 2. Check the full main.py context around lines 50-52
echo -e "\n=== Checking main.py retry settings context ==="
sed -n '40,65p' apps/async-worker/async_worker/main.py
# 3. Search for any SET-based or job_id tracking in outbox or redis logic
echo -e "\n=== Searching for idempotency mechanisms ==="
rg -n 'SET|setnx|set_if_not_exists|job_id.*track' apps/async-worker/async_worker/services/ -A2Repository: aturret/FastFetchBot
Length of output: 1635
Duplicate outbox entries possible with retry_jobs=True.
Enabling retry_jobs=True with max_tries=3 can cause duplicate messages. The outbox.push() function (apps/async-worker/async_worker/services/outbox.py) performs a simple lpush without idempotency checks. If a task succeeds (pushes to outbox) but crashes before ARQ marks the job complete, the retry will push a duplicate entry that the consumer will process.
The outbox consumer (apps/telegram-bot/core/services/outbox_consumer.py) extracts the job_id from the payload but uses it only for logging, with no deduplication logic.
Fix by:
- Adding idempotency to
outbox.push()using a Redis SET to track processed job IDs, or - Using
retry_jobs=Falseand implementing manual retry logic with idempotency guards, or - Accepting duplicates and implementing deduplication in the consumer
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@apps/async-worker/async_worker/main.py` around lines 50 - 52, The current
config enables retry_jobs=True with max_tries=3 which can produce duplicate
outbox entries because outbox.push() (in services/outbox.py) does a plain lpush
and the outbox consumer (apps/telegram-bot/core/services/outbox_consumer.py)
only logs job_id without deduplication; fix by adding idempotency: modify
outbox.push() to atomically check-and-set a Redis SET (or Redis SETNX key per
job_id) before performing lpush so the same job_id is not pushed twice, or
alternatively set retry_jobs=False in main.py (disable retry_jobs / adjust
max_tries) and implement manual retries that use the same idempotency guard, or
implement deduplication in outbox_consumer.py by ignoring payloads whose job_id
is already recorded in a Redis set; update references to retry_jobs, max_tries,
outbox.push(), and the consumer's job_id handling accordingly.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by CodeRabbit
New Features
Improvements
Tests