Sim2h recording & playback of connections and messages

.gitignore

@@ -7,6 +7,7 @@ node_modules
 package-lock.json
 **/*.rs.bk
 *.log
+*.cassette
 net_ipc/tests/node-p2p-ipc
 build_docs_key
 *.orig

Cargo.toml

@@ -20,7 +20,10 @@ members = [
   "crates/trycp_server",
   "crates/sim1h",
   "crates/sim2h",
+  "crates/sim2h_client",
   "crates/sim2h_server",
+  "crates/walkman_cli",
+  "crates/walkman_types",
   "crates/wasm_utils"
 ]
 exclude = [

crates/cli/Cargo.toml

@@ -18,6 +18,7 @@ holochain_locksmith = { version = "=0.0.42-alpha5", path = "../locksmith" }
 holochain_tracing_macros = "0.0.14"
 newrelic="0.2"
 sim2h = { version = "=0.0.42-alpha5", path = "../sim2h" }
+sim2h_client = { version = "=0.0.42-alpha5", path = "../sim2h_client" }
 lib3h_crypto_api = "=0.0.31"
 in_stream = { version = "=0.0.42-alpha5", path = "../in_stream" }
 url2 = "=0.0.4"

crates/cli/src/cli/mod.rs

@@ -7,6 +7,7 @@ pub mod package;
 pub mod run;
 mod sim2h_client;
 pub mod test;
+mod walkman;
 
 pub use self::{
     chain_log::{chain_list, chain_log},
@@ -18,4 +19,5 @@ pub use self::{
     run::{get_interface_type_string, hc_run_bundle_configuration, hc_run_configuration, run},
     sim2h_client::sim2h_client,
     test::{test, TEST_DIR_NAME},
+    walkman::walkman,
 };

crates/cli/src/cli/sim2h_client.rs

@@ -2,13 +2,9 @@ use crate::NEW_RELIC_LICENSE_KEY;
 use dns_lookup::lookup_host;
 use in_stream::*;
 use lib3h_crypto_api::CryptoSystem;
-use lib3h_protocol::data_types::*;
 use lib3h_sodium::SodiumCryptoSystem;
-use sim2h::{
-    crypto::{Provenance, SignedWireMessage},
-    WireMessage, WIRE_VERSION,
-};
-use std::sync::{Arc, Mutex};
+use sim2h::{WireMessage, WIRE_VERSION};
+use sim2h_client::Sim2hClient;
 use url2::prelude::*;
 
 #[holochain_tracing_macros::newrelic_autotrace(HOLOCHAIN_CLI)]
@@ -36,8 +32,8 @@ pub fn sim2h_client(url_string: String, message_string: String) -> Result<(), St
     let url = Url2::parse(format!("{}://{}:{}", url.scheme(), ip, maybe_port.unwrap()));
 
     println!("connecting to: {}", url);
-    let mut job = Job::new(&url)?;
-    job.send_wire(match message_string.as_ref() {
+    let mut client = Sim2hClient::new(&url)?;
+    client.send_wire(match message_string.as_ref() {
         "ping" => WireMessage::Ping,
         "hello" => WireMessage::Hello(WIRE_VERSION),
         "status" => WireMessage::Status,
@@ -54,7 +50,7 @@ pub fn sim2h_client(url_string: String, message_string: String) -> Result<(), St
     loop {
         std::thread::sleep(std::time::Duration::from_millis(10));
         let mut frame = WsFrame::default();
-        match job.connection.read(&mut frame) {
+        match client.connection().read(&mut frame) {
             Ok(_) => {
                 if let WsFrame::Binary(b) = frame {
                     let msg: WireMessage = serde_json::from_slice(&b).unwrap();
@@ -78,102 +74,3 @@ pub fn sim2h_client(url_string: String, message_string: String) -> Result<(), St
 thread_local! {
     pub static CRYPTO: Box<dyn CryptoSystem> = Box::new(SodiumCryptoSystem::new());
 }
-struct Job {
-    agent_id: String,
-    #[allow(dead_code)]
-    pub_key: Arc<Mutex<Box<dyn lib3h_crypto_api::Buffer>>>,
-    sec_key: Arc<Mutex<Box<dyn lib3h_crypto_api::Buffer>>>,
-    connection: InStreamWss<InStreamTcp>,
-    //    wss_connection: InStreamWss<InStreamTls<InStreamTcp>>,
-}
-
-#[holochain_tracing_macros::newrelic_autotrace(HOLOCHAIN_CLI)]
-impl Job {
-    pub fn new(connect_uri: &Url2) -> Result<Self, String> {
-        let (pub_key, sec_key) = CRYPTO.with(|crypto| {
-            let mut pub_key = crypto.buf_new_insecure(crypto.sign_public_key_bytes());
-            let mut sec_key = crypto.buf_new_secure(crypto.sign_secret_key_bytes());
-            crypto.sign_keypair(&mut pub_key, &mut sec_key).unwrap();
-            (pub_key, sec_key)
-        });
-        let enc = hcid::HcidEncoding::with_kind("hcs0").map_err(|e| format!("{}", e))?;
-        let agent_id = enc.encode(&*pub_key).unwrap();
-        println!("Generated agent id: {}", agent_id);
-        let connection = await_in_stream_connect(connect_uri)
-            .map_err(|e| format!("Error awaiting connection: {}", e))?;
-        println!("Await successfull");
-        let out = Self {
-            agent_id,
-            pub_key: Arc::new(Mutex::new(pub_key)),
-            sec_key: Arc::new(Mutex::new(sec_key)),
-            connection,
-        };
-
-        Ok(out)
-    }
-
-    /// sign a message and send it to sim2h
-    pub fn send_wire(&mut self, message: WireMessage) {
-        println!("Sending wire message to sim2h: {:?}", message);
-        let payload: Opaque = message.into();
-        let payload_buf: Box<dyn lib3h_crypto_api::Buffer> = Box::new(payload.clone().as_bytes());
-        let sig = base64::encode(
-            &*CRYPTO
-                .with(|crypto| {
-                    let mut sig = crypto.buf_new_insecure(crypto.sign_bytes());
-                    crypto
-                        .sign(&mut sig, &payload_buf, &*self.sec_key.lock().unwrap())
-                        .unwrap();
-                    sig
-                })
-                .read_lock(),
-        );
-        let signed_message = SignedWireMessage {
-            provenance: Provenance::new(self.agent_id.clone().into(), sig.into()),
-            payload,
-        };
-        let to_send: Opaque = signed_message.into();
-        self.connection.write(to_send.as_bytes().into()).unwrap();
-    }
-}
-
-#[holochain_tracing_macros::newrelic_autotrace(HOLOCHAIN_CLI)]
-fn await_in_stream_connect(connect_uri: &Url2) -> Result<InStreamWss<InStreamTcp>, String> {
-    let timeout = std::time::Instant::now()
-        .checked_add(std::time::Duration::from_millis(60000))
-        .unwrap();
-
-    let mut read_frame = WsFrame::default();
-
-    // keep trying to connect
-    loop {
-        //        let config = WssConnectConfig::new(TlsConnectConfig::new(TcpConnectConfig::default()));
-        let config = WssConnectConfig::new(TcpConnectConfig::default());
-        let mut connection =
-            InStreamWss::connect(connect_uri, config).map_err(|e| format!("{}", e))?;
-        connection.write(WsFrame::Ping(b"".to_vec())).unwrap();
-
-        loop {
-            let mut err = false;
-            match connection.read(&mut read_frame) {
-                Ok(_) => return Ok(connection),
-                Err(e) if e.would_block() => (),
-                Err(_) => {
-                    err = true;
-                }
-            }
-
-            if std::time::Instant::now() >= timeout {
-                Err("could not connect within timeout".to_string())?
-            }
-
-            if err {
-                break;
-            }
-
-            std::thread::sleep(std::time::Duration::from_millis(10));
-        }
-
-        std::thread::sleep(std::time::Duration::from_millis(500));
-    }
-}

crates/cli/src/cli/walkman.rs

@@ -0,0 +1,6 @@
+use std::path::PathBuf;
+// use crate::cli::sim2h_client::Sim2hClient;
+
+pub fn walkman(_cassette: PathBuf) {
+    println!("run walkman");
+}

crates/cli/src/main.rs

@@ -11,9 +11,9 @@ extern crate holochain_persistence_api;
 extern crate holochain_persistence_file;
 extern crate json_patch;
 extern crate lib3h_crypto_api;
-extern crate lib3h_protocol;
 extern crate lib3h_sodium;
 extern crate sim2h;
+extern crate sim2h_client;
 extern crate structopt;
 #[macro_use]
 extern crate failure;
@@ -164,6 +164,11 @@ enum Cli {
         /// message to send to the sim2h server ('ping' or 'status')
         message: String,
     },
+    Walkman {
+        #[structopt()]
+        /// Path to walkman cassette file for playback
+        cassette: PathBuf,
+    },
 }
 arg_enum! {
     #[derive(Debug)]
@@ -330,10 +335,12 @@ fn run() -> HolochainResult<()> {
         }
 
         Cli::Sim2hClient { url, message } => {
-            println!("url: {}", &url);
-            println!("message: {}", &message);
             cli::sim2h_client(url, message)?;
         }
+
+        Cli::Walkman { cassette } => {
+            cli::walkman(cassette);
+        }
     }
 
     Ok(())

crates/metrics/src/logger.rs

@@ -21,7 +21,7 @@ impl LoggerMetricPublisher {
 impl MetricPublisher for LoggerMetricPublisher {
     fn publish(&mut self, metric: &Metric) {
         let log_line: LogLine = metric.into();
-        debug!("{}", log_line);
+        trace!("{}", log_line);
     }
 }
 

crates/sim1h/Cargo.toml

@@ -21,7 +21,7 @@ holochain_tracing = "=0.0.13"
 uuid = { version = "0.4", features = ["v4"] }
 log = "0.4.8"
 env_logger = "=0.6.1"
-lazy_static = "1.2.0"
+lazy_static = "=1.4.0"
 holochain_persistence_api = "=0.0.13"
 holochain_json_api = "=0.0.17"
 url = "=2.1.0"

crates/sim2h/Cargo.toml

@@ -26,6 +26,7 @@ holochain_tracing = "=0.0.13"
 holochain_core_types = { version = "=0.0.42-alpha5", path = "../core_types" }
 holochain_locksmith = { version = "=0.0.42-alpha5", path = "../locksmith" }
 holochain_metrics = { version = "=0.0.42-alpha5", path = "../metrics" }
+holochain_walkman_types = { version = "=0.0.42-alpha5", path = "../walkman_types" }
 holochain_common = { version = "=0.0.42-alpha5", path = "../common" }
 holochain_tracing_macros = "0.0.14"
 in_stream = { version = "=0.0.42-alpha5", path = "../in_stream" }
@@ -52,5 +53,8 @@ native-tls = "=0.2.3"
 openssl = "=0.10.25"
 tungstenite = "=0.9.2"
 threadpool = "=1.7.1"
-chashmap = "=2.2.2"
 newrelic="0.2"
+
+[dev-dependencies]
+sim2h_client = { version = "=0.0.42-alpha5", path = "../sim2h_client" }
+chashmap = "=2.2.2"

crates/sim2h/src/debug.rs

@@ -0,0 +1,13 @@
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct DebugData {
+    pub limbo: DebugLimboData,
+    pub msg_queue_size: usize,
+    pub wss_queue_size: usize,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct DebugLimboData {
+    pub total_connections: usize,
+    pub total_messages: usize,
+    pub max_messages: usize,
+}

crates/sim2h/src/lib.rs

@@ -10,6 +10,7 @@ extern crate num_cpus;
 extern crate serde;
 #[macro_use]
 extern crate lazy_static;
+extern crate holochain_walkman_types;
 extern crate newrelic;
 
 #[macro_use]
@@ -21,6 +22,7 @@ mod naive_sharding;
 pub mod cache;
 pub mod connection_state;
 pub mod crypto;
+pub mod debug;
 pub mod error;
 use lib3h_protocol::types::{AgentPubKey, AspectHash, EntryHash};
 mod message_log;
@@ -31,6 +33,11 @@ pub use crate::message_log::MESSAGE_LOGGER;
 use crate::{crypto::*, error::*, naive_sharding::entry_location};
 use cache::*;
 use connection_state::*;
+use holochain_locksmith::Mutex;
+use holochain_metrics::{config::MetricPublisherConfig, Metric};
+use holochain_walkman_types::{walkman_log_sim2h, WalkmanSim2hEvent};
+use in_stream::*;
+use lib3h::rrdht_util::*;
 use lib3h_crypto_api::CryptoSystem;
 use lib3h_protocol::{
     data_types::{
@@ -50,21 +57,29 @@ use futures::{
     future::{BoxFuture, FutureExt},
     stream::StreamExt,
 };
-use in_stream::*;
 use log::*;
 use rand::{seq::SliceRandom, thread_rng};
 use std::{
     collections::{HashMap, HashSet},
     convert::TryFrom,
     sync::Arc,
+    time::Instant,
 };
-
-use holochain_locksmith::Mutex;
-use holochain_metrics::{config::MetricPublisherConfig, Metric};
+lazy_static! {
+    static ref HOLOCHAIN_WALKMAN_SIM2H: bool = std::env::var("HOLOCHAIN_WALKMAN_SIM2H").is_ok();
+}
 
 /// if we can't acquire a lock in 20 seconds, panic!
 const MAX_LOCK_TIMEOUT: u64 = 20000;
 
+fn walkman_log<F: FnOnce() -> WalkmanSim2hEvent>(event: F) {
+    if *HOLOCHAIN_WALKMAN_SIM2H {
+        let log_item = event();
+        let json =
+            serde_json::to_string(&walkman_log_sim2h(log_item)).expect("Serialized walkman event");
+        println!("<walkman>{}</walkman>", json);
+    }
+}
 //set up license_key
 new_relic_setup!("NEW_RELIC_LICENSE_KEY");
 
@@ -99,7 +114,6 @@ impl<T> SendExt<T> for crossbeam_channel::Sender<T> {
         }
     }
 }
-
 const RETRY_FETCH_MISSING_ASPECTS_INTERVAL_MS: u64 = 30000; // 30 seconds
 
 fn conn_lifecycle(desc: &str, uuid: &str, obj: &ConnectionState, uri: &Lib3hUri) {
@@ -405,7 +419,7 @@ pub struct Sim2h {
     connection_mgr_evt_recv: ConnectionMgrEventRecv,
     num_ticks: u64,
     /// when should we try to resync nodes that are still missing aspect data
-    missing_aspects_resync: std::time::Instant,
+    missing_aspects_resync: Instant,
     dht_algorithm: DhtAlgorithm,
     recv_com: tokio::sync::mpsc::UnboundedReceiver<Sim2hCom>,
     sim2h_handle: Sim2hHandle,
@@ -502,6 +516,7 @@ impl Sim2h {
                         .connection_states
                         .insert(url.clone(), (nanoid::simple(), ConnectionState::new()));
 
+                    walkman_log(|| WalkmanSim2hEvent::Connect(url.to_string()));
                     state.connection_mgr.connect(url, wss);
                 }
             });
@@ -806,6 +821,12 @@ impl Sim2h {
                 Ok((signed_message.provenance.source().into(), wire_message))
             })() {
                 Ok((source, wire_message)) => {
+                    walkman_log(|| {
+                        let signed_message = SignedWireMessage::try_from(payload.clone()).unwrap();
+                        let msg_serialized = serde_json::to_string(&signed_message)
+                            .expect("SignedWireMessage serialized");
+                        WalkmanSim2hEvent::Message(url.to_string(), msg_serialized)
+                    });
                     sim2h_handle.handle_message(url, wire_message, source)
                 }
                 Err(error) => {

crates/sim2h/src/sim2h_state.rs

@@ -1,5 +1,4 @@
 use crate::*;
-use lib3h::rrdht_util::*;
 
 pub(crate) struct Sim2hState {
     pub(crate) crypto: Box<dyn CryptoSystem>,
@@ -86,6 +85,7 @@ impl Sim2hState {
                 }
             }
         }
+        walkman_log(|| WalkmanSim2hEvent::Disconnect(uri.0.to_string()));
         trace!("disconnect done");
     }