FullTransaction: add state

Author: tharvik

src/at2.proto

@@ -31,14 +31,21 @@ message GetLastSequenceReply {
 	uint32 sequence = 1;
 }
 
-message ProcessedTransaction {
+message FullTransaction {
 	string timestamp = 1;
 	bytes sender = 2;
 	bytes recipient = 3;
 	uint64 amount = 4;
+	enum State {
+		Pending = 0;
+		Success = 1;
+		Failure = 2;
+	}
+	State state = 5;
+	uint32 sender_sequence = 6;
 }
 
 message GetLatestTransactionsRequest {}
 message GetLatestTransactionsReply {
-	repeated ProcessedTransaction transactions = 1;
+	repeated FullTransaction transactions = 1;
 }

src/bin/client/main.rs

@@ -1,6 +1,9 @@
 use std::io::{stdin, stdout};
 
-use at2_node::client::{self, Client};
+use at2_node::{
+    client::{self, Client},
+    TransactionState,
+};
 use drop::crypto::sign;
 use hex::FromHex;
 use http::Uri;
@@ -134,8 +137,16 @@ async fn get_latest_transactions() -> Result<(), CommandError> {
         .iter()
         .for_each(|tx| {
             println!(
-                "{}: {} send {}¤ to {}",
-                tx.timestamp, tx.sender, tx.amount, tx.recipient,
+                "{}: {} send {}¤ to {} ({})",
+                tx.timestamp,
+                tx.sender,
+                tx.amount,
+                tx.recipient,
+                match tx.state {
+                    TransactionState::Pending => "pending",
+                    TransactionState::Success => "success",
+                    TransactionState::Failure => "failure",
+                },
             )
         });
 

src/bin/server/recent_transactions.rs

@@ -1,7 +1,8 @@
 use std::collections::VecDeque;
 
-use at2_node::{FullTransaction, ThinTransaction};
+use at2_node::{FullTransaction, ThinTransaction, TransactionState};
 use drop::crypto::sign;
+use snafu::ensure;
 use tokio::sync::{mpsc, oneshot};
 
 const LATEST_TRANSACTIONS_MAX_SIZE: usize = 10;
@@ -12,13 +13,23 @@ pub enum Error {
     GoneOnSend,
     #[snafu(display("gone on recv"))]
     GoneOnRecv,
+    PutAlreadyExisting,
 }
 
+type Result<T> = std::result::Result<T, Error>;
+
 #[derive(Debug)]
 enum Commands {
     Put {
+        sender: Box<sign::PublicKey>,
+        sender_sequence: sieve::Sequence,
         thin: ThinTransaction,
+        resp: oneshot::Sender<Result<()>>,
+    },
+    Update {
         sender: Box<sign::PublicKey>,
+        sender_sequence: sieve::Sequence,
+        state: TransactionState,
         resp: oneshot::Sender<()>,
     },
     GetAll {
@@ -46,24 +57,48 @@ impl RecentTransactions {
     pub async fn put(
         &self,
         sender: Box<sign::PublicKey>,
+        sender_sequence: sieve::Sequence,
         thin: ThinTransaction,
-    ) -> Result<(), Error> {
+    ) -> Result<()> {
         let (tx, rx) = oneshot::channel();
 
         self.agent
             .send(Commands::Put {
                 sender,
+                sender_sequence,
                 thin,
                 resp: tx,
             })
             .await
             .map_err(|_| Error::GoneOnSend)?;
 
+        rx.await.map_err(|_| Error::GoneOnRecv)?
+    }
+
+    /// Update an already put transaction, to resolve its state
+    pub async fn update(
+        &self,
+        sender: Box<sign::PublicKey>,
+        sender_sequence: sieve::Sequence,
+        state: TransactionState,
+    ) -> Result<()> {
+        let (tx, rx) = oneshot::channel();
+
+        self.agent
+            .send(Commands::Update {
+                sender,
+                sender_sequence,
+                state,
+                resp: tx,
+            })
+            .await
+            .map_err(|_| Error::GoneOnSend)?;
+
         rx.await.map_err(|_| Error::GoneOnRecv)
     }
 
     /// Return the recently seen transactions
-    pub async fn get_all(&self) -> Result<Vec<FullTransaction>, Error> {
+    pub async fn get_all(&self) -> Result<Vec<FullTransaction>> {
         let (tx, rx) = oneshot::channel();
 
         self.agent
@@ -86,8 +121,21 @@ impl RecentTransactionsHandler {
         tokio::spawn(async move {
             while let Some(cmd) = rx.recv().await {
                 match cmd {
-                    Commands::Put { sender, thin, resp } => {
-                        self.put(*sender, thin);
+                    Commands::Put {
+                        sender,
+                        sender_sequence,
+                        thin,
+                        resp,
+                    } => {
+                        let _ = resp.send(self.put(*sender, sender_sequence, thin));
+                    }
+                    Commands::Update {
+                        sender,
+                        sender_sequence,
+                        state,
+                        resp,
+                    } => {
+                        self.update(*sender, sender_sequence, state);
                         let _ = resp.send(());
                     }
                     Commands::GetAll { resp } => {
@@ -100,12 +148,52 @@ impl RecentTransactionsHandler {
         tx
     }
 
-    fn put(&mut self, sender: sign::PublicKey, thin: ThinTransaction) {
+    fn put(
+        &mut self,
+        sender: sign::PublicKey,
+        sender_sequence: sieve::Sequence,
+        thin: ThinTransaction,
+    ) -> Result<()> {
+        ensure!(
+            !self
+                .0
+                .iter()
+                .any(|tx| tx.sender_sequence == sender_sequence && tx.sender == sender),
+            PutAlreadyExisting
+        );
+
+        let full = FullTransaction {
+            timestamp: chrono::Utc::now(),
+            sender,
+            sender_sequence,
+            recipient: thin.recipient,
+            amount: thin.amount,
+            state: TransactionState::Pending,
+        };
+
         if self.0.len() == LATEST_TRANSACTIONS_MAX_SIZE {
             self.0.pop_front();
         }
-        let full = FullTransaction::with_thin(sender, thin);
+
         self.0.push_back(full);
+
+        Ok(())
+    }
+
+    fn update(
+        &mut self,
+        sender: sign::PublicKey,
+        sender_sequence: sieve::Sequence,
+        state: TransactionState,
+    ) {
+        // NOP if not found as the transaction may resolve late
+        if let Some(tx) = self
+            .0
+            .iter_mut()
+            .find(|tx| tx.sender_sequence == sender_sequence && tx.sender == sender)
+        {
+            tx.state = state;
+        }
     }
 
     fn get_all(&self) -> Vec<FullTransaction> {

src/bin/server/rpc.rs

@@ -1,6 +1,9 @@
 use std::fmt;
 
-use at2_node::{proto, ThinTransaction};
+use at2_node::{
+    proto::{self, *},
+    ThinTransaction, TransactionState,
+};
 use contagion::{Contagion, ContagionConfig, ContagionMessage};
 use drop::{
     crypto::key::exchange::{self, Exchanger},
@@ -168,22 +171,31 @@ impl Service {
 
         let sender = Box::new(msg.sender().to_owned());
 
-        self.accounts
+        let processed = self
+            .accounts
             .transfer(
                 sender.clone(),
                 msg.sequence(),
                 Box::new(msg.payload().recipient),
                 msg.payload().amount,
             )
             .await
-            .context(ProcessTxForAccounts)?;
+            .context(ProcessTxForAccounts);
 
         self.recent_transactions
-            .put(sender, msg.payload().to_owned())
+            .update(
+                sender,
+                msg.sequence(),
+                if processed.is_ok() {
+                    TransactionState::Success
+                } else {
+                    TransactionState::Failure
+                },
+            )
             .await
             .context(ProcessTxForRecent)?;
 
-        Ok(())
+        processed
     }
 }
 
@@ -204,49 +216,57 @@ impl From<recent_transactions::Error> for tonic::Status {
 }
 
 #[tonic::async_trait]
-impl proto::at2_server::At2 for Service {
+impl at2_server::At2 for Service {
     async fn send_asset(
         &self,
-        request: tonic::Request<proto::SendAssetRequest>,
-    ) -> Result<tonic::Response<proto::SendAssetReply>, tonic::Status> {
+        request: tonic::Request<SendAssetRequest>,
+    ) -> Result<tonic::Response<SendAssetReply>, tonic::Status> {
         let message = request.into_inner();
 
+        let thin = at2_node::ThinTransaction {
+            recipient: bincode::deserialize(&message.recipient).context(Deserialize)?,
+            amount: message.amount,
+        };
+
+        let sender = bincode::deserialize(&message.sender).context(Deserialize)?;
+
+        self.recent_transactions
+            .put(Box::new(sender), message.sequence, thin.clone())
+            .await?;
+
         self.handle
             .clone()
             .broadcast(&sieve::Payload::new(
-                bincode::deserialize(&message.sender).context(Deserialize)?,
+                sender,
                 message.sequence,
-                at2_node::ThinTransaction {
-                    recipient: bincode::deserialize(&message.recipient).context(Deserialize)?,
-                    amount: message.amount,
-                },
+                thin,
                 bincode::deserialize(&message.signature).context(Deserialize)?,
             ))
             .await
-            .expect("broadcasting failed");
+            .map_err(|err| tonic::Status::invalid_argument(err.to_string()))?;
 
-        Ok(Response::new(proto::SendAssetReply {}))
+        Ok(Response::new(SendAssetReply {}))
     }
 
     async fn get_last_sequence(
         &self,
-        request: tonic::Request<proto::GetLastSequenceRequest>,
-    ) -> Result<tonic::Response<proto::GetLastSequenceReply>, tonic::Status> {
+        request: tonic::Request<GetLastSequenceRequest>,
+    ) -> Result<tonic::Response<GetLastSequenceReply>, tonic::Status> {
         let sequence = self
             .accounts
             .get_last_sequence(
                 bincode::deserialize(&request.get_ref().sender).context(Deserialize)?,
             )
             .await?;
 
-        Ok(Response::new(proto::GetLastSequenceReply { sequence }))
+        Ok(Response::new(GetLastSequenceReply { sequence }))
     }
 
     async fn get_balance(
         &self,
-        request: tonic::Request<proto::GetBalanceRequest>,
-    ) -> Result<tonic::Response<proto::GetBalanceReply>, tonic::Status> {
-        Ok(Response::new(proto::GetBalanceReply {
+        request: tonic::Request<GetBalanceRequest>,
+    ) -> Result<tonic::Response<GetBalanceReply>, tonic::Status> {
+        Ok(Response::new(GetBalanceReply {
             amount: self
                 .accounts
                 .get_balance(bincode::deserialize(&request.get_ref().sender).context(Deserialize)?)
@@ -256,20 +276,28 @@ impl proto::at2_server::At2 for Service {
 
     async fn get_latest_transactions(
         &self,
-        _: tonic::Request<proto::GetLatestTransactionsRequest>,
-    ) -> Result<tonic::Response<proto::GetLatestTransactionsReply>, tonic::Status> {
-        Ok(Response::new(proto::GetLatestTransactionsReply {
+        _: tonic::Request<GetLatestTransactionsRequest>,
+    ) -> Result<tonic::Response<GetLatestTransactionsReply>, tonic::Status> {
+        use full_transaction::State;
+
+        Ok(Response::new(GetLatestTransactionsReply {
             transactions: self
                 .recent_transactions
                 .get_all()
                 .await?
                 .iter()
                 .map(|tx| {
-                    Ok(proto::ProcessedTransaction {
+                    Ok(proto::FullTransaction {
                         timestamp: tx.timestamp.to_rfc3339(),
                         sender: bincode::serialize(&tx.sender).context(Serialize)?,
+                        sender_sequence: tx.sender_sequence,
                         recipient: bincode::serialize(&tx.recipient).context(Serialize)?,
                         amount: tx.amount,
+                        state: match tx.state {
+                            TransactionState::Pending => State::Pending as i32,
+                            TransactionState::Success => State::Success as i32,
+                            TransactionState::Failure => State::Failure as i32,
+                        },
                     })
                 })
                 .collect::<Result<_, ProtoError>>()?,