Merge branch 'main' into feat-cannon

.github/workflows/rust.yml

@@ -2,9 +2,9 @@ name: Rust
 
 on:
   push:
-    branches: ['main']
+    branches: ["main"]
   pull_request:
-    branches: ['main']
+    branches: ["main"]
 
 env:
   CARGO_TERM_COLOR: always
@@ -39,14 +39,55 @@ jobs:
         with:
           components: rustc-codegen-cranelift-preview
 
+      - uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: snarkos-test
+          key: cache-v1
+
       - name: Build
         working-directory: ./snarkos-test
         env:
           RUSTFLAGS: -Zcodegen-backend=cranelift
         run: cargo +nightly build --verbose
 
+  test:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+        with:
+          path: snarkos-test
+
+      - name: Checkout snarkOS
+        uses: actions/checkout@v4
+        with:
+          repository: AleoNet/snarkOS
+          path: snarkos
+
+      - name: Checkout snarkVM
+        uses: actions/checkout@v4
+        with:
+          repository: AleoNet/snarkVM
+          path: snarkvm
+
+      - name: Use mold linker
+        uses: rui314/setup-mold@v1
+
+      - name: Install nightly and cranelift
+        uses: dtolnay/rust-toolchain@nightly
+        with:
+          components: rustc-codegen-cranelift-preview
+
+      - uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: snarkos-test
+          key: cache-v1
+
+      - uses: taiki-e/install-action@nextest
+
       - name: Run tests
         working-directory: ./snarkos-test
         env:
           RUSTFLAGS: -Zcodegen-backend=cranelift
-        run: cargo +nightly test --verbose
+        run: cargo +nightly nextest run --all --verbose --fail-fast

Cargo.toml

@@ -14,6 +14,7 @@ axum = "0.7.5"
 anyhow = "1"
 bimap = "0.6"
 bincode = "1.3"
+chrono = "0.4"
 clap = { version = "4.5", features = ["derive"] }
 colored = "2"
 external-ip = "4.2.0"
@@ -40,6 +41,7 @@ tracing = "0.1"
 tracing-appender = "0.2"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 url = "2.5"
+uuid = "1.8"
 nix = { version = "0.28", features = ["process"] }
 
 snarkos-account = { path = "../snarkos/account" }

crates/snot-agent/src/main.rs

