Add terminator for stopping channel via message broadcast.

include/bitcoin/node/chase.hpp

@@ -45,7 +45,7 @@ enum class chase
     /// Issued by 'organize' and handled by 'check', 'validate', 'confirm'.
     bump,
 
-    /// Channels (all) directed to stop (default).
+    /// Channels (all) directed to stop with the given code (default).
     /// Issued by 'full_node' and handled by 'observer'.
     suspend,
 

include/bitcoin/node/protocols/protocol_observer.hpp

@@ -68,6 +68,10 @@ class BCN_API protocol_observer
     virtual bool handle_broadcast_diagnostics(const code& ec,
         const network::diagnostics::cptr& message, uint64_t sender) NOEXCEPT;
 
+    /// Stop the channel if it is the member of a stop.
+    virtual bool handle_broadcast_terminator(const code& ec,
+        const network::terminator::cptr& message, uint64_t sender) NOEXCEPT;
+
     /// The capture group of the channel (the session determines the group).
     virtual network::diagnostics::target group() const NOEXCEPT;
 

include/bitcoin/node/protocols/protocol_transaction_out_70013.hpp

@@ -58,8 +58,7 @@ class BCN_API protocol_transaction_out_70013
     bool insufficient(const database::fee_rate& rate) const NOEXCEPT;
     void do_send_fee_filter() NOEXCEPT;
 
-    // These are protected by strand.
-    uint64_t minimum_fee_{};
+    // This is protected by strand.
     uint64_t sent_fee_{};
 };
 

src/full_node.cpp

@@ -293,7 +293,7 @@ void full_node::suspend(const code& ec) NOEXCEPT
 {
     LOGS("Suspending network, " << ec.message());
     net::suspend(ec);
-    notify(ec, chase::suspend, {});
+    notify(error::suspended_channel, chase::suspend, {});
 }
 
 void full_node::fault(const code& ec) NOEXCEPT

src/protocols/protocol_observer.cpp

@@ -50,10 +50,8 @@ void protocol_observer::start() NOEXCEPT
         SUBSCRIBE_CHANNEL(inventory, handle_receive_inventory, _1, _2);
     }
 
-    SUBSCRIBE_BROADCAST(network::diagnostics, handle_broadcast_diagnostics,
-        _1, _2, _3);
-
-    ////SUBSCRIBE_CHANNEL(get_data, handle_receive_get_data, _1, _2);
+    SUBSCRIBE_BROADCAST(network::diagnostics, handle_broadcast_diagnostics, _1, _2, _3);
+    SUBSCRIBE_BROADCAST(network::terminator, handle_broadcast_terminator, _1, _2, _3);
     protocol_peer::start();
 }
 
@@ -68,7 +66,7 @@ void protocol_observer::stopping(const code& ec) NOEXCEPT
 // handle events (suspend)
 // ----------------------------------------------------------------------------
 
-bool protocol_observer::handle_chase(const code&, chase event_,
+bool protocol_observer::handle_chase(const code& ec, chase event_,
     event_value) NOEXCEPT
 {
     // Do not pass ec to stopped as it is not a call status.
@@ -79,7 +77,8 @@ bool protocol_observer::handle_chase(const code&, chase event_,
     {
         case chase::suspend:
         {
-            stop(error::suspended_channel);
+            // The code distinguishes a stop from a drop (manual removal).
+            stop(ec);
             break;
         }
         case chase::stop:
@@ -139,28 +138,68 @@ bool protocol_observer::handle_broadcast_diagnostics(const code& ec,
     if (!message->member(identifier()) && !message->member(group()))
         return true;
 
+    using namespace system;
     const auto peer = peer_version();
+    uint64_t services{ service::node_none };
+    network::config::address local{};
+    int64_t time_offset{};
+    std::string agent{};
+    bool relay{};
+
+    if (peer)
+    {
+        time_offset = subtract<int64_t>(peer->timestamp, created());
+        local = { peer->address_receiver };
+        services = peer->services;
+        agent = peer->user_agent;
+        relay = peer->relay;
+    }
 
     message->add(
     {
         .identifier = identifier(),
         .address = outbound(),
+        .local = local,
+        .binding = binding(),
         .group = group(),
         .version = negotiated_version(),
-        .services = peer ? peer->services : service::node_none,
+        .services = services,
         .sent = sent(),
         .received = received(),
         .created = created(),
         .last_read = last_read(),
         .last_write = last_write(),
+        .time_offset = time_offset,
+        .minimum_fee = minimum_fee(),
+        .ping_time = ping_time(),
+        .minimum_ping_time = minimum_ping_time(),
+        .pending_ping_time = pending_ping_time(),
         .start_height = start_height(),
         .encrypted = encrypted(),
-        .agent = peer ? peer->user_agent : std::string{}
+        .relay = relay,
+        .agent = agent
     });
 
     return true;
 }
 
+// The first member stops, so the round completes without the other channels.
+bool protocol_observer::handle_broadcast_terminator(const code& ec,
+    const network::terminator::cptr& message, uint64_t) NOEXCEPT
+{
+    BC_ASSERT(stranded());
+
+    if (stopped(ec))
+        return false;
+
+    if (!message->member(identifier(), outbound()))
+        return true;
+
+    message->stopped();
+    stop(message->reason());
+    return false;
+}
+
 BC_POP_WARNING()
 BC_POP_WARNING()
 

src/protocols/protocol_transaction_out_70013.cpp

@@ -106,7 +106,7 @@ bool protocol_transaction_out_70013::handle_receive_fee_filter(const code& ec,
     if (stopped(ec))
         return false;
 
-    minimum_fee_ = message->minimum_fee;
+    set_minimum_fee(message->minimum_fee);
     return true;
 }
 
@@ -140,7 +140,7 @@ bool protocol_transaction_out_70013::insufficient(
     const database::fee_rate& rate) const NOEXCEPT
 {
     return ceilinged_multiply(rate.fee, vbytes_per_vkbyte) <
-        ceilinged_multiply(minimum_fee_,
+        ceilinged_multiply(minimum_fee(),
             possible_wide_cast<uint64_t>(rate.bytes));
 }