neorados: Actually enforce notification queue limit

We were adding an overflow marker on every message above capacity, not
just the first, vitiating the purpose of the bound. The shame, the
shame.

Signed-off-by: Adam C. Emerson <aemerson@redhat.com>
(cherry picked from commit 4cc2dbe316be9d8ac63ee00a6aff5455ecb4cd86)
Fixes: https://tracker.ceph.com/issues/76434
Signed-off-by: Adam C. Emerson <aemerson@redhat.com>

diff --git a/src/neorados/RADOS.cc b/src/neorados/RADOS.cc
index ef27c0779c6990dfeeb88861130e14b652713971..8716ca10747a2b9f73de9004b776c6a23ffebd18 100644 (file)
--- a/src/neorados/RADOS.cc
+++ b/src/neorados/RADOS.cc
@@ -1496,7 +1496,9 @@ public:
     } else if (capacity && notifications.size() >= capacity) {
       // We are allowed one over, so the client knows where in the
       // sequence of notifications we started losing data.
-      notifications.push({errc::notification_overflow, {}});
+      if (notifications.size() == capacity) {
+        notifications.push({errc::notification_overflow, {}});
+      }
     } else {
       notifications.push({{},
                          Notification{

diff --git a/src/test/neorados/watch_notify.cc b/src/test/neorados/watch_notify.cc
index 2e86af8d717ad8177a6d1360a16575e2e5f2b43b..574720375feaacb5392450e4eb76c102bfed1bb6 100644 (file)
--- a/src/test/neorados/watch_notify.cc
+++ b/src/test/neorados/watch_notify.cc
@@ -9,8 +9,8 @@
  *
  */
 
+#include <boost/asio/detached.hpp>
 #include <boost/system/detail/errc.hpp>
-#include <coroutine>
 #include <cstdint>
 #include <iostream>
 #include <utility>
@@ -35,8 +35,6 @@
 
 #include "gtest/gtest.h"
 
-using std::uint64_t;
-
 namespace asio = boost::asio;
 namespace buffer = ceph::buffer;
 namespace container = boost::container;
@@ -45,7 +43,6 @@ namespace sys = boost::system;
 using namespace std::literals;
 
 using neorados::ReadOp;
-using neorados::WriteOp;
 
 using std::uint64_t;
 
@@ -233,6 +230,31 @@ CORO_TEST_F(NeoRadosWatchNotifyPoll, WatchNotify, NeoRadosTest) {
   co_return;
 }
 
+CORO_TEST_F(NeoRadosWatchNotifyPoll, WatchNotifyOverflow, NeoRadosTest) {
+  static constexpr auto oid = "obj"sv;
+  co_await create_obj(oid);
+  auto handle = co_await rados().watch(oid, pool(), asio::use_awaitable,
+                                       std::nullopt, 1);
+  EXPECT_TRUE(rados().check_watch(handle));
+  // TODO: Write an awaitable future. This should work for testing for now.
+  rados().notify(oid, pool(), {}, 1s, asio::detached);
+  rados().notify(oid, pool(), {}, 1s, asio::detached);
+  rados().notify(oid, pool(), {}, 1s, asio::detached);
+  rados().notify(oid, pool(), {}, 1s, asio::detached);
+  co_await wait_for(1s);
+  co_await rados().next_notification(handle, asio::use_awaitable);
+  rados().notify(oid, pool(), {}, 1s, asio::detached);
+  co_await expect_error_code(rados().next_notification(handle, asio::use_awaitable),
+                             neorados::errc::notification_overflow);
+  rados().notify(oid, pool(), {}, 1s, asio::detached);
+  co_await rados().next_notification(handle, asio::use_awaitable);
+  co_await rados().unwatch(handle, pool(), asio::use_awaitable);
+  std::cout << "Flushing..." << std::endl;
+  co_await rados().flush_watch(asio::use_awaitable);
+  std::cout << "Flushed..." << std::endl;
+  co_return;
+}
+
 CORO_TEST_F(NeoRadosWatchNotifyPoll, WatchNotifyTimeout, NeoRadosTest) {
   static constexpr auto oid = "obj"sv;
   static constexpr auto timeout = 1s;