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>

diff --git a/src/neorados/RADOS.cc b/src/neorados/RADOS.cc
index 3cd8fcc50b09d218e1c10c7cb01c331e80fa9f2d..f4f2686b2c6899a279948604dfc0d76e3d1db305 100644 (file)
--- a/src/neorados/RADOS.cc
+++ b/src/neorados/RADOS.cc
@@ -1486,7 +1486,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 c567a5b3e226d896af4007692b4bbafc9f2baf33..f7305c37ef1ab208b800fe8c7121fe8bf1a0a312 100644 (file)
--- a/src/test/neorados/watch_notify.cc
+++ b/src/test/neorados/watch_notify.cc
@@ -10,8 +10,8 @@
  *
  */
 
+#include <boost/asio/detached.hpp>
 #include <boost/system/detail/errc.hpp>
-#include <coroutine>
 #include <cstdint>
 #include <iostream>
 #include <utility>
@@ -36,8 +36,6 @@
 
 #include "gtest/gtest.h"
 
-using std::uint64_t;
-
 namespace asio = boost::asio;
 namespace buffer = ceph::buffer;
 namespace container = boost::container;
@@ -46,7 +44,6 @@ namespace sys = boost::system;
 using namespace std::literals;
 
 using neorados::ReadOp;
-using neorados::WriteOp;
 
 using std::uint64_t;
 
@@ -234,6 +231,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;