Merge pull request #344 from PeggyJV/bolten/ignore-empty-batches

Only process valid transaction batches

orchestrator/cosmos_gravity/src/query.rs

@@ -77,10 +77,10 @@ pub async fn get_oldest_unsigned_transaction_batch(
             address: address.to_string(),
         })
         .await?;
-    let batches = request.into_inner().batches;
+    let batches = extract_valid_batches(request.into_inner().batches);
     let batch = batches.get(0);
     match batch {
-        Some(batch) => Ok(Some(TransactionBatch::from_proto(batch.clone())?)),
+        Some(batch) => Ok(Some(batch.clone())),
         None => Ok(None),
     }
 }
@@ -93,12 +93,20 @@ pub async fn get_latest_transaction_batches(
     let request = client
         .batch_txs(BatchTxsRequest { pagination: None })
         .await?;
-    let batches = request.into_inner().batches;
-    let mut out = Vec::new();
+    Ok(extract_valid_batches(request.into_inner().batches))
+}
+
+// If we can't serialize a batch from a proto, but it was committed to the chain,
+// we should just ignore it. It should eventually time out and be canceled.
+fn extract_valid_batches(batches: Vec<BatchTx>) -> Vec<TransactionBatch> {
+    let mut valid_batches = Vec::new();
     for batch in batches {
-        out.push(TransactionBatch::from_proto(batch)?)
+        match TransactionBatch::from_proto(batch.clone()) {
+            Ok(valid_batch) => valid_batches.push(valid_batch),
+            Err(e) => warn!("{}, skipping invalid batch: {:?}", e, batch)
+        }
     }
-    Ok(out)
+    valid_batches
 }
 
 /// get all batch confirmations for a given nonce and denom
@@ -185,3 +193,47 @@ pub async fn get_oldest_unsigned_logic_call(
     }
     Ok(out)
 }
+
+#[test]
+fn extract_valid_batches_test() {
+    let erc20_addr = "0x0635FF793Edf48cf5dB294916720A78e6e490E40".to_string();
+    let token_contract = "0xC26eFfa98B8A2632141562Ae7E34953Cfe5B4888".to_string();
+    let transactions = vec![
+        SendToEthereum {
+            id: 1,
+            sender: "cosmos1g0etv93428tvxqftnmj25jn06mz6dtdasj5nz7".to_string(),
+            ethereum_recipient: "0x64D110e00064F2b428476cD64295d8E35836ffd6".to_string(),
+            erc20_token: Some(gravity_proto::gravity::Erc20Token {
+                contract: erc20_addr.clone(),
+                amount: "1".to_string(),
+            }),
+            erc20_fee: Some(gravity_proto::gravity::Erc20Token {
+                contract: erc20_addr.clone(),
+                amount: "1".to_string(),
+            })
+        }
+    ];
+
+    let valid_batch = BatchTx {
+        batch_nonce: 1,
+        timeout: 3,
+        transactions,
+        token_contract: token_contract.clone(),
+        height: 2,
+    };
+
+    let invalid_batch = BatchTx {
+        batch_nonce: 2,
+        timeout: 3,
+        transactions: Vec::new(),
+        token_contract: token_contract.clone(),
+        height: 2,
+    };
+
+    let valid_batches = extract_valid_batches(vec![valid_batch.clone(), invalid_batch.clone()]);
+    assert_eq!(valid_batches.len(), 1);
+    assert_eq!(valid_batches.get(0).unwrap().nonce, 1);
+
+    let should_be_empty = extract_valid_batches(vec![invalid_batch.clone()]);
+    assert_eq!(should_be_empty.len(), 0);
+}