Improve stability of channels_test:Key rotation (#6782)

Co-authored-by: Amaury Chamayou <[email protected]>

src/ds/non_blocking.h

@@ -102,10 +102,12 @@ namespace ringbuffer
       {
         for (auto& it : pending)
         {
-          const auto buffer_end = it.buffer.data() + it.buffer.size();
-          if (
-            it.marker == marker.value() &&
-            marker.value() != reinterpret_cast<uint64_t>(buffer_end))
+          if (it.finished)
+          {
+            continue;
+          }
+
+          if (it.marker == marker.value())
           {
             // This is a pending write - dump data directly to write marker,
             // which should be within the appropriate buffer
@@ -118,6 +120,7 @@ namespace ringbuffer
                 (size_t)it.buffer.data()));
             }
 
+            const auto buffer_end = it.buffer.data() + it.buffer.size();
             if (dest + size > buffer_end)
             {
               throw std::runtime_error(fmt::format(
@@ -128,7 +131,11 @@ namespace ringbuffer
                 (size_t)buffer_end));
             }
 
-            std::memcpy(dest, bytes, size);
+            if (size != 0)
+            {
+              std::memcpy(dest, bytes, size);
+            }
+
             dest += size;
             it.marker = (size_t)dest;
             return {it.marker};

src/node/channels.h

@@ -145,9 +145,6 @@ namespace ccf
     }
   }
 
-  class KeyExchangeProtocol
-  {};
-
   // Key exchange states are:
   // - Have nothing
   // - Initiated (have my own share)
@@ -251,6 +248,8 @@ namespace ccf
 
         send_key = nullptr;
         recv_key = nullptr;
+        reset_key_exchange();
+        initiate();
       }
     }
 
@@ -542,6 +541,8 @@ namespace ccf
       // shares back to the initiator
       send_key_exchange_response();
 
+      flush_pending_outgoing();
+
       return true;
     }
 
@@ -633,6 +634,8 @@ namespace ccf
 
       send_key_exchange_final();
 
+      flush_pending_outgoing();
+
       update_recv_key();
 
       establish();
@@ -804,7 +807,10 @@ namespace ccf
         "Node certificate serial numbers: node={} peer={}",
         node_cv->serial_number(),
         peer_cv->serial_number());
+    }
 
+    void flush_pending_outgoing()
+    {
       if (outgoing_consensus_msg.has_value())
       {
         send_unsafe(