From 53cb0b196a54f5335d4a9056bcd539a6f80d35e4 Mon Sep 17 00:00:00 2001
From: Stavros Michalovits <stmichalovits@gmail.com>
Date: Wed, 4 Nov 2026 11:35:01 +0000
Subject: [PATCH] Separate remaining and acked counters for race in rmqperftest

---
 examples/rmqperftest/rmqperftest_runner.cpp | 17 ++++++++++++-----
 1 file changed, 12 insertions(+), 5 deletions(-)

diff --git a/examples/rmqperftest/rmqperftest_runner.cpp b/examples/rmqperftest/rmqperftest_runner.cpp
index dcc79008..c0e01f37 100644
--- a/examples/rmqperftest/rmqperftest_runner.cpp
+++ b/examples/rmqperftest/rmqperftest_runner.cpp
@@ -52,7 +52,8 @@ class ConsumerCallback {
                      const bsl::shared_ptr<bsls::AtomicBool>& finished)
     : d_args(args)
     , d_finished(finished)
-    , d_remainingCount(bsl::make_shared<bsls::AtomicUint64>())
+    , d_remainingCount(bsl::make_shared<bsls::AtomicInt64>())
+    , d_ackedCount(bsl::make_shared<bsls::AtomicInt64>())
     , d_limitCount()
     {
         if (d_args.consumerMessageCount > 0) {
@@ -63,12 +64,17 @@ class ConsumerCallback {
 
     void operator()(rmqp::MessageGuard& guard)
     {
-        if (d_limitCount && *d_remainingCount == 0) {
-            *d_finished = true;
+        if (!d_limitCount) {
+            guard.ack();
             return;
         }
+
+        if (--*d_remainingCount < 0) {
+            return;
+        }
+
         guard.ack();
-        if (d_limitCount && --*d_remainingCount == 0) {
+        if (++*d_ackedCount == d_args.consumerMessageCount) {
             *d_finished = true;
         }
     }
@@ -76,7 +82,8 @@ class ConsumerCallback {
   private:
     ConsumerArgs d_args;
     bsl::shared_ptr<bsls::AtomicBool> d_finished;
-    bsl::shared_ptr<bsls::AtomicUint64> d_remainingCount;
+    bsl::shared_ptr<bsls::AtomicInt64> d_remainingCount;
+    bsl::shared_ptr<bsls::AtomicInt64> d_ackedCount;
     bool d_limitCount;
 };