Merge pull request #27 from cardano-scaling/sample-transactions

Endpoint to fetch the next N transactions from a given node

Author: Quantumplation

.gitignore

@@ -12,4 +12,4 @@ utxo.json
 .pre-commit-config.yaml
 admin.sk
 run.sh
-/keys
+crates/rpc/keys

Cargo.lock

@@ -43,6 +43,7 @@ k8s-openapi = { version = "0.23.0", features = ["latest"] }
 kube = { version = "0.96.0", features = ["client", "derive", "runtime"] }
 schemars = "0.8.21"
 rocket-errors = "0.1.0"
+rand = "0.8.5"
 
 [profile.release]
 debug = true

crates/rpc/Cargo.toml

@@ -1,10 +1,10 @@
-
 use anyhow::{Context, Result};
 use model::cluster::ClusterState;
 use rocket::{http::Method, routes};
 use rocket_cors::{AllowedOrigins, CorsOptions};
 use routes::{
     add_player::add_player, cleanup::cleanup, head::head, heads::heads, new_game::new_game,
+    sample_transactions::sample_transactions,
 };
 use serde::Deserialize;
 
@@ -41,7 +41,17 @@ async fn main() -> Result<()> {
 
     let _rocket = rocket::build()
         .manage(cluster)
-        .mount("/", routes![new_game, heads, head, add_player, cleanup])
+        .mount(
+            "/",
+            routes![
+                new_game,
+                heads,
+                head,
+                add_player,
+                cleanup,
+                sample_transactions
+            ],
+        )
         .attach(cors.to_cors().unwrap())
         .launch()
         .await?;

crates/rpc/src/main.rs

@@ -13,7 +13,10 @@ use tracing::debug;
 
 use crate::model::{
     game::player::Player,
-    hydra::{hydra_socket, messages::new_tx::NewTx},
+    hydra::{
+        hydra_socket,
+        messages::{new_tx::NewTx, tx_valid::TxValid},
+    },
     tx_builder::TxBuilder,
 };
 
@@ -207,6 +210,16 @@ impl NodeClient {
 
         Ok(utxos)
     }
+
+    pub async fn sample_txs(&self, count: usize) -> Result<Vec<TxValid>> {
+        //TODO: make duration configurable
+        hydra_socket::sample_txs(
+            &format!("{}/?history=no", &self.connection.to_websocket_url()),
+            count,
+            Duration::from_secs(10),
+        )
+        .await
+    }
 }
 
 impl ConnectionInfo {

crates/rpc/src/model/cluster/node.rs

@@ -20,7 +20,7 @@ use futures_util::{
 use tokio::{
     net::TcpStream,
     sync::{mpsc::UnboundedSender, Mutex},
-    task::yield_now,
+    task::{yield_now, JoinHandle},
 };
 use tokio_native_tls::TlsStream;
 use tracing::{debug, info, warn};
@@ -29,7 +29,7 @@ use crate::model::hydra::hydra_message::HydraEventMessage;
 
 use super::{
     hydra_message::{HydraData, HydraMessage},
-    messages::new_tx::NewTx,
+    messages::{new_tx::NewTx, tx_valid::TxValid},
 };
 
 #[allow(dead_code)]
@@ -159,6 +159,49 @@ impl HydraSender {
     }
 }
 
+pub async fn sample_txs(url: &str, count: usize, timeout: Duration) -> Result<Vec<TxValid>> {
+    let request = url.into_client_request().unwrap();
+    info!("attempting to connect to {}", &url);
+    let (ws_stream, _) = connect_async(request).await.context("failed to connect")?;
+    info!("connected to {}", &url);
+
+    let (_, mut receiver) = ws_stream.split();
+    let fetch_transactions: JoinHandle<Result<Vec<TxValid>>> = tokio::spawn(async move {
+        let mut transactions: Vec<TxValid> = Vec::with_capacity(count);
+        loop {
+            let next = receiver.next().await.context("failed to receive")??;
+            let msg = HydraMessage::try_from(next).context("failed to parse hydra message")?;
+
+            match msg {
+                HydraMessage::HydraEvent(x) => match x {
+                    HydraEventMessage::TxValid(tx) => {
+                        transactions.push(tx);
+                        if transactions.len() == count {
+                            break;
+                        }
+                    }
+                    _ => {}
+                },
+                _ => {}
+            }
+        }
+
+        Ok(transactions)
+    });
+
+    tokio::select! {
+        join = fetch_transactions => {
+            match join {
+                Ok(result) => result,
+                Err(e) => Err(e.into()),
+            }
+        }
+        _ = tokio::time::sleep(timeout) => {
+            Err(anyhow!("failed to fetch {} transactions within timeout", count))
+        }
+    }
+}
+
 pub async fn submit_tx_roundtrip(url: &str, tx: NewTx, timeout: Duration) -> Result<()> {
     let request = url.into_client_request().unwrap();
     let (ws_stream, _) = connect_async(request).await.context("failed to connect")?;
@@ -188,7 +231,10 @@ pub async fn submit_tx_roundtrip(url: &str, tx: NewTx, timeout: Duration) -> Res
         }
     });
 
-    sender.send(Message::Text(tx.into())).await.context("failed to send transaction")?;
+    sender
+        .send(Message::Text(tx.into()))
+        .await
+        .context("failed to send transaction")?;
 
     tokio::select! {
         // TODO: result.flatten https://github.com/rust-lang/rust/issues/70142

crates/rpc/src/model/hydra/hydra_socket.rs

@@ -3,3 +3,4 @@ pub mod cleanup;
 pub mod head;
 pub mod heads;
 pub mod new_game;
+pub mod sample_transactions;

crates/rpc/src/routes/mod.rs

@@ -0,0 +1,56 @@
+use rand::seq::SliceRandom;
+
+use crate::model::{
+    cluster::{ClusterState, NodeClient},
+    hydra::messages::tx_valid::TxValid,
+};
+use rand::thread_rng;
+use rocket::{get, http::Status, serde::json::Json, State};
+use rocket_errors::anyhow::Result;
+use serde::Serialize;
+use tracing::error;
+
+#[derive(Serialize)]
+pub struct SampleTransaction {
+    cbor: String,
+    tx_id: String,
+}
+
+#[get("/sample_transactions?<count>&<id>")]
+pub async fn sample_transactions(
+    count: usize,
+    id: Option<&str>,
+    state: &State<ClusterState>,
+) -> Result<Json<Vec<SampleTransaction>>, Status> {
+    let node = match id {
+        Some(id) => state.get_node_by_id(id).ok_or(Status::NotFound)?,
+        None => state
+            .get_all_nodes()
+            .choose(&mut thread_rng())
+            .ok_or(Status::NotFound)?
+            .to_owned(),
+    };
+    let client = NodeClient::new(node, state.admin_sk.clone(), true)
+        .inspect_err(|err| error!("error connecting to node: {}", err))
+        .map_err(|_| Status::InternalServerError)?;
+
+    let transactions = client
+        .sample_txs(count)
+        .await
+        .inspect_err(|err| error!("error sampling transactions: {}", err))
+        .map_err(|_| Status::InternalServerError)?
+        .into_iter()
+        .map(|x| x.into())
+        .collect::<Vec<SampleTransaction>>();
+
+    Ok(Json(transactions))
+}
+
+impl From<TxValid> for SampleTransaction {
+    fn from(value: TxValid) -> Self {
+        Self {
+            cbor: hex::encode(value.cbor),
+            tx_id: value.tx_id,
+        }
+    }
+}