Helium Lib attempt: 2

Cargo.toml

@@ -135,3 +135,6 @@ sqlx = { git = "https://github.com/launchbadge/sqlx.git", rev = "42dd78fe931df65
 # [patch.'https://github.com/helium/proto']
 # helium-proto = { path = "../proto" }
 # beacon = { path = "../proto/beacon" }
+
+# [patch.'https://github.com/helium/helium-wallet-rs.git']
+# helium-lib = { path = "../helium-wallet-rs/helium-lib" }

boost_manager/tests/integrations/updater_tests.rs

@@ -20,7 +20,7 @@ pub struct MockTransaction {
 
 pub struct MockSolanaConnection {
     submitted: Mutex<Vec<MockTransaction>>,
-    error: Option<String>,
+    error: Option<tonic::Status>,
 }
 
 impl MockSolanaConnection {
@@ -34,7 +34,7 @@ impl MockSolanaConnection {
     fn with_error(error: String) -> Self {
         Self {
             submitted: Mutex::new(vec![]),
-            error: Some(error),
+            error: Some(tonic::Status::internal(error)),
         }
     }
 }
@@ -58,7 +58,11 @@ impl SolanaNetwork for MockSolanaConnection {
 
         self.error
             .as_ref()
-            .map(|err| Err(SolanaRpcError::Test(err.to_owned())))
+            .map(|err| {
+                Err(SolanaRpcError::HeliumLib(solana::error::Error::Grpc(
+                    err.to_owned(),
+                )))
+            })
             .unwrap_or(Ok(()))
     }
 

iot_packet_verifier/src/balances.rs

@@ -12,6 +12,7 @@ use tokio::sync::Mutex;
 
 /// Caches balances fetched from the solana chain and debits made by the
 /// packet verifier.
+#[derive(Clone)]
 pub struct BalanceCache<S> {
     payer_accounts: BalanceStore,
     solana: S,
@@ -55,6 +56,10 @@ impl<S> BalanceCache<S> {
     pub fn balances(&self) -> BalanceStore {
         self.payer_accounts.clone()
     }
+
+    pub async fn get_payer_balance(&self, payer: &PublicKeyBinary) -> Option<PayerAccount> {
+        self.payer_accounts.lock().await.get(payer).cloned()
+    }
 }
 
 #[async_trait::async_trait]

iot_packet_verifier/src/burner.rs

@@ -5,10 +5,12 @@ use crate::{
     },
 };
 use futures::{future::LocalBoxFuture, TryFutureExt};
-use solana::{burn::SolanaNetwork, GetSignature, SolanaRpcError};
+use helium_crypto::PublicKeyBinary;
+use solana::{burn::SolanaNetwork, sender, SolanaRpcError};
 use std::time::Duration;
 use task_manager::ManagedTask;
 use tokio::time::{self, MissedTickBehavior};
+use tracing::Instrument;
 
 pub struct Burner<P, S> {
     pending_tables: P,
@@ -19,7 +21,7 @@ pub struct Burner<P, S> {
 
 impl<P, S> ManagedTask for Burner<P, S>
 where
-    P: PendingTables + Send + Sync + 'static,
+    P: PendingTables,
     S: SolanaNetwork,
 {
     fn start_task(
@@ -76,9 +78,9 @@ where
 
         loop {
             tokio::select! {
-                    biased;
-                    _ = shutdown.clone() => break,
-                    _ = burn_timer.tick() => {
+                biased;
+                _ = shutdown.clone() => break,
+                _ = burn_timer.tick() => {
                     match self.burn().await {
                         Ok(()) => continue,
                         Err(err) => {
@@ -94,49 +96,147 @@ where
     }
 
     pub async fn burn(&mut self) -> Result<(), BurnError> {
+        // There should only be a single pending txn at a time
+        let pending_txns = self.pending_tables.fetch_all_pending_txns().await?;
+        if !pending_txns.is_empty() {
+            tracing::info!(pending_txns = pending_txns.len(), "skipping burn");
+            return Ok(());
+        }
+
         // Fetch the next payer and amount that should be burn. If no such burn
         // exists, perform no action.
         let Some(Burn { payer, amount }) = self.pending_tables.fetch_next_burn().await? else {
+            tracing::info!("no pending burns");
             return Ok(());
         };
 
         tracing::info!(%amount, %payer, "Burning DC");
 
-        // Create a burn transaction and execute it:
         let txn = self
             .solana
             .make_burn_transaction(&payer, amount)
             .await
             .map_err(BurnError::SolanaError)?;
-        self.pending_tables
-            .add_pending_transaction(&payer, amount, txn.get_signature())
-            .await?;
+
+        let store = BurnTxnStore::new(
+            self.pending_tables.clone(),
+            self.balances.clone(),
+            payer.clone(),
+            amount,
+        );
+
+        let burn_span = tracing::info_span!("burn_txn", %payer, amount);
         self.solana
-            .submit_transaction(&txn)
+            .submit_transaction(&txn, &store)
+            .map_err(BurnError::SolanaError)
+            .instrument(burn_span)
             .await
-            .map_err(BurnError::SolanaError)?;
+    }
+}
+
+pub struct BurnTxnStore<PT> {
+    pool: PT,
+    balances: BalanceStore,
+    payer: PublicKeyBinary,
+    amount: u64,
+}
 
-        // Removing the pending transaction and subtract the burn amount
-        // now that we have confirmation that the burn transaction is confirmed
-        // on chain:
-        let mut pending_tables_txn = self.pending_tables.begin().await?;
-        pending_tables_txn
-            .remove_pending_transaction(txn.get_signature())
-            .await?;
-        pending_tables_txn
-            .subtract_burned_amount(&payer, amount)
-            .await?;
-        pending_tables_txn.commit().await?;
+impl<PT: PendingTables + Clone> BurnTxnStore<PT> {
+    pub fn new(pool: PT, balances: BalanceStore, payer: PublicKeyBinary, amount: u64) -> Self {
+        Self {
+            pool,
+            balances,
+            payer,
+            amount,
+        }
+    }
+}
 
-        let mut balance_lock = self.balances.lock().await;
-        let payer_account = balance_lock.get_mut(&payer).unwrap();
-        // Reduce the pending burn amount and the payer's balance by the amount
-        // we've burned.
-        payer_account.burned = payer_account.burned.saturating_sub(amount);
-        payer_account.balance = payer_account.balance.saturating_sub(amount);
+#[async_trait::async_trait]
+impl<PT: PendingTables> sender::TxnStore for BurnTxnStore<PT> {
+    async fn on_prepared(&self, txn: &solana::Transaction) -> sender::SenderResult<()> {
+        tracing::info!("txn prepared");
 
-        metrics::counter!("burned", "payer" => payer.to_string()).increment(amount);
+        let signature = txn.get_signature();
+        let add_pending = self
+            .pool
+            .add_pending_transaction(&self.payer, self.amount, signature);
+
+        let Ok(()) = add_pending.await else {
+            tracing::error!("failed to add pending transcation");
+            return Err(sender::SenderError::preparation(
+                "could not add pending transaction",
+            ));
+        };
 
         Ok(())
     }
+
+    async fn on_finalized(&self, txn: &solana::Transaction) {
+        tracing::info!("txn finalized");
+
+        let Ok(mut db_txn) = self.pool.begin().await else {
+            tracing::error!("failed to start finalized txn db transaction");
+            return;
+        };
+
+        let signature = txn.get_signature();
+        let Ok(()) = db_txn.remove_pending_transaction(signature).await else {
+            tracing::error!("failed to remove pending");
+            return;
+        };
+
+        let Ok(()) = db_txn
+            .subtract_burned_amount(&self.payer, self.amount)
+            .await
+        else {
+            tracing::error!("failed to subtract burned amount");
+            return;
+        };
+
+        // Subtract balances from map before submitted db txn
+        let mut balance_lock = self.balances.lock().await;
+        let payer_account = balance_lock.get_mut(&self.payer).unwrap();
+        // Reduce the pending burn amount and the payer's balance by the amount we've burned
+        payer_account.burned = payer_account.burned.saturating_sub(self.amount);
+        payer_account.balance = payer_account.balance.saturating_sub(self.amount);
+
+        let Ok(()) = db_txn.commit().await else {
+            tracing::error!("failed to commit finalized txn db transaction");
+            return;
+        };
+
+        metrics::counter!(
+            "burned",
+            "payer" => self.payer.to_string(),
+            "success" => "true"
+        )
+        .increment(self.amount);
+    }
+
+    async fn on_error(&self, txn: &solana::Transaction, err: sender::SenderError) {
+        tracing::warn!(?err, "txn failed");
+        let Ok(mut db_txn) = self.pool.begin().await else {
+            tracing::error!("failed to start error transaction");
+            return;
+        };
+
+        let signature = txn.get_signature();
+        let Ok(()) = db_txn.remove_pending_transaction(signature).await else {
+            tracing::error!("failed to remove pending transaction on error");
+            return;
+        };
+
+        let Ok(()) = db_txn.commit().await else {
+            tracing::error!("failed to commit on error transaction");
+            return;
+        };
+
+        metrics::counter!(
+            "burned",
+            "payer" => self.payer.to_string(),
+            "success" => "false"
+        )
+        .increment(self.amount);
+    }
 }

iot_packet_verifier/src/daemon.rs

@@ -112,7 +112,7 @@ impl Cmd {
                 bail!("Missing solana section in settings");
             };
             // Set up the solana RpcClient:
-            Some(SolanaRpc::new(solana_settings).await?)
+            Some(SolanaRpc::new(solana_settings, solana::SubDao::Iot).await?)
         } else {
             None
         };