From f25d77668bddf1c8c4e3062da361c4b6f45fd12e Mon Sep 17 00:00:00 2001 From: Ayush Jhanwar Date: Thu, 9 Jul 2026 19:23:10 +0530 Subject: [PATCH] fix(worker): await GroupMQ worker.run() to prevent silent shard-stuck 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 #412 --- apps/worker/src/boot-workers.ts | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/apps/worker/src/boot-workers.ts b/apps/worker/src/boot-workers.ts index 3ca6b7ef3..d3451bb50 100644 --- a/apps/worker/src/boot-workers.ts +++ b/apps/worker/src/boot-workers.ts @@ -167,7 +167,17 @@ export function bootWorkers() { worker.on('completed', markEventsActivity); worker.on('drained', markEventsActivity); - worker.run(); + // Fail loud on startup — silent stuck shard otherwise. + // Runtime errors are handled by the shared workers.forEach listener below. + worker.run().catch((err) => { + logger.error( + { shard: index, queueName, err }, + 'Worker startup failed — exiting', + ); + // setTimeout+unref to let the logger flush before exit (matches the + // pattern used by uncaughtException/unhandledRejection handlers below). + setTimeout(() => process.exit(1), 1000).unref(); + }); workers.push(worker); logger.info({ concurrency }, `Started worker for ${queueName}`); }