fix(worker): await GroupMQ worker.run() to prevent silent shard-stuck#413
fix(worker): await GroupMQ worker.run() to prevent silent shard-stuck#413ayushjhanwar-png wants to merge 1 commit into
Conversation
|
|
📝 WalkthroughWalkthrough
ChangesFail-fast Worker Startup
Estimated code review effort: 1 (Trivial) | ~5 minutes 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
apps/worker/src/boot-workers.ts (1)
169-171: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDuplicate
errorlistener on GroupWorker shards.GroupWorker shards are pushed into the
workersarray at line 181, and the loop at lines 294–297 attaches a genericworker.on('error', ...)listener to every worker in that array. As a result, each event shard now has twoerrorlisteners: the specific one added here (lines 169–171) and the generic one (lines 295–297). Runtime errors will be logged twice with different messages and context shapes.Consider either removing the specific listener (the generic one already logs the error, though without
shard/queueName) or skipping GroupWorker shards in the generic loop.🤖 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/worker/src/boot-workers.ts` around lines 169 - 171, The GroupWorker shard error handler is being attached twice, causing duplicate logs from both the shard-specific listener in boot-workers.ts and the generic worker error loop. Fix this by either removing the specific `worker.on('error', ...)` inside the GroupWorker shard setup or excluding GroupWorker instances from the later generic error-registration loop, and keep a single `error` listener per worker using the existing `GroupWorker`, `workers`, and `logger.error` paths.
🤖 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/worker/src/boot-workers.ts`:
- Around line 173-180: The worker startup failure path in boot-workers should
follow the same shutdown pattern used by the existing error handlers. In
worker.run().catch(...) after logger.error, call setShuttingDown(true) before
exiting, and replace the immediate process.exit(1) with a delayed exit via
setTimeout(..., 1000).unref() so the startup failure log can flush reliably. Use
the existing shutdown/error symbols in this file (worker.run, logger.error,
setShuttingDown, process.exit) to align the .catch() behavior with the
uncaughtException and unhandledRejection handlers.
---
Nitpick comments:
In `@apps/worker/src/boot-workers.ts`:
- Around line 169-171: The GroupWorker shard error handler is being attached
twice, causing duplicate logs from both the shard-specific listener in
boot-workers.ts and the generic worker error loop. Fix this by either removing
the specific `worker.on('error', ...)` inside the GroupWorker shard setup or
excluding GroupWorker instances from the later generic error-registration loop,
and keep a single `error` listener per worker using the existing `GroupWorker`,
`workers`, and `logger.error` paths.
🪄 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
Run ID: e46f95bb-d6c6-45da-8558-c16452e269e2
📒 Files selected for processing (1)
apps/worker/src/boot-workers.ts
| // Fail loud on startup — silent stuck shard otherwise. | ||
| worker.run().catch((err) => { | ||
| logger.error( | ||
| { shard: index, queueName, err }, | ||
| 'Worker startup failed — exiting', | ||
| ); | ||
| process.exit(1); | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
process.exit(1) without delay risks losing the startup failure log.
The existing uncaughtException/unhandledRejection handlers in this same file (lines 406–414) use setTimeout(() => process.exit(1), 1000).unref() to give the logger time to flush before exiting. The new .catch() handler calls process.exit(1) immediately after logger.error(). If @openpanel/logger writes asynchronously, the log may never be flushed — leaving the failure silent, which is exactly what this PR aims to fix.
Additionally, setShuttingDown(true) is not called here, unlike every other exit path in the file (lines 399, 408, 413).
🔧 Proposed fix: follow existing exit pattern
worker.run().catch((err) => {
logger.error(
{ shard: index, queueName, err },
'Worker startup failed — exiting',
);
- process.exit(1);
+ setShuttingDown(true);
+ setTimeout(() => process.exit(1), 1000).unref();
});📝 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.
| // Fail loud on startup — silent stuck shard otherwise. | |
| worker.run().catch((err) => { | |
| logger.error( | |
| { shard: index, queueName, err }, | |
| 'Worker startup failed — exiting', | |
| ); | |
| process.exit(1); | |
| }); | |
| worker.run().catch((err) => { | |
| logger.error( | |
| { shard: index, queueName, err }, | |
| 'Worker startup failed — exiting', | |
| ); | |
| setShuttingDown(true); | |
| setTimeout(() => process.exit(1), 1000).unref(); | |
| }); |
🤖 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/worker/src/boot-workers.ts` around lines 173 - 180, The worker startup
failure path in boot-workers should follow the same shutdown pattern used by the
existing error handlers. In worker.run().catch(...) after logger.error, call
setShuttingDown(true) before exiting, and replace the immediate process.exit(1)
with a delayed exit via setTimeout(..., 1000).unref() so the startup failure log
can flush reliably. Use the existing shutdown/error symbols in this file
(worker.run, logger.error, setShuttingDown, process.exit) to align the .catch()
behavior with the uncaughtException and unhandledRejection handlers.
worker.run() was fire-and-forget: no await, no .catch(). If the underlying
redis.duplicate() blocking-client setup silently fails for a specific shard
(e.g., a transient Redis blip during a rolling-deploy connection burst), the
promise rejects with no handler and the worker enters GroupMQ's exponential
backoff sleep forever. The pod appears healthy, "Started worker for events_N"
is logged, but that shard never consumes.
Real production incident: 3 of 24 event shards silently stopped consuming
after a rolling deploy; 570K jobs accumulated in Redis over ~2h before it
was noticed. Zero error logs anywhere. Manual restart of the StatefulSet
resolved it (fresh redis.duplicate() succeeded on the second attempt).
The fix mirrors the pattern already used for the Kafka consumer startup
right below (which properly .catches errors from startKafkaEventsConsumer).
Also adds a worker.on('error', ...) listener so runtime errors on the
blocking client (BZPOPMIN etc.) after startup are visible.
Fixes Openpanel-dev#412
ayushjhanwar-png
force-pushed
the
upstream-fix/worker-await-run
branch
from
94e5a6c to
f25d776
Compare
There was a problem hiding this comment.
♻️ Duplicate comments (1)
apps/worker/src/boot-workers.ts (1)
172-180: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winCall
setShuttingDown(true)before scheduling exit.The delay fix from the previous review was applied, but
setShuttingDown(true)is still missing. During the 1-second window beforeprocess.exit(1), the/healthz/readyprobe checksisShuttingDown()and will return 200 if it's still false — Kubernetes won't know the pod is failing until the process actually exits. Every other exit path in this file (lines 399, 408, 413) callssetShuttingDown(true)first.🔧 Proposed fix
worker.run().catch((err) => { logger.error( { shard: index, queueName, err }, 'Worker startup failed — exiting', ); + setShuttingDown(true); // setTimeout+unref to let the logger flush before exit (matches the // pattern used by uncaughtException/unhandledRejection handlers below). setTimeout(() => process.exit(1), 1000).unref(); });🤖 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/worker/src/boot-workers.ts` around lines 172 - 180, In worker.run().catch in boot-workers.ts, the startup-failure path still schedules process.exit(1) without first marking the process as shutting down, which leaves /healthz/ready reporting healthy during the delay. Update this catch block to call setShuttingDown(true) before the setTimeout exit, matching the other exit paths in this file and ensuring isShuttingDown() flips immediately when startup fails.
🤖 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.
Duplicate comments:
In `@apps/worker/src/boot-workers.ts`:
- Around line 172-180: In worker.run().catch in boot-workers.ts, the
startup-failure path still schedules process.exit(1) without first marking the
process as shutting down, which leaves /healthz/ready reporting healthy during
the delay. Update this catch block to call setShuttingDown(true) before the
setTimeout exit, matching the other exit paths in this file and ensuring
isShuttingDown() flips immediately when startup fails.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 006e429b-e08f-4be2-a4e0-e781146a0e9c
📒 Files selected for processing (1)
apps/worker/src/boot-workers.ts
2 participants
Fixes #412.
worker.run()was fire-and-forget in the GroupMQ event-worker loop. If the underlyingredis.duplicate()blocking-client setup silently fails for a specific shard, the promise rejects with no handler and the worker enters GroupMQ's exponential backoff sleep forever — pod appears healthy,"Started worker for events_N"is logged, but that shard never consumes.Real incident: 3 of 24 event shards silently stopped consuming after a rolling deploy; 570K jobs accumulated in Redis over ~2h before it was noticed. Zero error logs. Manual restart of the StatefulSet resolved it (fresh
redis.duplicate()succeeded on retry).The fix mirrors what's already done for the Kafka consumer startup right below (which
.catcheserrors fromstartKafkaEventsConsumer).Also adds a
worker.on('error', ...)listener so runtime errors on the blocking client (BZPOPMIN etc.) after startup are visible.Summary by CodeRabbit