ack each message individually

1 files changed, 2 insertions, 2 deletions

diff --git a/worker.js b/worker.js
index b277eca..4fcb0bf 100644
--- a/worker.js
+++ b/worker.js
@@ -209,11 +209,11 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost",
         } catch (err) {
             // this should only happen for deadlocks or non-data related issues
             console.error(err);
-            await ch.nack(msgs.pop(), true, true);  // nack all messages until the last and requeue
+            await Promise.all(msgs.map((m) => ch.nack(m, true)) );  // requeue
             return;  // give up
         }
         console.log("acking batch");
-        await ch.ack(msgs.pop(), true);  // ack all messages until the last
+        await Promise.all(msgs.map((m) => ch.ack(m)) );
 
         // notify analyzer
         Promise.all(participant_ext_records.map(async (p) =>