Prevent coalesced libuv wakes from stranding inbound reads

CHANGELOG.md

@@ -5,6 +5,14 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## [7.0.19]
+
+[7.0.19]: https://github.com/microsoft/CCF/releases/tag/ccf-7.0.19
+
+### Fixed
+
+- Paused RPC reads now resume when another interface releases the inbound budget, even if older libuv versions coalesce the notification. Previously, reads could remain paused until an unrelated event triggered a recheck (#8498).
+
 ## [7.0.18]
 
 [7.0.18]: https://github.com/microsoft/CCF/releases/tag/ccf-7.0.18

doc/architecture/tls_internals.rst

@@ -86,6 +86,8 @@ The budget is node-wide rather than per-connection, so a single connection can e
 
 Releasing the budget wakes every registered transport, not only the one whose session made room, so an interface with no traffic of its own does not stay paused. A session which never reports - a custom protocol, or one whose queued work is cancelled as it is destroyed - cannot strand the budget either: whatever is still outstanding for a connection is released when that connection is torn down.
 
+Admission rechecks use the same mutex as queued responses and worker completions. This matters with ``uv_async_t`` notification coalescing on libuv versions before 1.53: the loop clears libuv's pending bit before entering its callback, but an unsynchronised sender could still observe its previous value and skip a notification. In the reduced ordering, the callback could clear the admission flag before a worker sets it, the worker could acquire the lifecycle lock and send, and the callback could acquire the lifecycle lock only later. These shared acquisitions may overlap; making the sender's lock exclusive would still leave the callback's later acquisition too late to order the worker's earlier read of libuv's pending bit with the clear. With the queue mutex, either the worker publishes the recheck before the callback drains it, or the callback releases the mutex before the worker publishes and sends. In the latter case, the pending-bit clear happens before the sender reads it. The separate lifecycle lock continues to protect the async handle during shutdown.
+
 A connection paused this way is performing no I/O, so it ages towards the idle timeout in the usual way. That is intentional: a connection which cannot make progress in either direction for the whole idle period is one the node is too far behind to serve.
 
 Writing and backpressure

python/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "ccf"
-version = "7.0.18"
+version = "7.0.19"
 authors = [
   { name="CCF Team", email="CCF-Sec@microsoft.com" },
 ]

src/tls/openssl_server.h

@@ -372,7 +372,8 @@ namespace ccf::tls
     // Loop-owned connections affected by commands or worker completions.
     std::vector<std::shared_ptr<Conn>> dirty_connections;
     // Remember resumption even if saturation clears between loop passes.
-    std::atomic<bool> recheck_read_interest{false};
+    // Share the queue mutex so a coalesced wake cannot miss a later recheck.
+    bool recheck_read_interest CCF_GUARDED_BY(out_mutex) = false;
 
     // Cross-thread server-cert (re)load requests (deferred cert / rotation),
     // applied on the loop thread so `ctx` is only ever touched there.
@@ -1340,13 +1341,13 @@ namespace ccf::tls
     // Apply worker completions and cross-thread commands on the libuv thread.
     void drain_pending_out() CCF_EXCLUDES(out_mutex, lifecycle_mutex)
     {
-      const bool recheck_all =
-        recheck_read_interest.exchange(false, std::memory_order_acq_rel);
+      bool recheck_all = false;
       std::vector<OutItem> items;
       std::vector<DriveResult> completions;
       std::vector<std::pair<std::string, std::string>> certs;
       {
         ccf::ds::MutexGuard g(out_mutex);
+        recheck_all = std::exchange(recheck_read_interest, false);
         std::swap(items, pending_out);
         std::swap(completions, completed_drives);
         std::swap(certs, pending_certs);
@@ -1877,8 +1878,10 @@ namespace ccf::tls
           inbound_admission->register_waker([weak = weak_from_this()]() {
             if (auto self = weak.lock())
             {
-              self->recheck_read_interest.store(
-                true, std::memory_order_release);
+              {
+                ccf::ds::MutexGuard g(self->out_mutex);
+                self->recheck_read_interest = true;
+              }
               self->wake();
             }
           });

src/tls/test/openssl_server_test.cpp

@@ -2369,11 +2369,11 @@ TEST_CASE("Budget resumption revisits idle connections on every interface")
   constexpr size_t interface_count = 2;
   constexpr size_t limit = 1;
   auto admission = std::make_shared<InboundAdmission>(limit);
-  admission->queued(limit);
 
   std::array<std::atomic<size_t>, interface_count> received{};
   std::array<std::atomic<bool>, interface_count> correct_payload{};
   std::atomic<size_t> accepted{0};
+  std::atomic<bool> release_source{false};
   std::array<std::shared_ptr<OpenSSLServer>, interface_count> servers;
   std::array<int, interface_count> clients;
   clients.fill(-1);
@@ -2392,6 +2392,15 @@ TEST_CASE("Budget resumption revisits idle connections on every interface")
         correct_payload[i] =
           data.size() == 1 && data[0] == static_cast<uint8_t>('a' + i);
         received[i] += data.size();
+        if (i == 0)
+        {
+          admission->queued(data.size());
+          while (!release_source.load())
+          {
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+          }
+          admission->consumed(data.size());
+        }
       },
       OpenSSLServer::OnClose{},
       [&](::tcp::ConnID) -> std::optional<bool> {
@@ -2431,6 +2440,21 @@ TEST_CASE("Budget resumption revisits idle connections on every interface")
       setup_error = sent < 0 ? errno : EIO;
       break;
     }
+    if (i == 0)
+    {
+      const auto saturated_deadline =
+        std::chrono::steady_clock::now() + std::chrono::seconds(5);
+      while (!admission->saturated() &&
+             std::chrono::steady_clock::now() < saturated_deadline)
+      {
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+      }
+      if (!admission->saturated())
+      {
+        setup_error = ETIMEDOUT;
+        break;
+      }
+    }
   }
 
   const auto accepted_deadline =
@@ -2441,17 +2465,18 @@ TEST_CASE("Budget resumption revisits idle connections on every interface")
     std::this_thread::sleep_for(std::chrono::milliseconds(1));
   }
 
