feat(network): implement GetUTxOByAddress local state query (#341)

examples/n2c-miniprotocols/src/main.rs

@@ -1,6 +1,13 @@
-use pallas::network::{
+use pallas::{
-    facades::NodeClient,
+    ledger::addresses::Address,
-    miniprotocols::{chainsync, localstate::queries_v16, Point, PRE_PRODUCTION_MAGIC},
+    network::{
+        facades::NodeClient,
+        miniprotocols::{
+            chainsync,
+            localstate::queries_v16::{self, Addr, Addrs},
+            Point, PRE_PRODUCTION_MAGIC,
+        },
+    },
 };
 use tracing::info;
 
@@ -28,6 +35,20 @@ async fn do_localstate_query(client: &mut NodeClient) {
         .unwrap();
     info!("result: {:?}", result);
 
+    let addrx = "addr_test1vr80076l3x5uw6n94nwhgmv7ssgy6muzf47ugn6z0l92rhg2mgtu0".to_string();
+    let addrx: Address = Address::from_bech32(&addrx).unwrap();
+    let addrx: Addr = addrx.to_vec().into();
+
+    let addry = "008c5bf0f2af6f1ef08bb3f6ec702dd16e1c514b7e1d12f7549b47db9f4d943c7af0aaec774757d4745d1a2c8dd3220e6ec2c9df23f757a2f8".to_string();
+    let addry: Address = Address::from_hex(&addry).unwrap();
+    let addry: Addr = addry.to_vec().into();
+
+    let addrs: Addrs = vec![addrx, addry];
+    let result = queries_v16::get_utxo_by_address(client, era, addrs)
+        .await
+        .unwrap();
+    info!("result: {:?}", result);
+
     client.send_release().await.unwrap();
 }
 

pallas-codec/src/utils.rs

@@ -37,7 +37,7 @@ impl<C, const N: usize> minicbor::Encode<C> for SkipCbor<N> {
 /// canonicalization for isomorphic decoding / encoding operators, we use a Vec
 /// as the underlaying struct for storage of the items (as opposed to a BTreeMap
 /// or HashMap).
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 #[serde(from = "Vec::<(K, V)>", into = "Vec::<(K, V)>")]
 pub enum KeyValuePairs<K, V>
 where
@@ -475,7 +475,7 @@ where
 }
 
 /// A uint structure that preserves original int length
-#[derive(Debug, PartialEq, Copy, Clone, PartialOrd, Eq, Ord)]
+#[derive(Debug, PartialEq, Copy, Clone, PartialOrd, Eq, Ord, Hash)]
 pub enum AnyUInt {
     MajorByte(u8),
     U8(u8),
@@ -881,7 +881,7 @@ impl fmt::Display for Bytes {
 }
 
 #[derive(
-    Serialize, Deserialize, Clone, Copy, Encode, Decode, Debug, PartialEq, Eq, PartialOrd, Ord,
+    Serialize, Deserialize, Clone, Copy, Encode, Decode, Debug, PartialEq, Eq, PartialOrd, Ord, Hash,
 )]
 #[cbor(transparent)]
 #[serde(into = "i128")]

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

@@ -33,10 +33,10 @@ impl Encode<()> for BlockQuery {
                 e.array(1)?;
                 e.u16(5)?;
             }
-            BlockQuery::GetUTxOByAddress(x) => {
+            BlockQuery::GetUTxOByAddress(addrs) => {
                 e.array(2)?;
                 e.u16(6)?;
-                e.encode(x)?;
+                e.encode(addrs)?;
             }
             BlockQuery::GetUTxOWhole => {
                 e.encode((7,))?;
@@ -129,7 +129,7 @@ impl<'b> Decode<'b, ()> for BlockQuery {
             3 => Ok(Self::GetCurrentPParams),
             4 => Ok(Self::GetProposedPParamsUpdates),
             5 => Ok(Self::GetStakeDistribution),
-            // 6 => Ok(Self::GetUTxOByAddress(())),
+            6 => Ok(Self::GetUTxOByAddress(d.decode()?)),
             // 7 => Ok(Self::GetUTxOWhole),
             // 8 => Ok(Self::DebugEpochState),
             // 9 => Ok(Self::GetCBOR(())),
@@ -264,3 +264,44 @@ impl<'b> Decode<'b, ()> for Request {
         }
     }
 }
+
+impl<'b, C> minicbor::decode::Decode<'b, C> for Value {
+    fn decode(d: &mut minicbor::Decoder<'b>, ctx: &mut C) -> Result<Self, minicbor::decode::Error> {
+        match d.datatype()? {
+            minicbor::data::Type::U8 => Ok(Value::Coin(d.decode_with(ctx)?)),
+            minicbor::data::Type::U16 => Ok(Value::Coin(d.decode_with(ctx)?)),
+            minicbor::data::Type::U32 => Ok(Value::Coin(d.decode_with(ctx)?)),
+            minicbor::data::Type::U64 => Ok(Value::Coin(d.decode_with(ctx)?)),
+            minicbor::data::Type::Array => {
+                d.array()?;
+                let coin = d.decode_with(ctx)?;
+                let multiasset = d.decode_with(ctx)?;
+                Ok(Value::Multiasset(coin, multiasset))
+            }
+            _ => Err(minicbor::decode::Error::message(
+                "unknown cbor data type for Value enum",
+            )),
+        }
+    }
+}
+
+impl<C> minicbor::encode::Encode<C> for Value {
+    fn encode<W: minicbor::encode::Write>(
+        &self,
+        e: &mut minicbor::Encoder<W>,
+        ctx: &mut C,
+    ) -> Result<(), minicbor::encode::Error<W::Error>> {
+        match self {
+            Value::Coin(coin) => {
+                e.encode_with(coin, ctx)?;
+            }
+            Value::Multiasset(coin, other) => {
+                e.array(2)?;
+                e.encode_with(coin, ctx)?;
+                e.encode_with(other, ctx)?;
+            }
+        };
+
+        Ok(())
+    }
+}

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

@@ -1,9 +1,11 @@
 // TODO: this should move to pallas::ledger crate at some point
 
+use pallas_crypto::hash::Hash;
+use std::hash::Hash as StdHash;
 // required for derive attrs to work
 use pallas_codec::minicbor::{self};
 
-use pallas_codec::utils::{Bytes, KeyValuePairs};
+use pallas_codec::utils::{AnyUInt, Bytes, KeyValuePairs, TagWrap};
 use pallas_codec::{
     minicbor::{Decode, Encode},
     utils::AnyCbor,
@@ -25,7 +27,7 @@ pub enum BlockQuery {
     GetCurrentPParams,
     GetProposedPParamsUpdates,
     GetStakeDistribution,
-    GetUTxOByAddress(AnyCbor),
+    GetUTxOByAddress(Addrs),
     GetUTxOWhole,
     DebugEpochState,
     GetCBOR(AnyCbor),
@@ -69,6 +71,12 @@ pub enum Request {
     GetChainPoint,
 }
 
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub enum Value {
+    Coin(Coin),
+    Multiasset(Coin, Multiasset<Coin>),
+}
+
 #[derive(Debug, Encode, Decode, PartialEq)]
 pub struct SystemStart {
     #[n(0)]
@@ -105,6 +113,49 @@ pub struct Fraction {
     pub dem: u64,
 }
 
+pub type Addr = Bytes;
+
+pub type Addrs = Vec<Addr>;
+
+pub type Coin = AnyUInt;
+
+pub type PolicyId = Hash<28>;
+
+pub type AssetName = Bytes;
+
+pub type Multiasset<A> = KeyValuePairs<PolicyId, KeyValuePairs<AssetName, A>>;
+
+#[derive(Debug, Encode, Decode, PartialEq, Clone)]
+pub struct UTxOByAddress {
+    #[n(0)]
+    pub utxo: KeyValuePairs<UTxO, Values>,
+}
+
+// Bytes CDDL ->  #6.121([ * #6.121([ *datum ]) ])
+pub type Datum = (Era, TagWrap<Bytes, 24>);
+
+#[derive(Debug, Encode, Decode, PartialEq, Clone)]
+#[cbor(map)]
+pub struct Values {
+    #[n(0)]
+    pub address: Bytes,
+
+    #[n(1)]
+    pub amount: Value,
+
+    #[n(2)]
+    pub inline_datum: Option<Datum>,
+}
+
+#[derive(Debug, Encode, Decode, PartialEq, Clone, StdHash, Eq)]
+pub struct UTxO {
+    #[n(0)]
+    pub transaction_id: Hash<32>,
+
+    #[n(1)]
+    pub index: AnyUInt,
+}
+
 pub async fn get_chain_point(client: &mut Client) -> Result<Point, ClientError> {
     let query = Request::GetChainPoint;
     let result = client.query(query).await?;
@@ -148,3 +199,16 @@ pub async fn get_stake_distribution(
 
     Ok(result)
 }
+
+pub async fn get_utxo_by_address(
+    client: &mut Client,
+    era: u16,
+    addrs: Addrs,
+) -> Result<UTxOByAddress, ClientError> {
+    let query = BlockQuery::GetUTxOByAddress(addrs);
+    let query = LedgerQuery::BlockQuery(era, query);
+    let query = Request::LedgerQuery(query);
+    let result = client.query(query).await?;
+
+    Ok(result)
+}

pallas-network/tests/protocols.rs

@@ -2,11 +2,13 @@ use std::fs;
 use std::net::{Ipv4Addr, SocketAddrV4};
 use std::time::Duration;
 
-use pallas_codec::utils::{AnyCbor, KeyValuePairs};
+use pallas_codec::utils::{AnyCbor, AnyUInt, KeyValuePairs, TagWrap};
+use pallas_crypto::hash::Hash;
 use pallas_network::facades::{NodeClient, PeerClient, PeerServer};
 use pallas_network::miniprotocols::blockfetch::BlockRequest;
 use pallas_network::miniprotocols::chainsync::{ClientRequest, HeaderContent, Tip};
 use pallas_network::miniprotocols::handshake::n2n::VersionData;
+use pallas_network::miniprotocols::localstate::queries_v16::{Addr, Addrs, Value};
 use pallas_network::miniprotocols::localstate::ClientQueryRequest;
 use pallas_network::miniprotocols::{
     blockfetch,
@@ -539,7 +541,59 @@ pub async fn local_state_query_server_and_client_happy_path() {
             let pools = KeyValuePairs::from(pools);
 
             let result = AnyCbor::from_encode(localstate::queries_v16::StakeDistribution { pools });
+            server.statequery().send_result(result).await.unwrap();
 
+            // server receives query from client
+
+            let query: localstate::queries_v16::Request =
+                match server.statequery().recv_while_acquired().await.unwrap() {
+                    ClientQueryRequest::Query(q) => q.into_decode().unwrap(),
+                    x => panic!("unexpected message from client: {x:?}"),
+                };
+
+            let addr_hex = "981D186018CE18F718FB185F188918A918C7186A186518AC18DD1874186D189E188410184D186F1882184D187D18C4184F1842187F18CA18A118DD";
+            let addr = hex::decode(addr_hex).unwrap();
+            let addr: Addr = addr.to_vec().into();
+            let addrs: Addrs = Vec::from([addr]);
+
+            assert_eq!(
+                query,
+                localstate::queries_v16::Request::LedgerQuery(
+                    localstate::queries_v16::LedgerQuery::BlockQuery(
+                        5,
+                        localstate::queries_v16::BlockQuery::GetUTxOByAddress(addrs),
+                    ),
+                )
+            );
+
+            assert_eq!(*server.statequery().state(), localstate::State::Querying);
+
+            let tx_hex = "1e4e5cf2889d52f1745b941090f04a65dea6ce56c5e5e66e69f65c8e36347c17";
+            let txbytes: [u8; 32] = hex::decode(tx_hex).unwrap().try_into().unwrap();
+            let transaction_id = Hash::from(txbytes);
+            let index = AnyUInt::MajorByte(2);
+            let lovelace = AnyUInt::MajorByte(2);
+            let hex_datum = "9118D81879189F18D81879189F1858181C18C918CF18711866181E185316189118BA";
+            let datum = hex::decode(hex_datum).unwrap().into();
+            let tag = TagWrap::<_, 24>::new(datum);
+            let inline_datum = Some((1_u16, tag));
+            let values = localstate::queries_v16::Values {
+                address: b"addr_test1vr80076l3x5uw6n94nwhgmv7ssgy6muzf47ugn6z0l92rhg2mgtu0"
+                    .to_vec()
+                    .into(),
+                amount: Value::Coin(lovelace),
+                inline_datum,
+            };
+
+            let utxo = KeyValuePairs::from(vec![(
+                localstate::queries_v16::UTxO {
+                    transaction_id,
+                    index,
+                },
+                values,
+            )]);
+
+            let result = AnyCbor::from_encode(localstate::queries_v16::UTxOByAddress { utxo });
             server.statequery().send_result(result).await.unwrap();
 
             assert_eq!(*server.statequery().state(), localstate::State::Acquired);
@@ -650,8 +704,56 @@ pub async fn local_state_query_server_and_client_happy_path() {
 
         assert_eq!(result, localstate::queries_v16::StakeDistribution { pools });
 
-        // client sends a ReAquire
+        let addr_hex = "981D186018CE18F718FB185F188918A918C7186A186518AC18DD1874186D189E188410184D186F1882184D187D18C4184F1842187F18CA18A118DD";
+        let addr = hex::decode(addr_hex).unwrap();
+        let addr: Addr = addr.to_vec().into();
+        let addrs: Addrs = Vec::from([addr]);
 
+        let request = AnyCbor::from_encode(localstate::queries_v16::Request::LedgerQuery(
+            localstate::queries_v16::LedgerQuery::BlockQuery(
+                5,
+                localstate::queries_v16::BlockQuery::GetUTxOByAddress(addrs),
+            ),
+        ));
+
+        client.statequery().send_query(request).await.unwrap();
+
+        let result: localstate::queries_v16::UTxOByAddress = client
+            .statequery()
+            .recv_while_querying()
+            .await
+            .unwrap()
+            .into_decode()
+            .unwrap();
+
+        let tx_hex = "1e4e5cf2889d52f1745b941090f04a65dea6ce56c5e5e66e69f65c8e36347c17";
+        let txbytes: [u8; 32] = hex::decode(tx_hex).unwrap().try_into().unwrap();
+        let transaction_id = Hash::from(txbytes);
+        let index = AnyUInt::MajorByte(2);
+        let lovelace = AnyUInt::MajorByte(2);
+        let hex_datum = "9118D81879189F18D81879189F1858181C18C918CF18711866181E185316189118BA";
+        let datum = hex::decode(hex_datum).unwrap().into();
+        let tag = TagWrap::<_, 24>::new(datum);
+        let inline_datum = Some((1_u16, tag));
+        let values = localstate::queries_v16::Values {
+            address: b"addr_test1vr80076l3x5uw6n94nwhgmv7ssgy6muzf47ugn6z0l92rhg2mgtu0"
+                .to_vec()
+                .into(),
+            amount: Value::Coin(lovelace),
+            inline_datum,
+        };
+
+        let utxo = KeyValuePairs::from(vec![(
+            localstate::queries_v16::UTxO {
+                transaction_id,
+                index,
+            },
+            values,
+        )]);
+
+        assert_eq!(result, localstate::queries_v16::UTxOByAddress { utxo });
+
+        // client sends a ReAquire
         client
             .statequery()
             .send_reacquire(Some(Point::Specific(1337, vec![1, 2, 3])))