Enqueue inactivity notifications concurrently - #2661
Conversation
📝 WalkthroughWalkthroughThe inactivity notification flow now enqueues notifications with unbounded concurrency. Per-notification error handling remains unchanged. ChangesInactivity notifications
Estimated code review effort: 1 (Trivial) | ~5 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 ESLint
ESLint install failed. For unrecoverable errors, disable the tool in CodeRabbit configuration. 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 |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 9322337f3b
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| ) | ||
| ), | ||
| Effect.all, | ||
| Effect.allWith({concurrency: 'unbounded'}), |
There was a problem hiding this comment.
When the first cadence run returns a large inactive-user backlog, this starts every queue.add operation simultaneously; createFindUsersToNotifyAboutInactivity has no pagination or limit, so the batch can span the entire eligible user population. That creates an equally large number of fibers, promises, and outstanding Redis commands, risking contact-service memory exhaustion or Redis overload. Because the users are marked as notified before this loop and per-entry enqueue failures are swallowed, overload can also suppress reminders until their next cadence step. Use a finite concurrency limit or process the query in bounded batches instead.
Useful? React with 👍 / 👎.
Greptile SummaryThis PR parallelizes inactivity-notification MQ enqueues to reduce the duration of large scheduled runs.
Confidence Score: 4/5The PR appears safe to merge, although the enqueue loop should use a finite concurrency limit to avoid unlimited process and Redis-command pressure during unusually large runs. The change retains functional enqueue behavior and per-entry error isolation, but starts every operation in the unpaginated batch simultaneously rather than preserving bounded backpressure. Files Needing Attention: apps/contact-service/src/services/UserNotificationService.ts Important Files Changed
Reviews (1): Last reviewed commit: "Enqueue inactivity notifications concurr..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| ) | ||
| ), | ||
| Effect.all, | ||
| Effect.allWith({concurrency: 'unbounded'}), |
There was a problem hiding this comment.
If a daily run selects a large number of inactive users, unbounded concurrency allocates every payload encoding, promise, closure, and pending Redis command at once, creating avoidable memory and event-loop pressure; use a finite concurrency limit to retain parallelism with backpressure.
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
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 `@apps/contact-service/src/services/UserNotificationService.ts`:
- Line 426: Limit the enqueue work in notifyUsersAboutInactivity by replacing
the unbounded Effect.allWith concurrency with a bounded concurrency or batching
strategy before enqueueUserNotification calls. Preserve the existing per-user
notification behavior while preventing Queue.add from overwhelming client,
Redis, or BullMQ resources.
🪄 Autofix
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: e75c5ef9-68cd-4df3-bcf9-51f5acb5f080
📒 Files selected for processing (1)
apps/contact-service/src/services/UserNotificationService.ts
| ) | ||
| ), | ||
| Effect.all, | ||
| Effect.allWith({concurrency: 'unbounded'}), |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
rg -n -C 8 \
'\bEnqueueUserNotification\b|\benqueueUserNotification\s*=' \
apps/contact-service packages/server-utils
rg -n -C 8 \
'notifyUsersAboutInactivity|scheduledTaskWorkers|Schedule|repeat|cron|interval|Semaphore|Queue|rate.?limit' \
apps/contact-service packages/server-utils
rg -n -C 8 \
'findUsersToNotifyAboutInactivity|LIMIT|batch|chunk|concurrency' \
apps/contact-service packages/server-utilsRepository: vexl-it/vexl
Length of output: 50369
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
# Inspect the exact producer implementation and scheduled task around the warning.
sed -n '1,150p' packages/server-utils/src/mqService.ts
printf '\n--- schedule task around inactivity worker ---\n'
rg -n -C 12 'notifyUsersAboutInactivity|schedule.*Inactivity|inactivity|intervalMs|cronPattern|limit' apps/contact-service/packages/contact-service apps/contact-service/src packages/server-utils/src || true
printf '\n--- UserNotificationService relevant lines ---\n'
sed -n '340,440p' apps/contact-service/src/services/UserNotificationService.ts
sed -n '1,130p' apps/contact-service/src/scheduledTaskWorkers.tsRepository: vexl-it/vexl
Length of output: 50369
🏁 Script executed:
#!/usr/bin/env bash
set -euo pipefail
printf '%s\n' '--- packages/server-utils/src/mqService.ts ---'
sed -n '1,140p' packages/server-utils/src/mqService.ts
printf '%s\n' '--- apps/contact-service/src/scheduledTaskWorkers.ts ---'
sed -n '1,160p' apps/contact-service/src/scheduledTaskWorkers.ts
printf '%s\n' '--- UserNotificationService enqueue lines ---'
sed -n '350,435p' apps/contact-service/src/services/UserNotificationService.ts
printf '%s\n' '--- focused config/search for inactivity concurrency/interval ---'
rg -n -i -C 6 'inactivity|notifyUsersAboutInactivity|intervalMs|cronPattern|concurrency|maxJobRuntime|stallCheck|connection|Queue|latency' apps/contact-service src packages/server-utils/src --glob '!**/node_modules/**' --glob '!**/*.test.ts' --glob '!**/dist/**' | sed -n '1,220p'Repository: vexl-it/vexl
Length of output: 29042
Limit inactivity notification enqueues before adding jobs.
notifyUsersAboutInactivity scans all due inactive users and uses Effect.allWith({concurrency: 'unbounded'}) to enqueue one BullMQ job per user. enqueueUserNotification creates jobs directly via Queue.add without concurrency or rate-limiting protection, so a large backlog can saturate client/Redis/MQ resources while previously recorded sends are already marked. Use a bounded concurrency limit or bounded queue for these enqueues, or batch them before enqueueing.
🤖 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 `@apps/contact-service/src/services/UserNotificationService.ts` at line 426,
Limit the enqueue work in notifyUsersAboutInactivity by replacing the unbounded
Effect.allWith concurrency with a bounded concurrency or batching strategy
before enqueueUserNotification calls. Preserve the existing per-user
notification behavior while preventing Queue.add from overwhelming client,
Redis, or BullMQ resources.
Source: MCP tools
📱 Preview on the staging appOpen Vexl (stage) → Account → Scan QR code and scan this: Preview link:
The preview only loads into staging builds with a matching runtime — if this PR changes native code, ship a new staging build first. To go back to the staging channel: debug screen → "Clear PR preview". |
1 participant
The inactivity-notification enqueue loop in
notifyUsersAboutInactivityranEffect.allwith the default sequential concurrency, so each MQ entry was enqueued one round-trip at a time. The first run after deploying the new inactivity cadence can select every long-inactive user at once, which would serialize thousands of enqueues for no reason.Switch to
Effect.allWith({concurrency: 'unbounded'}), matching theNewUserNotificationMqEntryenqueue loop earlier in the same file. Per-entry failures are already caught and logged individually, so concurrent execution doesn't change error behavior.Summary by CodeRabbit