@@ -57,9 +57,9 @@ async fn main() {
     tracing_subscriber::registry()
         .with(
             tracing_subscriber::EnvFilter::builder()
-                .with_env_var("RUST_LOG")
+                .with_env_var("SNOT_AGENT_LOG")
                 .with_default_directive(LevelFilter::TRACE.into())
-                .parse_lossy("")
+                .from_env_lossy()
                 .add_directive("neli=off".parse().unwrap())
                 .add_directive("hyper_util=off".parse().unwrap())
                 .add_directive("reqwest=off".parse().unwrap())

crates/snot-cli/src/commands/env.rs

@@ -41,29 +41,34 @@ impl Env {
         let client = reqwest::blocking::Client::new();
 
         use Commands::*;
-        match self.command {
+        let response = match self.command {
             Prepare { spec } => {
                 let ep = format!("{}/api/v1/env/prepare", self.url);
                 let file: String = std::fs::read_to_string(spec)?;
 
-                let id: Value = client.post(ep).body(file).send()?.json()?;
-                println!("{}", serde_json::to_string(&id)?);
-                Ok(())
+                client.post(ep).body(file).send()?
             }
 
             Start { id } => {
                 let ep = format!("{}/api/v1/env/{id}", self.url);
 
-                client.post(ep).send()?;
-                Ok(())
+                client.post(ep).send()?
             }
 
             Stop { id } => {
                 let ep = format!("{}/api/v1/env/{id}", self.url);
 
-                client.delete(ep).send()?;
-                Ok(())
+                client.delete(ep).send()?
             }
-        }
+        };
+
+        let value = match response.content_length() {
+            Some(0) | None => None,
+            _ => response.json::<Value>().map(Some)?,
+        };
+
+        println!("{}", serde_json::to_string_pretty(&value)?);
+
+        Ok(())
     }
 }

crates/snot/Cargo.toml

@@ -8,6 +8,7 @@ anyhow.workspace = true
 axum = { workspace = true, features = ["ws"] }
 bimap.workspace = true
 bincode.workspace = true
+chrono = { workspace = true, features = ["serde"] }
 clap.workspace = true
 duration-str = { version = "0.7", default-features = false }
 external-ip.workspace = true
@@ -37,4 +38,5 @@ tracing-appender.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true
 url = { workspace = true, features = ["serde"] }
+uuid = { workspace = true, features = ["v4", "fast-rng"] }
 wildmatch = "2.3.3"

crates/snot/src/env/mod.rs

@@ -362,9 +362,7 @@ impl Environment {
             .filter(|(key, _)| targets.matches(key))
             .filter_map(move |(key, value)| match value {
                 EnvPeer::Internal(id) => {
-                    let Some(agent) = pool.get(id) else {
-                        return None;
-                    };
+                    let agent = pool.get(id)?;
 
                     Some(AgentPeer::Internal(
                         *id,

crates/snot/src/logging.rs

@@ -0,0 +1,93 @@
+use axum::{
+    async_trait,
+    extract::{FromRequestParts, Request},
+    http::{request::Parts, Method, StatusCode, Uri},
+    middleware::Next,
+    response::Response,
+};
+use chrono::{DateTime, Utc};
+use serde::Serialize;
+use serde_json::{json, Value};
+use tracing::debug;
+use uuid::Uuid;
+
+#[derive(Debug, Clone)]
+pub struct ReqStamp {
+    pub uuid: Uuid,
+    pub time_in: DateTime<Utc>,
+}
+
+pub async fn req_stamp(mut req: Request, next: Next) -> Response {
+    let time_in = Utc::now();
+    let uuid = Uuid::new_v4();
+
+    req.extensions_mut().insert(ReqStamp { uuid, time_in });
+
+    next.run(req).await
+}
+
+#[async_trait]
+impl<S: Send + Sync> FromRequestParts<S> for ReqStamp {
+    // TODO replace with our own error type that implements IntoResponse
+    type Rejection = StatusCode;
+
+    async fn from_request_parts(
+        parts: &mut Parts,
+        _state: &S,
+    ) -> core::result::Result<Self, Self::Rejection> {
+        parts
+            .extensions
+            .get::<ReqStamp>()
+            .cloned()
+            .ok_or(StatusCode::INTERNAL_SERVER_ERROR)
+    }
+}
+
+#[derive(Serialize)]
+struct RequestLogLine {
+    /// Unique request identifier.
+    uuid: String,
+    /// Timestamp(rfc3339) of the log line.
+    timestamp: String,
+    /// Timestamp(rfc3339) of the request.
+    time_in: String,
+    /// Duration of the request in milliseconds.
+    duration_ms: i64,
+
+    /// HTTP path of the request.
+    http_path: String,
+    /// HTTP method of the request.
+    http_method: String,
+
+    // TODO: error handling
+    /// The error variant.
+    error_type: Option<String>,
+    /// The error data.
+    error_data: Option<Value>,
+}
+
+pub async fn log_request(uri: Uri, method: Method, req_stamp: ReqStamp, res: Response) -> Response {
+    // TODO: grab error data from response
+    // something like:
+    // res.extensions_mut().get::<OurErrorType>();
+
+    let ReqStamp { uuid, time_in } = req_stamp;
+    let now = Utc::now();
+    let duration = now - time_in;
+
+    let log_line = RequestLogLine {
+        uuid: uuid.to_string(),
+        timestamp: now.to_rfc3339(),
+        time_in: time_in.to_rfc3339(),
+        duration_ms: duration.num_milliseconds(),
+        http_path: uri.to_string(),
+        http_method: method.to_string(),
+        error_type: None,
+        error_data: None,
+    };
+
+    // TODO: send to logging services
+    debug!("REQUEST LOG LINE:\n{}", json!(log_line));
+
+    res
+}

crates/snot/src/main.rs

@@ -8,6 +8,7 @@ use tracing_subscriber::prelude::*;
 pub mod cannon;
 pub mod cli;
 pub mod env;
+pub mod logging;
 pub mod schema;
 pub mod server;
 pub mod state;
@@ -21,7 +22,8 @@ async fn main() {
     };
 
     let env_filter = env_filter
-        .parse_lossy("")
+        .with_env_var("SNOT_LOG")
+        .from_env_lossy()
         .add_directive("hyper_util=off".parse().unwrap())
         .add_directive("hyper=off".parse().unwrap())
         .add_directive("reqwest=off".parse().unwrap())
@@ -34,6 +36,8 @@ async fn main() {
         .add_directive("tower_http::trace::on_request=off".parse().unwrap())
         .add_directive("tower_http::trace::on_response=off".parse().unwrap());
 
+    dbg!(env_filter.to_string());
+
     let (stdout, _guard) = tracing_appender::non_blocking(io::stdout());
 
     let output = tracing_subscriber::fmt::layer().with_writer(stdout);

crates/snot/src/server/mod.rs

@@ -8,6 +8,7 @@ use axum::{
         State, WebSocketUpgrade,
     },
     http::HeaderMap,
+    middleware,
     response::IntoResponse,
     routing::get,
     Router,
@@ -20,7 +21,6 @@ use snot_common::{
 use surrealdb::Surreal;
 use tarpc::server::Channel;
 use tokio::select;
-use tower_http::trace::{DefaultMakeSpan, TraceLayer};
 use tracing::{info, warn};
 
 use self::{
@@ -29,6 +29,7 @@ use self::{
 };
 use crate::{
     cli::Cli,
+    logging::{log_request, req_stamp},
     server::rpc::{MuxedMessageIncoming, MuxedMessageOutgoing},
     state::{Agent, AppState, GlobalState},
 };
@@ -57,16 +58,19 @@ pub async fn start(cli: Cli) -> Result<()> {
         // /env/<id>/ledger/* - ledger query service reverse proxying /mainnet/latest/stateRoot
         .nest("/content", content::init_routes(&state).await)
         .with_state(Arc::new(state))
-        .layer(
-            TraceLayer::new_for_http().make_span_with(DefaultMakeSpan::new().include_headers(true)),
-            //.on_request(|request: &Request<Body>, _span: &Span| {
-            //    tracing::info!("req {} - {}", request.method(), request.uri());
-            //})
-            //.on_response(|response: &Response, _latency: Duration, span: &Span| {
-            //    span.record("status_code", &tracing::field::display(response.status()));
-            //    tracing::info!("res {}", response.status())
-            //}),
-        );
+        .layer(middleware::map_response(log_request))
+        .layer(middleware::from_fn(req_stamp));
+    // .layer(
+    // TraceLayer::new_for_http().make_span_with(DefaultMakeSpan::new().
+    // include_headers(true)),
+    //.on_request(|request: &Request<Body>, _span: &Span| {
+    //    tracing::info!("req {} - {}", request.method(), request.uri());
+    //})
+    //.on_response(|response: &Response, _latency: Duration, span: &Span| {
+    //    span.record("status_code", &tracing::field::display(response.status()));
+    //    tracing::info!("res {}", response.status())
+    //}),
+    // );
 
     let listener = tokio::net::TcpListener::bind("0.0.0.0:1234").await?;
     axum::serve(listener, app).await?;

crates/snot/src/state.rs

@@ -274,9 +274,7 @@ pub fn resolve_addrs(
             }
 
             // if the agent has no addresses, skip it
-            let Some(addrs) = addr_map.get(id) else {
-                return None;
-            };
+            let addrs = addr_map.get(id)?;
 
             // if there are no external addresses in the entire addr map,
             // use the first internal address