Reconnect peerd if connection is dropped during running swap

doc/reconnect_sequencediagram.txt

@@ -0,0 +1,24 @@
+title Peer Reconnect Procedure
+
+participant t_peerd_run
+participant t_peerd_recv
+participant m_peerd_recv
+participant m_peerd_run
+participant m_farcasterd
+participant m_swapd
+
+==Peer reconnect procedure==
+m_swapd -> m_farcasterd : Ctl PeerdUnreachable
+m_farcasterd -> m_peerd_run : Ctl Terminate
+m_peerd_run -> m_farcasterd : Ctl PeerdTerminated
+m_peerd_recv -> m_peerd_recv: terminate
+m_peerd_run -> m_peerd_run: terminate
+t_peerd_recv -> t_peerd_run : Ctl PeerdListenerRuntimeShutdown
+t_peerd_recv -> t_peerd_recv: terminate
+==Taker peerd restarts connection, connects to maker forked peerd==
+m_peerd_run -> m_peerd_run : forked from listener
+t_peerd_run -> m_peerd_run : NodeId (raw)
+t_peerd_run -> t_peerd_recv: spawn
+m_peerd_run -> m_farcasterd: Ctl Hello
+m_farcasterd -> m_swapd: Ctl PeerdReconnected
+

src/bin/peerd.rs

@@ -101,11 +101,10 @@ use std::net::TcpListener;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::time::Duration;
 
-use bitcoin::secp256k1::PublicKey;
 use farcaster_node::peerd::{self, Opts};
 use farcaster_node::LogStyle;
 use farcaster_node::ServiceConfig;
-use internet2::{session, FramingProtocol, NodeAddr, RemoteNodeAddr, RemoteSocketAddr};
+use internet2::{session, FramingProtocol, RemoteNodeAddr, RemoteSocketAddr};
 use microservices::peer::PeerConnection;
 
 /*
@@ -177,26 +176,23 @@ fn main() {
      */
 
     let local_node = opts.peer_key_opts.local_node();
-    let local_id = local_node.node_id();
     info!(
         "{}: {}",
         "Local node id".bright_green_bold(),
-        local_id.bright_yellow_bold()
+        local_node.node_id().bright_yellow_bold()
     );
 
     let peer_socket = PeerSocket::from(opts.clone());
     debug!("Peer socket parameter interpreted as {}", peer_socket);
 
-    let internal_id: NodeAddr;
     let mut local_socket: Option<InetSocketAddr> = None;
