Merge pull request #85 from monadicus/feat-cannon

feat(cannon): transaction cannon!!!

Cargo.toml

@@ -17,6 +17,7 @@ bincode = "1.3"
 chrono = "0.4"
 clap = { version = "4.5", features = ["derive"] }
 colored = "2"
+external-ip = "4.2.0"
 futures = "0.3"
 futures-util = "0.3.30"
 http = "1.1"

crates/aot/src/authorized.rs

@@ -4,13 +4,16 @@ use anyhow::{bail, Result};
 use clap::Args;
 use rand::{CryptoRng, Rng};
 use snarkvm::{
+    console::program::Network as NetworkTrait,
     ledger::{
         query::Query,
         store::{helpers::memory::ConsensusMemory, ConsensusStore},
     },
     prelude::{
         de, Deserialize, DeserializeExt, Deserializer, Serialize, SerializeStruct, Serializer,
     },
+    synthesizer::process::cost_in_microcredits,
+    utilities::ToBytes,
 };
 use tracing::error;
 
@@ -111,8 +114,8 @@ impl Authorized {
         // Retrieve the execution ID.
         let execution_id: snarkvm::prelude::Field<Network> = function.to_execution_id()?;
         // Determine the base fee in microcredits.
-        let base_fee_in_microcredits = 0;
-        // get_base_fee_in_microcredits(program_id, function_name)?;
+        let base_fee_in_microcredits = estimate_cost(&function)?;
+
         // Authorize the fee.
         let fee = match base_fee_in_microcredits == 0 && priority_fee_in_microcredits == 0 {
             true => None,
@@ -181,6 +184,66 @@ impl Authorized {
     }
 }
 
+fn estimate_cost(func: &Authorization) -> Result<u64> {
+    let transitions = func.transitions();
+
+    let storage_cost = {
+        let mut cost = 0u64;
+
+        cost += 1; // execution version, 1 byte
+        cost += 1; // number of transitions, 1 byte
+
+        // write each transition
+        for transition in transitions.values() {
+            cost += transition.to_bytes_le()?.len() as u64;
+        }
+
+        // state root (this is 32 bytes)
+        cost += <Network as NetworkTrait>::StateRoot::default()
+            .to_bytes_le()?
+            .len() as u64;
+
+        // proof option is_some (1 byte)
+        cost += 1;
+        // Proof<Network> version
+        cost += 1;
+
+        cost += 956; // size of proof with 1 batch size
+
+        /* cost += varuna::Proof::<<Network as Environment>::PairingCurve>::new(
+            todo!("batch_sizes"),
+            todo!("commitments"),
+            todo!("evaluations"),
+            todo!("prover_third_message"),
+            todo!("prover_fourth_message"),
+            todo!("pc_proof"),
+        )?
+        .to_bytes_le()?
+        .len() as u64; */
+
+        cost
+    };
+    //execution.size_in_bytes().map_err(|e| e.to_string())?;
+
+    // Compute the finalize cost in microcredits.
+    let mut finalize_cost = 0u64;
+    // Iterate over the transitions to accumulate the finalize cost.
+    for (_key, transition) in transitions {
+        // Retrieve the function name, program id, and program.
+        let function_name = transition.function_name();
+        let stack = PROCESS.get_stack(transition.program_id())?;
+        let cost = cost_in_microcredits(stack, function_name)?;
+
+        // Accumulate the finalize cost.
+        if let Some(cost) = finalize_cost.checked_add(cost) {
+            finalize_cost = cost;
+        } else {
+            bail!("The finalize cost computation overflowed for an execution")
+        };
+    }
+    Ok(storage_cost + finalize_cost)
+}
+
 impl Serialize for Authorized {
     /// Serializes the authorization into string or bytes.
     fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {

crates/aot/src/ledger/truncate.rs

@@ -46,7 +46,7 @@ impl Truncate {
                 for i in 1..target_height {
                     let block = db_ledger.get_block(i)?;
                     let buf = block.to_bytes_le()?;
-                    tracing::info!("Writing block {i}... {}", buf.len());
+                    // println!("Writing block {i}... {}", buf.len());
 
                     unistd::write(&write_fd, &(buf.len() as u32).to_le_bytes())?;
                     unistd::write(&write_fd, &buf)?;
@@ -77,14 +77,22 @@ impl Truncate {
                     while read < amount as usize {
                         read += unistd::read(read_fd, &mut buf[read..])?;
                     }
-                    tracing::info!(
-                        "Reading block {}... {}",
-                        db_ledger.latest_height() + 1,
-                        buf.len()
-                    );
-
                     let block = Block::from_bytes_le(&buf)?;
-                    db_ledger.advance_to_next_block(&block)?;
+                    if db_ledger.latest_height() + 1 != block.height() {
+                        println!(
+                            "Skipping block {}, waiting for {}",
+                            block.height(),
+                            db_ledger.latest_height() + 1,
+                        );
+                    } else {
+                        println!(
+                            "Reading block {}... {}",
+                            db_ledger.latest_height() + 1,
+                            buf.len()
+                        );
+
+                        db_ledger.advance_to_next_block(&block)?;
+                    }
                 }
 
                 unistd::close(read_fd.as_raw_fd())?;

crates/snot-agent/Cargo.toml

@@ -9,7 +9,7 @@ edition = "2021"
 anyhow.workspace = true
 bincode.workspace = true
 clap.workspace = true
-external-ip = "4.2.0"
+external-ip.workspace = true
 futures.workspace = true
 futures-util.workspace = true
 http.workspace = true

crates/snot-agent/src/rpc.rs

@@ -405,11 +405,73 @@ impl AgentService for AgentRpcServer {
             .map_err(|_| AgentError::FailedToParseJson)
     }
 
+    async fn broadcast_tx(self, _: context::Context, tx: String) -> Result<(), AgentError> {
+        if !matches!(
+            self.state.agent_state.read().await.deref(),
+            AgentState::Node(_, _)
+        ) {
+            return Err(AgentError::InvalidState);
+        }
+
+        let url = format!(
+            "http://127.0.0.1:{}/mainnet/transaction/broadcast",
+            self.state.cli.rest
+        );
+        let response = reqwest::Client::new()
+            .post(url)
+            .header("Content-Type", "application/json")
+            .body(tx)
+            .send()
+            .await
+            .map_err(|_| AgentError::FailedToMakeRequest)?;
+        if response.status().is_success() {
+            Ok(())
+        } else {
+            Err(AgentError::FailedToMakeRequest)
+        }
+    }
+
     async fn get_metric(self, _: context::Context, metric: AgentMetric) -> f64 {
         let metrics = self.state.metrics.read().await;
 
         match metric {
             AgentMetric::Tps => metrics.tps.get(),
         }
     }
+
+    async fn execute_authorization(
+        self,
+        _: context::Context,
+        _env_id: usize,
+        query: String,
+        auth: String,
+    ) -> Result<(), AgentError> {
+        info!("executing authorization...");
+        // TODO: ensure binary associated with this env_id is present
+
+        let res = Command::new(dbg!(self.state.cli.path.join(SNARKOS_FILE)))
+            .stdout(std::io::stdout())
+            .stderr(std::io::stderr())
+            .arg("execute")
+            .arg("--query")
+            .arg(&format!("http://{}{query}", self.state.endpoint))
+            .arg(auth)
+            .spawn()
+            .map_err(|e| {
+                warn!("failed to spawn auth exec process: {e}");
+                AgentError::FailedToSpawnProcess
+            })?
+            .wait()
+            .await
+            .map_err(|e| {
+                warn!("auth exec process failed: {e}");
+                AgentError::ProcessFailed
+            })?;
+
+        if !res.success() {
+            warn!("auth exec process exited with status: {res}");
+            return Err(AgentError::ProcessFailed);
+        }
+        Ok(())
+    }
 }

crates/snot-common/Cargo.toml

@@ -8,6 +8,7 @@ futures.workspace = true
 lazy_static.workspace = true
 regex.workspace = true
 serde.workspace = true
+serde_json.workspace = true
 tarpc.workspace = true
 thiserror.workspace = true
 tokio.workspace = true

crates/snot-common/src/rpc/agent.rs

@@ -22,6 +22,18 @@ pub trait AgentService {
 
     /// Get the state root from the running node
     async fn get_state_root() -> Result<String, AgentError>;
+
+    /// Broadcast a transaction locally
+    async fn broadcast_tx(tx: String) -> Result<(), AgentError>;
+
+    /// Locally execute an authorization, using the given query
+    /// environment id is passed so the agent can determine which aot binary to use
+    async fn execute_authorization(
+        env_id: usize,
+        query: String,
+        auth: String,
+    ) -> Result<(), AgentError>;
+
     async fn get_metric(metric: AgentMetric) -> f64;
 }
 
@@ -45,6 +57,10 @@ pub enum AgentError {
     FailedToParseJson,
     #[error("failed to make a request")]
     FailedToMakeRequest,
+    #[error("failed to spawn a process")]
+    FailedToSpawnProcess,
+    #[error("process failed")]
+    ProcessFailed,
 }
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum AgentMetric {

crates/snot-common/src/state.rs

@@ -199,3 +199,12 @@ impl Display for NodeKey {
         Ok(())
     }
 }
+
+impl Serialize for NodeKey {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        serializer.serialize_str(&self.to_string())
+    }
+}

crates/snot/Cargo.toml

@@ -11,6 +11,7 @@ bincode.workspace = true
 chrono = { workspace = true, features = ["serde"] }
 clap.workspace = true
 duration-str = { version = "0.7", default-features = false }
+external-ip.workspace = true
 futures-util.workspace = true
 hmac = "0.12.1"
 indexmap.workspace = true
@@ -19,7 +20,7 @@ lazy_static.workspace = true
 rand.workspace = true
 rand_chacha.workspace = true
 regex.workspace = true
-reqwest = { workspace = true, features = ["stream"] }
+reqwest = { workspace = true, features = ["stream", "json"] }
 serde.workspace = true
 serde_json.workspace = true
 serde_yaml.workspace = true

crates/snot/src/cannon/authorized.rs

@@ -0,0 +1,66 @@
+use std::path::PathBuf;
+
+use anyhow::{ensure, Result};
+use tokio::process::Command;
+
+#[derive(Clone, Debug)]
+pub enum Authorize {
+    TransferPublic {
+        private_key: String,
+        recipient: String,
+        amount: u64,
+        priority_fee: u64,
+    },
+}
+
+impl Authorize {
+    pub async fn run(self, bin: &PathBuf) -> Result<serde_json::Value> {
+        let mut command = Command::new(bin);
+        command
+            .stdout(std::io::stdout())
+            .stderr(std::io::stderr())
+            .arg("authorize");
+
+        match self {
+            Self::TransferPublic {
+                private_key,
+                recipient,
+                amount,
+                priority_fee,
+            } => {
+                command
+                    .arg("transfer-public")
+                    .arg("--private-key")
+                    .arg(private_key)
+                    .arg("--recipient")
+                    .arg(recipient)
+                    .arg("--amount")
+                    .arg(amount.to_string())
+                    .arg("--priority-fee")
+                    .arg(priority_fee.to_string());
+            }
+        }
+
+        command.arg("--broadcast");
+
+        let res = command.output().await?;
+
+        if !res.status.success() {
+            return Err(anyhow::anyhow!(
+                "command failed with status {}: {}",
+                res.status,
+                String::from_utf8_lossy(&res.stderr)
+            ));
+        }
+
+        let blob: serde_json::Value = serde_json::from_slice(&res.stdout)?;
+
+        ensure!(blob.is_object(), "expected JSON object in response");
+        ensure!(
+            blob.get("function").is_some() && blob.get("broadcast").is_some(),
+            "expected function, fee, and broadcast fields in response"
+        );
+
+        Ok(blob)
+    }
+}