-  // Let the masked readability passes finish, leaving no dirty connections.
+  // No more data is sent on either interface. Let the masked readability pass
+  // finish, leaving the second connection idle with data waiting to be read.
   std::this_thread::sleep_for(std::chrono::milliseconds(100));
   const std::array<size_t, interface_count> stalled_at = {
     received[0].load(), received[1].load()};
 
-  // Only the shared admission wake can revisit these otherwise idle sockets.
-  admission->consumed(limit);
+  // The first interface releases the budget from its worker. Only the shared
+  // admission wake can revisit the otherwise idle second interface.
+  release_source.store(true);
   const auto resumed_deadline =
     std::chrono::steady_clock::now() + std::chrono::seconds(5);
-  while (setup_error == 0 &&
-         (received[0].load() == 0 || received[1].load() == 0) &&
+  while (setup_error == 0 && received[1].load() == 0 &&
          std::chrono::steady_clock::now() < resumed_deadline)
   {
     std::this_thread::sleep_for(std::chrono::milliseconds(1));
@@ -2477,9 +2502,10 @@ TEST_CASE("Budget resumption revisits idle connections on every interface")
   INFO("Socket setup error: " << setup_error);
   CHECK(setup_error == 0);
   CHECK(accepted.load() == interface_count);
+  CHECK(stalled_at[0] == 1);
+  CHECK(stalled_at[1] == 0);
   for (size_t i = 0; i < interface_count; ++i)
   {
-    CHECK(stalled_at[i] == 0);
     CHECK(resumed_at[i] == 1);
     CHECK(valid_at_resume[i]);
   }