-    let mut remote_id: Option<PublicKey> = None;
-    let mut remote_socket: InetSocketAddr;
-    let connect: bool;
+    let mut remote_node_addr: Option<RemoteNodeAddr> = None;
+    let forked_from_listener: bool;
     let connection = match peer_socket {
         PeerSocket::Listen(RemoteSocketAddr::Ftcp(inet_addr)) => {
             debug!("Running in LISTEN mode");
 
-            connect = false;
+            forked_from_listener = true;
             local_socket = Some(inet_addr);
 
             debug!("Binding TCP socket {}", inet_addr);
@@ -213,8 +209,6 @@ fn main() {
                     .expect("Error accepting incoming peer connection");
                 debug!("New connection from {}", remote_socket_addr);
 
-                remote_socket = remote_socket_addr.into();
-
                 // TODO: Support multithread mode
                 debug!("Forking child process");
                 if let ForkResult::Child = unsafe { fork().expect("Unable to fork child process") }
@@ -228,30 +222,25 @@ fn main() {
                         session::Raw::with_brontide(stream, local_node.private_key(), inet_addr)
                             .expect("Unable to establish session with the remote peer");
 
-                    internal_id = NodeAddr::Remote(RemoteNodeAddr {
-                        node_id: opts.peer_key_opts.internal_node().node_id(),
-                        remote_addr: RemoteSocketAddr::Ftcp(remote_socket),
-                    });
                     debug!(
-                        "Session successfully established with new unique id: {}",
-                        internal_id
+                        "Session successfully established with {}",
+                        remote_socket_addr
                     );
+
                     break PeerConnection::with(session);
                 }
                 debug!("Child forked; returning into main listener event loop");
                 continue;
             }
         }
-        PeerSocket::Connect(remote_node_addr) => {
-            debug!("Running in CONNECT mode");
+        PeerSocket::Connect(remote_node) => {
+            debug!("Peerd running in CONNECT mode");
 
-            connect = true;
-            internal_id = NodeAddr::Remote(remote_node_addr.clone());
-            remote_id = Some(remote_node_addr.node_id);
-            remote_socket = remote_node_addr.remote_addr.into();
+            forked_from_listener = false;
+            remote_node_addr = Some(remote_node.clone());
 
-            debug!("Connecting to {}", &remote_node_addr.addr());
-            PeerConnection::connect(remote_node_addr, &local_node)
+            debug!("Connecting to {}", &remote_node.addr());
+            PeerConnection::connect(remote_node, &local_node)
                 .expect("Unable to connect to the remote peer")
         }
         _ => unimplemented!(),
@@ -260,27 +249,21 @@ fn main() {
     debug!("Starting runtime ...");
 
     /* A maker / listener passes the following content
-        internal_id: local key and remote address
-        remote_id: None
+        remote_node_addr: none
         local_socket: local inet address
-        remote_socket: address of the remote socket
         connect: false
 
     A taker / connecter passes the following content
-        internal_id: remote key and remote address
-        remote_id: remote peer id
+        remote_node_addr: full internet2 remote node address
         local_socket: None
-        remote_socket: remote node addr
         connect: true */
     peerd::run(
         service_config,
         connection,
-        internal_id,
-        local_id,
-        remote_id,
+        remote_node_addr,
         local_socket,
-        remote_socket,
-        connect,
+        local_node,
+        forked_from_listener,
     )
     .expect("Error running peerd runtime");
 

src/farcasterd/runtime.rs

@@ -105,6 +105,7 @@ pub fn run(
         listens: none!(),
         started: SystemTime::now(),
         connections: none!(),
+        report_peerd_reconnect: none!(),
         running_swaps: none!(),
         spawning_services: none!(),
         making_swaps: none!(),
@@ -135,16 +136,17 @@ pub struct Runtime {
     listens: HashMap<OfferId, RemoteSocketAddr>,
     started: SystemTime,
     connections: HashSet<NodeAddr>,
+    report_peerd_reconnect: HashMap<NodeAddr, ServiceId>,
     running_swaps: HashSet<SwapId>,
     spawning_services: HashMap<ServiceId, ServiceId>,
     making_swaps: HashMap<ServiceId, (request::InitSwap, Network)>,
     taking_swaps: HashMap<ServiceId, (request::InitSwap, Network)>,
     public_offers: HashSet<PublicOffer<BtcXmr>>,
     arb_addrs: HashMap<PublicOfferId, bitcoin::Address>,
     acc_addrs: HashMap<PublicOfferId, monero::Address>,
-    consumed_offers: HashMap<OfferId, SwapId>,
-    node_ids: HashMap<OfferId, PublicKey>, // TODO is it possible? HashMap<SwapId, PublicKey>
-    peerd_ids: HashMap<OfferId, ServiceId>,
+    consumed_offers: HashMap<OfferId, (SwapId, ServiceId)>,
+    node_ids: HashMap<OfferId, PublicKey>, // Only populated by maker. TODO is it possible? HashMap<SwapId, PublicKey>
+    peerd_ids: HashMap<OfferId, ServiceId>, // Only populated by maker.
     wallet_token: Token,
     pending_requests: HashMap<request::RequestId, (Request, ServiceId)>,
     syncer_services: HashMap<(Coin, Network), ServiceId>,
@@ -285,9 +287,9 @@ impl Runtime {
         self.consumed_offers = self
             .consumed_offers
             .drain()
-            .filter_map(|(k, v)| {
-                if swapid != &v {
-                    Some((k, v))
+            .filter_map(|(k, (swap_id, service_id))| {
+                if swapid != &swap_id {
+                    Some((k, (swap_id, service_id)))
                 } else {
                     offerid = Some(k);
                     None
@@ -496,6 +498,17 @@ impl Runtime {
                                 connection_id.bright_blue_italic(),
                                 self.connections.len().bright_blue_bold()
                             );
+                            if let Some(swap_service_id) =
+                                self.report_peerd_reconnect.remove(connection_id)
+                            {
+                                debug!("Letting {} know of peer reconnection.", swap_service_id);
+                                endpoints.send_to(
+                                    ServiceBus::Ctl,
+                                    self.identity(),
+                                    swap_service_id,
+                                    Request::PeerdReconnected,
+                                )?;
+                            }
                         } else {
                             warn!(
                                 "Connection {} was already registered; the service probably was relaunched",
@@ -694,7 +707,7 @@ impl Runtime {
                     );
 
                     self.consumed_offers
-                        .insert(public_offer.offer.id(), swap_id);
+                        .insert(public_offer.offer.id(), (swap_id, peer.clone()));
                     self.stats.incr_initiated();
                     launch_swapd(
                         self,
@@ -715,7 +728,7 @@ impl Runtime {
             }
 
             Request::Keys(Keys(sk, pk, id)) if self.pending_requests.contains_key(&id) => {
-                trace!("received peerd keys");
+                debug!("received peerd keys {}", sk.display_secret());
                 if let Some((request, source)) = self.pending_requests.remove(&id) {
                     // storing node_id
                     trace!("Received expected peer keys, injecting key in request");
@@ -1295,10 +1308,46 @@ impl Runtime {
                             "removed connection {} from farcasterd registered connections",
                             addr
                         );
+
+                        // log a message if a swap running over this connection
+                        // is not completed, and thus present in consumed_offers
+                        let peerd_id = ServiceId::Peer(addr.clone());
+                        if self
+                            .consumed_offers
+                            .iter()
+                            .find_map(|(_, (_, service_id))| {
+                                if service_id.clone() == peerd_id {
+                                    Some(0)
+                                } else {
+                                    None
+                                }
+                            })
+                            .is_some()
+                        {
+                            info!("a swap is still running over the terminated peer {}, the counterparty will attempt to reconnect.", addr);
+                        }
                     }
                 }
             }
 
+            Request::PeerdUnreachable(ServiceId::Peer(addr)) => {
+                if self.connections.contains(&addr) {
+                    warn!(
+                        "Peerd {} was reported to be unreachable, attempting to
+                        terminate to kick-off re-connect procedure, if we are
+                        taker and the swap is still running.",
+                        addr
+                    );
+                    endpoints.send_to(
+                        ServiceBus::Ctl,
+                        ServiceId::Farcasterd,
+                        ServiceId::Peer(addr.clone()),
+                        Request::Terminate,
+                    )?;
+                }
+                self.report_peerd_reconnect.insert(addr, source);
+            }
+
             req => {
                 error!("Ignoring unsupported request: {}", req.err());
             }