fix: only use active consumers

hackmdio · Jan 21, 2025 · d0acec0 · d0acec0
1 parent efbb11f
commit d0acec0
Showing 1 changed file with 7 additions and 6 deletions.
diff --git a/src/api.js b/src/api.js
@@ -385,12 +385,13 @@ export class Worker {
   }
 
   async checkAndRecoverOrphanStreams () {
-    const consumers = (
-      await this.client.redis.xInfoConsumers(
-        this.client.redisWorkerStreamName,
-        this.client.redisWorkerGroupName
-      )
-    ).map(({ name }) => name)
+    const rawConsumers = await this.client.redis.xInfoConsumers(
+      this.client.redisWorkerStreamName,
+      this.client.redisWorkerGroupName
+    )
+    const consumers = rawConsumers
+      .filter((c) => c.pending > 0 || c.inactive < this.client.redisWorkerTimeout)
+      .map(({ name }) => name)
     const leader = consumers.sort()[0]
     if (this.client.consumername !== leader) return