feat(network): implement GetUTxOByTxIn state query (#550)

examples/n2c-miniprotocols/src/main.rs

@@ -7,18 +7,33 @@ use pallas::{
         facades::NodeClient,
         miniprotocols::{
             chainsync,
-            localstate::queries_v16::{self, Addr, Addrs},
+            localstate::queries_v16::{
+                self, Addr, Addrs, TransactionInput,
+            },
             Point, PRE_PRODUCTION_MAGIC,
         },
     },
+    crypto::hash::Hash,
 };
 use tracing::info;
+use hex::FromHex;
 
 async fn do_localstate_query(client: &mut NodeClient) {
     let client = client.statequery();
 
     client.acquire(None).await.unwrap();
 
+    // Get UTxO from a (singleton) set of tx inputs.
+    let transaction_id = Hash::<32>::from(<[u8; 32]>::from_hex(
+        "15244950ed56a3af61a00f62584779fb53a9f3910468013a2b00b94b8bbc10e0"
+    ).unwrap());
+    let tx_in = TransactionInput { transaction_id, index: 0 };
+    let mut txins = BTreeSet::new();
+    txins.insert(tx_in);
+    
+    let result = queries_v16::get_utxo_by_txin(client, 6, txins).await.unwrap();
+    info!("result: {:?}", result);
+
     let result = queries_v16::get_chain_point(client).await.unwrap();
     info!("result: {:?}", result);
 

pallas-network/src/miniprotocols/localstate/queries_v16/codec.rs

@@ -71,10 +71,10 @@ impl Encode<()> for BlockQuery {
                 e.array(1)?;
                 e.u16(14)?;
             }
-            BlockQuery::GetUTxOByTxIn(_) => {
+            BlockQuery::GetUTxOByTxIn(txin) => {
                 e.array(2)?;
                 e.u16(15)?;
-                e.encode(2)?;
+                e.encode(txin)?;
             }
             BlockQuery::GetStakePools => {
                 e.array(1)?;
@@ -144,7 +144,7 @@ impl<'b> Decode<'b, ()> for BlockQuery {
             // 12 => Ok(Self::DebugNewEpochState),
             13 => Ok(Self::DebugChainDepState),
             14 => Ok(Self::GetRewardProvenance),
-            // 15 => Ok(Self::GetUTxOByTxIn(())),
+            15 => Ok(Self::GetUTxOByTxIn(d.decode()?)),
             16 => Ok(Self::GetStakePools),
             // 17 => Ok(Self::GetStakePoolParams(())),
             18 => Ok(Self::GetRewardInfoPools),

pallas-network/src/miniprotocols/localstate/queries_v16/mod.rs

@@ -37,7 +37,7 @@ pub enum BlockQuery {
     DebugNewEpochState,
     DebugChainDepState,
     GetRewardProvenance,
-    GetUTxOByTxIn(AnyCbor),
+    GetUTxOByTxIn(TxIns),
     GetStakePools,
     GetStakePoolParams(AnyCbor),
     GetRewardInfoPools,
@@ -236,9 +236,25 @@ pub struct UTxOByAddress {
     pub utxo: KeyValuePairs<UTxO, TransactionOutput>,
 }
 
+pub type UTxOByTxin = UTxOByAddress;
+
 // Bytes CDDL ->  #6.121([ * #6.121([ *datum ]) ])
 pub type Datum = (Era, TagWrap<Bytes, 24>);
 
+// From `pallas-primitives`, with fewer `derive`s
+#[derive(
+    Encode, Decode, Debug, PartialEq, Eq, PartialOrd, Ord, Clone,
+)]
+pub struct TransactionInput {
+    #[n(0)]
+    pub transaction_id: Hash<32>,
+
+    #[n(1)]
+    pub index: u64,
+}
+
+pub type TxIns = BTreeSet<TransactionInput>;
+
 #[derive(Debug, PartialEq, Eq, Clone)]
 pub enum TransactionOutput {
     Current(PostAlonsoTransactionOutput),
@@ -477,3 +493,17 @@ pub async fn get_genesis_config(
 
     Ok(result)
 }
+
+/// Get a subset of the UTxO, filtered by transaction input.
+pub async fn get_utxo_by_txin(
+    client: &mut Client,
+    era: u16,
+    txins: TxIns,
+) -> Result<UTxOByTxin, ClientError> {
+    let query = BlockQuery::GetUTxOByTxIn(txins);
+    let query = LedgerQuery::BlockQuery(era, query);
+    let query = Request::LedgerQuery(query);
+    let result = client.query(query).await?;
+
+    Ok(result)
+}

pallas-network/tests/protocols.rs

@@ -260,7 +260,7 @@ pub async fn blockfetch_server_and_client_happy_path() {
         client_bf.send_done().await.unwrap();
     });
 
-    _ = tokio::join!(client, server);
+    tokio::try_join!(client, server).unwrap();
 }
 
 #[tokio::test]
@@ -457,7 +457,7 @@ pub async fn chainsync_server_and_client_happy_path_n2n() {
         client_cs.send_done().await.unwrap();
     });
 
-    _ = tokio::join!(client, server);
+    tokio::try_join!(client, server).unwrap();
 }
 
 #[cfg(unix)]
@@ -1092,7 +1092,7 @@ pub async fn local_state_query_server_and_client_happy_path() {
         client.statequery().send_done().await.unwrap();
     });
 
-    _ = tokio::join!(client, server);
+    tokio::try_join!(client, server).unwrap();
 }
 
 #[tokio::test]
@@ -1250,7 +1250,7 @@ pub async fn txsubmission_server_and_client_happy_path_n2n() {
         assert_eq!(*client_txsub.state(), txsubmission::State::Done);
     });
 
-    _ = tokio::join!(client, server);
+    tokio::try_join!(client, server).unwrap();
 }
 
 #[tokio::test]