RMME reset

Author: cowlicks

src/mqueue.rs

@@ -10,7 +10,7 @@ use std::{
 
 use compact_encoding::CompactEncoding as _;
 use futures::{Sink, Stream};
-use tracing::{error, instrument};
+use tracing::{error, info, instrument};
 
 use crate::{message::ChannelMessage, noise::EncryptionInfo, NoiseEvent};
 
@@ -119,9 +119,18 @@ impl<IO: Stream<Item = NoiseEvent> + Sink<Vec<u8>> + Send + Unpin + 'static> Str
 {
     type Item = MqueueEvent;
 
+    #[instrument(skip_all, ret)]
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let _ = self.poll_outbound(cx);
-        self.poll_inbound(cx)
+        match self.poll_inbound(cx) {
+            Poll::Ready(Some(MqueueEvent::Message(Ok(x)))) => {
+                for m in x.iter() {
+                    info!("RX ChannelMessage::{m}");
+                }
+                Poll::Ready(Some(MqueueEvent::Message(Ok(x))))
+            }
+            x => x,
+        }
     }
 }
 

src/noise.rs

@@ -110,7 +110,7 @@ where
     }
 
     /// Check that we've done as much work as possible. Sending, receiving, encrypting and decrypting.
-    #[instrument(skip_all, ret)]
+    #[instrument(name = "did_as_much_as_possible", skip_all, ret)]
     fn did_as_much_as_possible(&mut self, cx: &mut Context<'_>) -> bool {
         // No incoming encrypted messages available.
         self.poll_incomming_encrypted_messages(cx).is_pending()
@@ -124,8 +124,7 @@ where
 
     /// Handle all message throughput. Sends, encrypts and decrypts messages
     /// Returns `true` `step` is already [`Step::Established`].
-    #[allow(clippy::too_many_arguments)]
-    #[instrument(skip_all, ret)]
+    #[instrument(name = "poll_message_throughput", skip_all, ret)]
     fn poll_message_throughput(&mut self, cx: &mut Context<'_>) -> bool {
         self.poll_outgoing_encrypted_messages(cx);
         let _ = self.poll_incomming_encrypted_messages(cx);

src/test_utils.rs

@@ -86,9 +86,10 @@ pub(crate) fn log() {
             .with_targets(true)
             .with_bracketed_fields(true)
             .with_indent_lines(true)
-            .with_span_modes(true)
             .with_thread_ids(false)
-            .with_thread_names(false);
+            .with_thread_names(true)
+            //.with_span_modes(true)
+            ;
 
         tracing_subscriber::registry()
             .with(env_filter)

tests/js_interop.rs

@@ -27,6 +27,7 @@ use tokio::{
     task,
     time::sleep,
 };
+use tracing::instrument;
 
 use hypercore_protocol::{discovery_key, schema::*, Channel, Event, Message, ProtocolBuilder};
 
@@ -186,6 +187,7 @@ async fn rcns(server_writer: bool, port: u32) -> Result<()> {
         &result_path,
     )
     .await?;
+    dbg!();
     assert_result(result_path, item_count, item_size, data_char).await?;
     drop(server);
 
@@ -330,11 +332,15 @@ async fn run_client(
     data_path: &str,
     result_path: &str,
 ) -> Result<()> {
+    dbg!();
     let hypercore = if is_writer {
+        dbg!();
         create_writer_hypercore(data_count, data_size, data_char, data_path).await?
     } else {
+        dbg!();
         create_reader_hypercore(data_path).await?
     };
+    dbg!();
     let hypercore_wrapper = HypercoreWrapper::from_disk_hypercore(
         hypercore,
         if is_writer {
@@ -343,7 +349,9 @@ async fn run_client(
             Some(result_path.to_string())
         },
     );
+    dbg!();
     tcp_client(port, on_replication_connection, Arc::new(hypercore_wrapper)).await?;
+    dbg!();
     Ok(())
 }
 
@@ -433,28 +441,34 @@ pub fn get_test_key_pair(include_secret: bool) -> PartialKeypair {
 }
 
 #[cfg(feature = "tokio")]
+#[instrument(skip_all)]
 async fn on_replication_connection(
     stream: TcpStream,
     is_initiator: bool,
     hypercore: Arc<HypercoreWrapper>,
 ) -> Result<()> {
+    use tracing::info;
+
     let mut protocol = ProtocolBuilder::new(is_initiator).connect(stream.compat());
     while let Some(event) = protocol.next().await {
         let event = event?;
         match event {
             Event::Handshake(_) => {
+                info!("Event::Handshake");
                 if is_initiator {
                     protocol.open(*hypercore.key()).await?;
                 }
             }
             Event::DiscoveryKey(dkey) => {
+                info!("Event::DiscoveryKey");
                 if hypercore.discovery_key == dkey {
                     protocol.open(*hypercore.key()).await?;
                 } else {
                     panic!("Invalid discovery key");
                 }
             }
             Event::Channel(channel) => {
+                info!("Event::Channel");
                 hypercore.on_replication_peer(channel);
             }
             Event::Close(_dkey) => {