Refactor encryption

Cargo.toml

@@ -39,11 +39,13 @@ futures-lite = "1"
 sha2 = "0.10"
 curve25519-dalek = "4"
 crypto_secretstream = "0.2"
+futures = "0.3.31"
+compact-encoding = "2"
 
 [dependencies.hypercore]
-version = "0.14.0"
-default-features = false
-
+path = "../core"
+#version = "0.14.0"
+#default-features = false
 
 [dev-dependencies]
 async-std = { version = "1.12.0", features = ["attributes", "unstable"] }
@@ -58,8 +60,9 @@ duplexify = "1.1.0"
 sluice = "0.5.4"
 futures = "0.3.13"
 log = "0.4"
-test-log = { version = "0.2.11", default-features = false, features = ["trace"] }
-tracing-subscriber = { version = "0.3.16", features = ["env-filter", "fmt"] }
+tracing-subscriber = { version = "0.3.19", features = ["env-filter", "fmt"] }
+tracing-tree = "0.4.0"
+tokio-util = { version = "0.7.14", features = ["compat"] }
 
 [features]
 default = ["tokio", "sparse"]

benches/pipe.rs

@@ -18,7 +18,7 @@ fn bench_throughput(c: &mut Criterion) {
     env_logger::from_env(env_logger::Env::default().default_filter_or("error")).init();
     let mut group = c.benchmark_group("pipe");
     group.sample_size(10);
-    group.throughput(Throughput::Bytes(SIZE * COUNT * CONNS as u64));
+    group.throughput(Throughput::Bytes(SIZE * COUNT * CONNS));
     group.bench_function("pipe_echo", |b| {
         b.iter(|| {
             task::block_on(async move {
@@ -72,7 +72,7 @@ where
                 debug!("[{}] EVENT {:?}", is_initiator, event);
                 match event {
                     Event::Handshake(_) => {
-                        protocol.open(key.clone()).await?;
+                        protocol.open(key).await?;
                     }
                     Event::DiscoveryKey(_dkey) => {}
                     Event::Channel(channel) => {
@@ -92,7 +92,7 @@ where
             }
             Some(Err(err)) => {
                 error!("ERROR {:?}", err);
-                return Err(err.into());
+                return Err(err);
             }
             None => return Ok(0),
         }
@@ -127,20 +127,17 @@ async fn on_channel_init(i: u64, mut channel: Channel) -> Result<u64> {
     let start = std::time::Instant::now();
 
     while let Some(message) = channel.next().await {
-        match message {
-            Message::Data(mut data) => {
-                len += value_len(&data);
-                debug!("[a] recv {}", index(&data));
-                if index(&data) >= COUNT {
-                    debug!("close at {}", index(&data));
-                    channel.close().await?;
-                    break;
-                } else {
-                    increment_index(&mut data);
-                    channel.send(Message::Data(data)).await?;
-                }
+        if let Message::Data(mut data) = message {
+            len += value_len(&data);
+            debug!("[a] recv {}", index(&data));
+            if index(&data) >= COUNT {
+                debug!("close at {}", index(&data));
+                channel.close().await?;
+                break;
+            } else {
+                increment_index(&mut data);
+                channel.send(Message::Data(data)).await?;
             }
-            _ => {}
         }
     }
     // let bytes = (COUNT * SIZE) as f64;

benches/throughput.rs

@@ -4,7 +4,7 @@ use criterion::{criterion_group, criterion_main, Criterion, Throughput};
 use futures::future::Either;
 use futures::io::{AsyncRead, AsyncWrite};
 use futures::stream::{FuturesUnordered, StreamExt};
-use hypercore_protocol::{schema::*, Duplex};
+use hypercore_protocol::schema::*;
 use hypercore_protocol::{Channel, Event, Message, ProtocolBuilder};
 use log::*;
 use std::time::Instant;
@@ -71,24 +71,23 @@ async fn start_server(address: &str) -> futures::channel::oneshot::Sender<()> {
         // let kill_rx = &mut kill_rx;
         loop {
             match futures::future::select(incoming.next(), &mut kill_rx).await {
-                Either::Left((next, _)) => match next {
-                    Some(Ok(stream)) => {
+                Either::Left((next, _)) => {
+                    if let Some(Ok(stream)) = next {
                         let peer_addr = stream.peer_addr().unwrap();
                         debug!("new connection from {}", peer_addr);
                         task::spawn(async move {
                             onconnection(stream.clone(), stream, false).await;
                         });
                     }
-                    _ => {}
-                },
+                }
                 Either::Right((_, _)) => return,
             }
         }
     });
     kill_tx
 }
 
-async fn onconnection<R, W>(reader: R, writer: W, is_initiator: bool) -> Duplex<R, W>
+async fn onconnection<R, W>(reader: R, writer: W, is_initiator: bool)
 where
     R: AsyncRead + Send + Unpin + 'static,
     W: AsyncWrite + Send + Unpin + 'static,
@@ -101,19 +100,18 @@ where
         // eprintln!("RECV EVENT [{}] {:?}", protocol.is_initiator(), event);
         match event {
             Event::Handshake(_) => {
-                protocol.open(key.clone()).await.unwrap();
+                protocol.open(key).await.unwrap();
             }
             Event::DiscoveryKey(_) => {}
             Event::Channel(channel) => {
                 task::spawn(onchannel(channel, is_initiator));
             }
             Event::Close(_dkey) => {
-                return protocol.release();
+                return;
             }
             _ => {}
         }
     }
-    protocol.release()
 }
 
 async fn onchannel(mut channel: Channel, is_initiator: bool) {
@@ -127,9 +125,8 @@ async fn onchannel(mut channel: Channel, is_initiator: bool) {
 
 async fn channel_server(channel: &mut Channel) {
     while let Some(message) = channel.next().await {
-        match message {
-            Message::Data(_) => channel.send(message).await.unwrap(),
-            _ => {}
+        if let Message::Data(_) = message {
+            channel.send(message).await.unwrap()
         }
     }
 }
@@ -140,24 +137,21 @@ async fn channel_client(channel: &mut Channel) {
     let message = msg_data(0, data.clone());
     channel.send(message).await.unwrap();
     while let Some(message) = channel.next().await {
-        match message {
-            Message::Data(ref msg) => {
-                if index(msg) < COUNT {
-                    let message = msg_data(index(msg) + 1, data.clone());
-                    channel.send(message).await.unwrap();
-                } else {
-                    let time = start.elapsed();
-                    let bytes = COUNT * SIZE;
-                    trace!(
-                        "client completed. {} blocks, {} bytes, {:?}",
-                        index(msg),
-                        bytes,
-                        time
-                    );
-                    break;
-                }
+        if let Message::Data(ref msg) = message {
+            if index(msg) < COUNT {
+                let message = msg_data(index(msg) + 1, data.clone());
+                channel.send(message).await.unwrap();
+            } else {
+                let time = start.elapsed();
+                let bytes = COUNT * SIZE;
+                trace!(
+                    "client completed. {} blocks, {} bytes, {:?}",
+                    index(msg),
+                    bytes,
+                    time
+                );
+                break;
             }
-            _ => {}
         }
     }
 }

examples-nodejs/run.js

@@ -37,7 +37,8 @@ function startRust (mode, key, color, name) {
     color: color || 'blue',
     env: {
       ...process.env,
-      RUST_LOG_STYLE: 'always'
+      RUST_LOG_STYLE: 'always',
+      RUST_LOG: 'trace'
     }
   })
   return rust

examples/replication.rs

@@ -3,23 +3,23 @@ use async_std::net::{TcpListener, TcpStream};
 use async_std::prelude::*;
 use async_std::sync::{Arc, Mutex};
 use async_std::task;
-use env_logger::Env;
 use futures_lite::stream::StreamExt;
 use hypercore::{
     Hypercore, HypercoreBuilder, PartialKeypair, RequestBlock, RequestUpgrade, Storage,
     VerifyingKey,
 };
-use log::*;
 use std::collections::HashMap;
 use std::convert::TryInto;
 use std::env;
 use std::fmt::Debug;
+use std::sync::OnceLock;
+use tracing::{error, info};
 
 use hypercore_protocol::schema::*;
 use hypercore_protocol::{discovery_key, Channel, Event, Message, ProtocolBuilder};
 
 fn main() {
-    init_logger();
+    log();
     if env::args().count() < 3 {
         usage();
     }
@@ -93,8 +93,8 @@ async fn onconnection(
     let mut protocol = ProtocolBuilder::new(is_initiator).connect(stream);
     info!("protocol created, polling for next()");
     while let Some(event) = protocol.next().await {
-        let event = event?;
         info!("protocol event {:?}", event);
+        let event = event?;
         match event {
             Event::Handshake(_) => {
                 if is_initiator {
@@ -299,6 +299,8 @@ async fn onmessage(
                         start: info.length,
                         length: peer_state.remote_length - info.length,
                     }),
+                    manifest: false,
+                    priority: 0,
                 };
                 messages.push(Message::Request(msg));
             }
@@ -405,6 +407,8 @@ async fn onmessage(
                     block: Some(request_block),
                     seek: None,
                     upgrade: None,
+                    manifest: false,
+                    priority: 0,
                 }));
             }
             channel.send_batch(&messages).await.unwrap();
@@ -414,9 +418,21 @@ async fn onmessage(
     Ok(())
 }
 
-/// Init EnvLogger, logging info, warn and error messages to stdout.
-pub fn init_logger() {
-    env_logger::from_env(Env::default().default_filter_or("info")).init();
+#[allow(unused)]
+pub fn log() {
+    use tracing_subscriber::{fmt::format::FmtSpan, EnvFilter};
+    static START_LOGS: OnceLock<()> = OnceLock::new();
+    START_LOGS.get_or_init(|| {
+        tracing_subscriber::fmt()
+            .with_target(true)
+            .with_line_number(true)
+            // print when instrumented funtion enters
+            .with_span_events(FmtSpan::ENTER | FmtSpan::EXIT)
+            .with_file(true)
+            .with_env_filter(EnvFilter::from_default_env()) // Reads `RUST_LOG` environment variable
+            .without_time()
+            .init();
+    });
 }
 
 /// Log a result if it's an error.

src/channels.rs

@@ -13,7 +13,7 @@ use std::pin::Pin;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::task::Poll;
-use tracing::debug;
+use tracing::instrument;
 
 /// A protocol channel.
 ///
@@ -93,7 +93,6 @@ impl Channel {
                 "Channel is closed",
             ));
         }
-        debug!("TX:\n{message:?}\n");
         let message = ChannelMessage::new(self.local_id as u64, message);
         self.outbound_tx
             .send(vec![message])
@@ -122,10 +121,7 @@ impl Channel {
 
         let messages = messages
             .iter()
-            .map(|message| {
-                debug!("TX:\n{message:?}\n");
-                ChannelMessage::new(self.local_id as u64, message.clone())
-            })
+            .map(|message| ChannelMessage::new(self.local_id as u64, message.clone()))
             .collect();
         self.outbound_tx
             .send(messages)
@@ -249,6 +245,7 @@ impl ChannelHandle {
         self.remote_state.as_ref().map(|s| s.remote_id)
     }
 
+    #[instrument(skip_all, fields(local_id = local_id))]
     pub(crate) fn attach_local(&mut self, local_id: usize, key: Key) {
         let local_state = LocalState { local_id, key };
         self.local_state = Some(local_state);
@@ -276,6 +273,7 @@ impl ChannelHandle {
         Ok((&local_state.key, remote_state.remote_capability.as_ref()))
     }
 
+    #[instrument(skip_all)]
     pub(crate) fn open(&mut self, outbound_tx: Sender<Vec<ChannelMessage>>) -> Channel {
         let local_state = self
             .local_state
@@ -433,6 +431,7 @@ impl ChannelMap {
         self.channels.remove(&hdkey);
     }
 
+    #[instrument(skip(self))]
     pub(crate) fn prepare_to_verify(&self, local_id: usize) -> Result<(&Key, Option<&Vec<u8>>)> {
         let channel_handle = self
             .get_local(local_id)
@@ -477,6 +476,7 @@ impl ChannelMap {
         Ok(())
     }
 
+    #[instrument(skip_all)]
     fn alloc_local(&mut self) -> usize {
         let empty_id = self
             .local_id

src/constants.rs

@@ -1,15 +1,8 @@
 /// Seed for the discovery key hash
 pub(crate) const DISCOVERY_NS_BUF: &[u8] = b"hypercore";
 
-/// Default timeout (in seconds)
-pub(crate) const DEFAULT_TIMEOUT: u32 = 20;
-
 /// Default keepalive interval (in seconds)
 pub(crate) const DEFAULT_KEEPALIVE: u32 = 10;
 
-// 16,78MB is the max encrypted wire message size (will be much smaller usually).
-// This limitation stems from the 24bit header.
-pub(crate) const MAX_MESSAGE_SIZE: u64 = 0xFFFFFF;
-
 /// v10: Protocol name
 pub(crate) const PROTOCOL_NAME: &str = "hypercore/alpha";