Decouple channel management from host session management

.daily_canary

@@ -1 +1 @@
-cough
+LET them be triggered

src/consensus/aft/raft.h

@@ -824,6 +824,7 @@ namespace aft
               std::make_unique<NonceRevealCallback>(*this, from, std::move(r));
             break;
           }
+
           case bft_view_change:
           {
             RequestViewChangeMsg r =
@@ -1295,10 +1296,12 @@ namespace aft
         (state->new_view_idx > prev_idx) && (state->new_view_idx <= end_idx);
 
       LOG_DEBUG_FMT(
-        "Send append entries from {} to {}: {} to {} ({})",
-        state->my_node_id.trim(),
-        to.trim(),
-        start_idx,
+        "Send append entries from {} to {}: ({}.{}, {}.{}] ({})",
+        state->my_node_id,
+        to,
+        prev_term,
+        prev_idx,
+        term_of_idx,
         end_idx,
         state->commit_idx);
 
@@ -1480,6 +1483,7 @@ namespace aft
             prev_term,
             r.prev_term);
         }
+
         send_append_entries_response(from, AppendEntriesResponseType::FAIL);
         return;
       }
@@ -2123,12 +2127,10 @@ namespace aft
     void send_append_entries_response(
       ccf::NodeId to, AppendEntriesResponseType answer)
     {
-      LOG_DEBUG_FMT(
-        "Send append entries response from {} to {} for index {}: {}",
-        state->my_node_id.trim(),
-        to.trim(),
-        state->last_idx,
-        answer);
+      if (answer == AppendEntriesResponseType::REQUIRE_EVIDENCE)
+      {
+        state->requested_evidence_from = to;
+      }
 
       if (answer == AppendEntriesResponseType::REQUIRE_EVIDENCE)
       {
@@ -2148,6 +2150,13 @@ namespace aft
         state->commit_idx :
         state->last_idx;
 
+      LOG_DEBUG_FMT(
+        "Send append entries response from {} to {} for index {}: {}",
+        state->my_node_id.trim(),
+        to.trim(),
+        matching_idx,
+        answer);
+
       AppendEntriesResponse response = {{raft_append_entries_response},
                                         state->current_view,
                                         matching_idx,
@@ -2696,10 +2705,6 @@ namespace aft
       {
         for (auto it = nodes.begin(); it != nodes.end(); ++it)
         {
-          channels->create_channel(
-            it->first,
-            it->second.node_info.hostname,
-            it->second.node_info.port);
           send_request_vote(it->first);
         }
       }
@@ -3283,22 +3288,19 @@ namespace aft
           auto index = state->last_idx + 1;
           nodes.try_emplace(node_info.first, node_info.second, index, 0);
 
-          if (
-            replica_state == kv::ReplicaState::Leader ||
-            consensus_type == ConsensusType::BFT)
-          {
-            channels->create_channel(
-              node_info.first,
-              node_info.second.hostname,
-              node_info.second.port);
-          }
+          channels->associate_node_address(
+            node_info.first, node_info.second.hostname, node_info.second.port);
 
           if (replica_state == kv::ReplicaState::Leader)
           {
             send_append_entries(node_info.first, index);
           }
 
-          LOG_INFO_FMT("Added raft node {}", node_info.first);
+          LOG_INFO_FMT(
+            "Added raft node {} ({}:{})",
+            node_info.first,
+            node_info.second.hostname,
+            node_info.second.port);
         }
       }
     }

src/consensus/aft/test/logging_stub.h

@@ -115,18 +115,13 @@ namespace aft
       return std::nullopt;
     }
 
-    void create_channel(
+    void associate_node_address(
       const ccf::NodeId& peer_id,
       const std::string& peer_hostname,
-      const std::string& peer_service,
-      size_t message_limit = ccf::Channel::default_message_limit) override
+      const std::string& peer_service) override
     {}
 
-    void destroy_channel(const ccf::NodeId& peer_id) override {}
-
-    void destroy_all_channels() override {}
-
-    void close_all_outgoing() override {}
+    void close_channel(const ccf::NodeId& peer_id) override {}
 
     bool send_authenticated(
       const ccf::NodeId& to,
@@ -148,9 +143,11 @@ namespace aft
       return true;
     }
 
-    void recv_message(
+    bool recv_channel_message(
       const ccf::NodeId& from, const uint8_t* data, size_t size) override
-    {}
+    {
+      return true;
+    }
 
     void initialize(
       const ccf::NodeId& self_id,

src/crypto/symmetric_key.h

@@ -96,14 +96,19 @@ namespace crypto
       return serial_hdr;
     }
 
-    void deserialise(const std::vector<uint8_t>& serial_hdr)
+    void deserialise(const std::vector<uint8_t>& ser)
     {
-      auto data_ = serial_hdr.data();
-      auto size = serial_hdr.size();
+      auto data = ser.data();
+      auto size = ser.size();
+
+      deserialise(data, size);
+    }
 
+    void deserialise(const uint8_t*& data, size_t& size)
+    {
       memcpy(
-        tag, serialized::read(data_, size, GCM_SIZE_TAG).data(), GCM_SIZE_TAG);
-      memcpy(iv, serialized::read(data_, size, SIZE_IV).data(), SIZE_IV);
+        tag, serialized::read(data, size, GCM_SIZE_TAG).data(), GCM_SIZE_TAG);
+      memcpy(iv, serialized::read(data, size, SIZE_IV).data(), SIZE_IV);
     }
   };
 

src/host/main.cpp

@@ -645,8 +645,7 @@ int main(int argc, char** argv)
     // This includes DNS resolution and potentially dynamic port assignment (if
     // requesting port 0). The hostname and port may be modified - after calling
     // it holds the final assigned values.
-    asynchost::NodeConnectionsTickingReconnect node(
-      20ms, //< Flush reconnections every 20ms
+    asynchost::NodeConnections node(
       bp.get_dispatcher(),
       ledger,
       writer_factory,