Integrate the trie

kairos-server/Cargo.toml

@@ -27,6 +27,8 @@ tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["std", "env-filter"] }
 hex = "0.4"
 kairos-tx = { path = "../kairos-tx" }
+kairos-trie = { git = "https://github.com/cspr-rad/kairos-trie" }
+sha2 = "0.10"
 
 [dev-dependencies]
 proptest = "1"

kairos-server/src/errors.rs

@@ -66,3 +66,12 @@ impl From<anyhow::Error> for AppErr {
         }
     }
 }
+
+impl From<kairos_trie::TrieError> for AppErr {
+    fn from(error: kairos_trie::TrieError) -> Self {
+        Self {
+            error: anyhow::Error::msg(error.to_string()),
+            status: Some(StatusCode::INTERNAL_SERVER_ERROR),
+        }
+    }
+}

kairos-server/src/lib.rs

@@ -5,16 +5,17 @@ pub mod state;
 
 mod utils;
 
+use std::sync::Arc;
+
 use axum::Router;
 use axum_extra::routing::RouterExt;
-use state::LockedBatchState;
 
 pub use errors::AppErr;
 
 type PublicKey = Vec<u8>;
 type Signature = Vec<u8>;
 
-pub fn app_router(state: LockedBatchState) -> Router {
+pub fn app_router(state: Arc<state::BatchStateManager>) -> Router {
     Router::new()
         .typed_post(routes::deposit_handler)
         .typed_post(routes::withdraw_handler)

kairos-server/src/main.rs

@@ -1,7 +1,7 @@
 use std::net::SocketAddr;
 
 use dotenvy::dotenv;
-use kairos_server::{config::ServerConfig, state::BatchState};
+use kairos_server::{config::ServerConfig, state::BatchStateManager};
 
 #[tokio::main]
 async fn main() {
@@ -11,7 +11,7 @@ async fn main() {
     let config = ServerConfig::from_env()
         .unwrap_or_else(|e| panic!("Failed to parse server config from environment: {}", e));
 
-    let app = kairos_server::app_router(BatchState::new());
+    let app = kairos_server::app_router(BatchStateManager::new_empty());
 
     let axum_addr = SocketAddr::from(([127, 0, 0, 1], config.port));
 

kairos-server/src/routes/deposit.rs

@@ -1,4 +1,4 @@
-use std::ops::Deref;
+use std::sync::Arc;
 
 use anyhow::{anyhow, Context};
 use axum::{extract::State, http::StatusCode, Json};
@@ -7,8 +7,14 @@ use tracing::*;
 
 use kairos_tx::asn::{SigningPayload, TransactionBody};
 
-use crate::routes::PayloadBody;
-use crate::{state::LockedBatchState, AppErr};
+use crate::{
+    routes::PayloadBody,
+    state::{
+        transactions::{Signed, Transaction},
+        BatchStateManager,
+    },
+    AppErr,
+};
 
 #[derive(TypedPath, Debug, Clone, Copy)]
 #[typed_path("/api/v1/deposit")]
@@ -17,46 +23,30 @@ pub struct DepositPath;
 #[instrument(level = "trace", skip(state), ret)]
 pub async fn deposit_handler(
     _: DepositPath,
-    state: State<LockedBatchState>,
+    state: State<Arc<BatchStateManager>>,
     Json(body): Json<PayloadBody>,
 ) -> Result<(), AppErr> {
     tracing::info!("parsing transaction data");
     let signing_payload: SigningPayload =
         body.payload.as_slice().try_into().context("payload err")?;
     let deposit = match signing_payload.body {
-        TransactionBody::Deposit(deposit) => deposit,
+        TransactionBody::Deposit(deposit) => deposit.try_into().context("decoding deposit")?,
         _ => {
             return Err(AppErr::set_status(
                 anyhow!("invalid transaction type"),
                 StatusCode::BAD_REQUEST,
             ))
         }
     };
-    let amount = u64::try_from(deposit.amount).context("invalid amount")?;
-    let public_key = body.public_key;
-
-    tracing::info!("TODO: verifying deposit");
-
-    tracing::info!("TODO: adding deposit to batch");
-
-    let mut state = state.deref().write().await;
-    let account = state.balances.entry(public_key.clone());
 
-    let balance = account.or_insert(0);
-    let updated_balance = balance.checked_add(amount).ok_or_else(|| {
-        AppErr::set_status(
-            anyhow!("deposit would overflow account"),
-            StatusCode::CONFLICT,
-        )
-    })?;
-
-    *balance = updated_balance;
-
-    tracing::info!(
-        "Updated account public_key={:?} balance={}",
-        public_key,
-        updated_balance
-    );
-
-    Ok(())
+    let public_key = body.public_key;
+    let nonce = signing_payload.nonce.try_into().context("decoding nonce")?;
+
+    state
+        .enqueue_transaction(Signed {
+            public_key,
+            nonce,
+            transaction: Transaction::Deposit(deposit),
+        })
+        .await
 }

kairos-server/src/routes/transfer.rs

@@ -1,12 +1,20 @@
+use std::sync::Arc;
+
 use anyhow::{anyhow, Context};
 use axum::{extract::State, http::StatusCode, Json};
 use axum_extra::routing::TypedPath;
 use tracing::instrument;
 
 use kairos_tx::asn::{SigningPayload, TransactionBody};
 
-use crate::routes::PayloadBody;
-use crate::{state::LockedBatchState, AppErr, PublicKey};
+use crate::{
+    routes::PayloadBody,
+    state::{
+        transactions::{Signed, Transaction, Transfer},
+        BatchStateManager,
+    },
+    AppErr,
+};
 
 #[derive(TypedPath)]
 #[typed_path("/api/v1/transfer")]
@@ -15,106 +23,33 @@ pub struct TransferPath;
 #[instrument(level = "trace", skip(state), ret)]
 pub async fn transfer_handler(
     _: TransferPath,
-    State(state): State<LockedBatchState>,
+    State(state): State<Arc<BatchStateManager>>,
     Json(body): Json<PayloadBody>,
 ) -> Result<(), AppErr> {
     tracing::info!("parsing transaction data");
     let signing_payload: SigningPayload =
         body.payload.as_slice().try_into().context("payload err")?;
-    let transfer = match signing_payload.body {
-        TransactionBody::Transfer(transfer) => transfer,
+    let transfer: Transfer = match signing_payload.body {
+        TransactionBody::Transfer(transfer) => transfer.try_into().context("decoding transfer")?,
         _ => {
             return Err(AppErr::set_status(
                 anyhow!("invalid transaction type"),
                 StatusCode::BAD_REQUEST,
             ))
         }
     };
-    let amount = u64::try_from(transfer.amount).context("invalid amount")?;
-    let from = body.public_key;
-    let to = PublicKey::from(transfer.recipient);
-
-    if amount == 0 {
-        return Err(AppErr::set_status(
-            anyhow!("transfer amount must be greater than 0"),
-            StatusCode::BAD_REQUEST,
-        ));
-    }
+    let public_key = body.public_key;
+    let nonce = signing_payload.nonce.try_into().context("decoding nonce")?;
 
     tracing::info!("TODO: verifying transfer signature");
 
-    // We pre-check this read-only to error early without acquiring the write lock.
-    // This prevents a DoS attack exploiting the write lock.
-    tracing::info!("verifying transfer sender has sufficient funds");
-    check_sender_funds(&state, &from, amount, &to).await?;
-
-    let mut state = state.write().await;
-    let from_balance = state.balances.get_mut(&from).ok_or_else(|| {
-        AppErr::set_status(
-            anyhow!(
-                "Sender no longer has an account.
-                The sender just removed all their funds."
-            ),
-            StatusCode::CONFLICT,
-        )
-    })?;
-
-    *from_balance = from_balance.checked_sub(amount).ok_or_else(|| {
-        AppErr::set_status(
-            anyhow!(
-                "Sender no longer has sufficient funds, balance={}, transfer_amount={}.
-                The sender just moved their funds in a concurrent request",
-                from_balance,
-                amount
-            ),
-            StatusCode::CONFLICT,
-        )
-    })?;
-
-    let to_balance = state.balances.entry(to.clone()).or_insert_with(|| {
-        tracing::info!("creating new account for receiver");
-        0
-    });
-
-    *to_balance = to_balance.checked_add(amount).ok_or_else(|| {
-        AppErr::set_status(anyhow!("Receiver balance overflow"), StatusCode::CONFLICT)
-    })?;
-
-    Ok(())
-}
-
-async fn check_sender_funds(
-    state: &LockedBatchState,
-    from: &PublicKey,
-    amount: u64,
-    to: &PublicKey,
-) -> Result<(), AppErr> {
-    let state = state.read().await;
-    let from_balance = state.balances.get(from).ok_or_else(|| {
-        AppErr::set_status(
-            anyhow!("Sender does not have an account"),
-            StatusCode::BAD_REQUEST,
-        )
-    })?;
-
-    from_balance.checked_sub(amount).ok_or_else(|| {
-        AppErr::set_status(
-            anyhow!(
-                "Sender does not have sufficient funds, balance={}, transfer_amount={}",
-                from_balance,
-                amount
-            ),
-            StatusCode::FORBIDDEN,
-        )
-    })?;
-
-    let to_balance = state.balances.get(to).unwrap_or(&0);
-    if to_balance.checked_add(amount).is_none() {
-        return Err(AppErr::set_status(
-            anyhow!("Receiver balance overflow"),
-            StatusCode::CONFLICT,
-        ));
-    }
+    tracing::info!("queuing transaction for trie update");
 
-    Ok(())
+    state
+        .enqueue_transaction(Signed {
+            public_key,
+            nonce,
+            transaction: Transaction::Transfer(transfer),
+        })
+        .await
 }