P2p: improve timeout_when_socket_not_read test, so that it also checks disconnection timeout. Introduce BackendObserver, rename peer_manager::Observer to PeerManagerObserver

Author: ImplOfAnImpl

p2p/src/net/default_backend/backend.rs

@@ -47,7 +47,7 @@ use crate::{
     net::{
         default_backend::{
             peer,
-            types::{BackendEvent, Command, PeerEvent},
+            types::{BackendEvent, BackendObserver, Command, PeerEvent},
         },
         types::{
             services::Services, ConnectivityEvent, PeerInfo, PeerManagerMessageExt, SyncingEvent,
@@ -169,6 +169,9 @@ pub struct Backend<T: TransportSocket> {
     /// equal to default_networking_service::PREFERRED_PROTOCOL_VERSION, but it can be
     /// overridden for testing purposes.
     node_protocol_version: ProtocolVersion,
+
+    /// Observer object, used by tests.
+    observer: Option<Arc<dyn BackendObserver + Send + Sync>>,
 }
 
 impl<T> Backend<T>
@@ -190,6 +193,7 @@ where
         shutdown_receiver: oneshot::Receiver<()>,
         subscribers_receiver: mpsc::UnboundedReceiver<P2pEventHandler>,
         node_protocol_version: ProtocolVersion,
+        observer: Option<Arc<dyn BackendObserver + Send + Sync>>,
     ) -> Self {
         Self {
             networking_enabled,
@@ -210,6 +214,7 @@ where
             events_controller: EventsController::new(),
             subscribers_receiver,
             node_protocol_version,
+            observer,
         }
     }
 
@@ -403,6 +408,7 @@ where
             backend_event_receiver,
             self.node_protocol_version,
             self.time_getter.shallow_clone(),
+            self.observer.clone(),
         )?;
         let shutdown = Arc::clone(&self.shutdown);
         let handle = tokio_spawn_in_current_tracing_span(

p2p/src/net/default_backend/default_networking_service.rs

@@ -29,7 +29,7 @@ use utils::{atomics::SeqCstAtomicBool, tokio_spawn_in_tracing_span};
 
 use crate::{
     error::P2pError,
-    net::NetworkingService,
+    net::{default_backend::types::BackendObserver, NetworkingService},
     protocol::{ProtocolVersion, SupportedProtocolVersion},
     P2pConfig, P2pEventHandler,
 };
@@ -64,6 +64,7 @@ impl<T: TransportSocket> DefaultNetworkingService<T> {
         shutdown_receiver: oneshot::Receiver<()>,
         subscribers_receiver: mpsc::UnboundedReceiver<P2pEventHandler>,
         protocol_version: ProtocolVersion,
+        observer: Option<Arc<dyn BackendObserver + Send + Sync>>,
         tracing_span: tracing::Span,
     ) -> crate::Result<(
         <Self as NetworkingService>::ConnectivityHandle,
@@ -95,6 +96,7 @@ impl<T: TransportSocket> DefaultNetworkingService<T> {
             shutdown_receiver,
             subscribers_receiver,
             protocol_version,
+            observer,
         );
         let backend_task = tokio_spawn_in_tracing_span(
             async move {
@@ -154,6 +156,7 @@ impl<T: TransportSocket> DefaultNetworkingService<T> {
             shutdown_receiver,
             subscribers_receiver,
             protocol_version,
+            None,
             tracing::Span::current(),
         )
     }

p2p/src/net/default_backend/peer/mod.rs

@@ -46,7 +46,7 @@ use crate::{
     net::{
         default_backend::{
             peer::handshake_handler::HandshakeHandler,
-            types::{BackendEvent, MessageDebugLogSummary, MessageTag, PeerEvent},
+            types::{BackendEvent, BackendObserver, MessageDebugLogSummary, MessageTag, PeerEvent},
         },
         types::PeerManagerMessageExt,
     },
@@ -96,6 +96,9 @@ pub struct Peer<T: TransportSocket> {
 
     /// Time getter
     time_getter: TimeGetter,
+
+    /// Observer object, used by tests.
+    observer: Option<Arc<dyn BackendObserver + Send + Sync>>,
 }
 
 impl<T> Peer<T>
@@ -113,6 +116,7 @@ where
         backend_event_receiver: mpsc::UnboundedReceiver<BackendEvent>,
         node_protocol_version: ProtocolVersion,
         time_getter: TimeGetter,
+        observer: Option<Arc<dyn BackendObserver + Send + Sync>>,
     ) -> crate::Result<Self> {
         let peer_address = socket.remote_address()?.as_peer_address();
 
@@ -131,6 +135,7 @@ where
             backend_event_receiver,
             node_protocol_version,
             time_getter,
+            observer,
         })
     }
 
@@ -206,6 +211,7 @@ where
             mut backend_event_receiver,
             node_protocol_version,
             time_getter,
+            observer,
         } = self;
 
         let handshake_handler = HandshakeHandler::new(
@@ -238,6 +244,7 @@ where
             socket_writer,
             writer_cmd_receiver,
             writer_event_sender,
+            observer.clone(),
         );
 
         // Note: if the outer Option is set, an explicit disconnection should be initiated via
@@ -288,6 +295,10 @@ where
                 }
                 event = socket_reader.recv(), if sync_msg_senders_opt.is_some() => match event {
                     Ok(message) => {
+                        if let Some(observer) = &observer {
+                            observer.on_message_read(peer_id, &message);
+                        }
+
                         let sync_msg_senders = sync_msg_senders_opt.as_mut().expect("sync_msg_senders_opt is some");
                         Self::handle_socket_msg(
                             message,
@@ -416,14 +427,17 @@ fn spawn_writer<S: PeerStream + 'static>(
     socket_writer: MessageWriter<S, Message>,
     cmd_receiver: mpsc::UnboundedReceiver<WriterCommand>,
     event_sender: mpsc::UnboundedSender<WriterEvent>,
+    observer: Option<Arc<dyn BackendObserver + Send + Sync>>,
 ) -> JoinHandle<()> {
     tokio_spawn_in_current_tracing_span(
         async move {
             let writer_result = writer_loop(
                 &p2p_config,
+                peer_id,
                 common_protocol_version,
                 socket_writer,
                 cmd_receiver,
+                observer,
             )
             .await;
 
@@ -437,9 +451,11 @@ fn spawn_writer<S: PeerStream + 'static>(
 
 async fn writer_loop<S: PeerStream>(
     p2p_config: &P2pConfig,
+    peer_id: PeerId,
     common_protocol_version: SupportedProtocolVersion,
     mut socket_writer: MessageWriter<S, Message>,
     mut cmd_receiver: mpsc::UnboundedReceiver<WriterCommand>,
+    observer: Option<Arc<dyn BackendObserver + Send + Sync>>,
 ) -> crate::Result<()> {
     while let Some(cmd) = cmd_receiver.recv().await {
         match cmd {
@@ -450,6 +466,10 @@ async fn writer_loop<S: PeerStream>(
                     message.encoded_size()
                 );
 
+                if let Some(observer) = &observer {
+                    observer.on_message_write(peer_id, &message);
+                }
+
                 tokio::time::timeout(
                     *p2p_config.backend_timeouts.socket_write_timeout,
                     socket_writer.send(*message),

p2p/src/net/default_backend/types.rs

@@ -410,6 +410,15 @@ pub fn can_send_will_disconnect(peer_protocol_version: ProtocolVersion) -> bool
     peer_protocol_version >= SupportedProtocolVersion::V3.into()
 }
 
+/// Backend observer, used by tests.
+pub trait BackendObserver {
+    /// Called before the message is written to the socket.
+    fn on_message_write(&self, peer_id: PeerId, msg: &Message);
+
+    /// Called after the message has been read from the socket.
+    fn on_message_read(&self, peer_id: PeerId, msg: &Message);
+}
+
 #[cfg(test)]
 mod tests {
     use std::net::{IpAddr, Ipv4Addr, SocketAddr};

p2p/src/peer_manager/mod.rs

@@ -203,7 +203,7 @@ where
     addr_list_response_cache: AddrListResponseCache,
 
     /// PeerManager's observer for use by tests.
-    observer: Option<Box<dyn Observer + Send>>,
+    observer: Option<Box<dyn PeerManagerObserver + Send>>,
 
     /// Normally, this will be DefaultDnsSeed, which performs the actual address lookup, but tests can
     /// substitute it with a mock implementation.
@@ -274,7 +274,7 @@ where
         peer_mgr_event_receiver: mpsc::UnboundedReceiver<PeerManagerEvent>,
         time_getter: TimeGetter,
         peerdb_storage: S,
-        observer: Option<Box<dyn Observer + Send>>,
+        observer: Option<Box<dyn PeerManagerObserver + Send>>,
         dns_seed: Box<dyn DnsSeed + Send>,
         mut rng: impl RngCore + Send + 'static,
     ) -> crate::Result<Self> {
@@ -1157,7 +1157,7 @@ where
         }
 
         if let Some(o) = self.observer.as_mut() {
-            o.on_connection_accepted(peer_address, peer_role)
+            o.on_connection_accepted(peer_address, peer_id, peer_role)
         }
 
         Ok(())
@@ -2455,7 +2455,7 @@ where
     }
 }
 
-pub trait Observer {
+pub trait PeerManagerObserver {
     fn on_peer_ban_score_adjustment(&mut self, address: SocketAddress, new_score: u32);
 
     fn on_peer_ban(&mut self, address: BannableAddress);
@@ -2466,7 +2466,12 @@ pub trait Observer {
     fn on_heartbeat(&mut self);
 
     // This will be called for both incoming and outgoing connections.
-    fn on_connection_accepted(&mut self, address: SocketAddress, peer_role: PeerRole);
+    fn on_connection_accepted(
+        &mut self,
+        address: SocketAddress,
+        peer_id: PeerId,
+        peer_role: PeerRole,
+    );
 
     // This will be called after `ConnectivityEvent::ConnectionError` has been handled by
     // the peer manager.

p2p/src/peer_manager/tests/mod.rs

@@ -53,7 +53,7 @@ use crate::{
     },
     peer_manager::PeerManager,
     test_helpers::{peerdb_inmemory_store, test_p2p_config},
-    tests::helpers::{PeerManagerNotification, PeerManagerObserver},
+    tests::helpers::{PeerManagerNotification, PeerManagerObserverImpl},
     types::peer_id::PeerId,
     utils::oneshot_nofail,
     P2pConfig, P2pEventHandler, PeerManagerEvent,
@@ -173,7 +173,7 @@ pub fn make_standalone_peer_manager(
         conn_event_receiver,
     );
     let (peer_mgr_notification_sender, peer_mgr_notification_receiver) = mpsc::unbounded_channel();
-    let peer_mgr_observer = Box::new(PeerManagerObserver::new(peer_mgr_notification_sender));
+    let peer_mgr_observer = Box::new(PeerManagerObserverImpl::new(peer_mgr_notification_sender));
     let dns_seed = DefaultDnsSeed::new(Arc::clone(&chain_config), Arc::clone(&p2p_config));
 
     let peer_mgr = PeerManager::<TcpNetworkingService, _>::new_generic(

p2p/src/peer_manager/tests/unsuccessful_connection_counter_update.rs

@@ -1145,6 +1145,7 @@ async fn peer_accepted_by_peer_mgr(
         peer_mgr_notification,
         PeerManagerNotification::ConnectionAccepted {
             address: peer_address,
+            peer_id,
             peer_role
         }
     );

p2p/src/tests/connection_lockup_when_socket_not_read.rs

@@ -32,7 +32,7 @@ use p2p_test_utils::run_with_timeout;
 use randomness::Rng;
 use serialization::Encode as _;
 use test_utils::{
-    assert_matches,
+    assert_matches, assert_matches_return_val,
     random::{make_seedable_rng, Seed},
     BasicTestTimeGetter,
 };
@@ -41,7 +41,7 @@ use crate::{
     config::{BackendTimeoutsConfig, P2pConfig},
     message::{HeaderList, HeaderListRequest},
     net::{
-        default_backend::types::{HandshakeMessage, Message, P2pTimestamp},
+        default_backend::types::{HandshakeMessage, Message, MessageTag, P2pTimestamp},
         types::PeerManagerMessageExtTag,
     },
     sync::test_helpers::make_new_blocks,
@@ -242,35 +242,56 @@ async fn no_connection_lockup_when_socket_not_read(#[case] seed: Seed) {
     .await;
 }
 
-// Similar to the test above, here the peer also stops reading from the socket, but it doesn't do
-// any discourageable actions this time.
-// The expected result is that the connection should also be terminated.
-// This checks that there is a timeout on socket write attempts.
+#[derive(Debug)]
+enum TimeoutTestType {
+    SocketWrite,
+    Disconnect,
+}
+
+// Here the peer also stops reading from the socket, but doesn't do any discourageable actions.
+// Two cases exist:
+// 1) Socket write timeout is small, disconnection timeout is artificially large (just in case).
+//    The expected result is that the connection should be terminated, though not immediately.
+//    I.e. this checks that there is a timeout on socket write attempts.
+// 2) Disconnection timeout is small, socket write timeout is artificially large.
+//    We wait until the HeaderList message has been sent by the node and then initiate manual
+//    disconnection. The expected result is the same - the connection should be terminated,
+//    but not immediately.
+//    I.e. this checks that there is a timeout on disconnection attempts.
 #[tracing::instrument(skip(seed))]
 #[rstest]
 #[trace]
 #[case(Seed::from_entropy())]
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn timeout_when_socket_not_read(#[case] seed: Seed) {
+async fn timeout_when_socket_not_read(
+    #[case] seed: Seed,
+    #[values(TimeoutTestType::SocketWrite, TimeoutTestType::Disconnect)] test_type: TimeoutTestType,
+) {
     let mut rng = make_seedable_rng(seed);
 
     run_with_timeout(async {
         // Note: this is in bytes; this size should be less than the encoded size of `node_headers_count` headers.
         let channel_max_buf_size = 1000;
         let node_headers_count = 100;
 
-        let socket_write_timeout = Duration::from_secs(3);
+        let timeout = Duration::from_secs(3);
+        let large_timeout = Duration::from_secs(3600);
         let no_disconnection_after = Duration::from_secs(1);
 
+        let (socket_write_timeout, disconnection_timeout) = match test_type {
+            TimeoutTestType::SocketWrite => (timeout, large_timeout),
+            TimeoutTestType::Disconnect => (large_timeout, timeout),
+        };
+
         let time_getter = BasicTestTimeGetter::new();
         let chain_config = Arc::new(chain::config::create_unit_test_config());
         let p2p_config = Arc::new(P2pConfig {
             backend_timeouts: BackendTimeoutsConfig {
                 socket_write_timeout: socket_write_timeout.into(),
+                disconnection_timeout: disconnection_timeout.into(),
 
                 outbound_connection_timeout: Default::default(),
                 peer_handshake_timeout: Default::default(),
-                disconnection_timeout: Default::default(),
             },
 
             bind_addresses: Default::default(),
@@ -382,9 +403,10 @@ async fn timeout_when_socket_not_read(#[case] seed: Seed) {
 
         log::debug!("Expecting PeerManagerNotification::ConnectionAccepted");
         let peer_mgr_notif = test_node.peer_mgr_notification_receiver().recv().await.unwrap();
-        assert_matches!(
+        let peer_id = assert_matches_return_val!(
             peer_mgr_notif,
-            PeerManagerNotification::ConnectionAccepted { .. }
+            PeerManagerNotification::ConnectionAccepted { peer_id, .. },
+            peer_id
         );
 
         log::debug!("Expecting PeerManagerNotification::FirstSyncMessageReceived");
@@ -397,6 +419,16 @@ async fn timeout_when_socket_not_read(#[case] seed: Seed) {
             }
         );
 
+        let manual_disconnect_result_receiver = match test_type {
+            TimeoutTestType::SocketWrite => None,
+            TimeoutTestType::Disconnect => {
+                test_node
+                    .wait_for_next_backend_socket_write(peer_id, MessageTag::HeaderList)
+                    .await;
+                Some(test_node.start_disconnecting(peer_id))
+            }
+        };
+
         // Wait for a short period of time, there should be no disconnection.
         tokio::time::timeout(
             no_disconnection_after,
@@ -413,6 +445,10 @@ async fn timeout_when_socket_not_read(#[case] seed: Seed) {
             PeerManagerNotification::ConnectionClosed { .. }
         );
 
+        if let Some(manual_disconnect_result_receiver) = manual_disconnect_result_receiver {
+            manual_disconnect_result_receiver.await.unwrap().unwrap();
+        }
+
         test_node.join().await;
     })
     .await;