feat(network): implement stake distribution local state query (#340)

examples/n2c-miniprotocols/src/main.rs

@@ -21,7 +21,11 @@ async fn do_localstate_query(client: &mut NodeClient) {
     let result = queries_v16::get_block_epoch_number(client, era)
         .await
         .unwrap();
+    info!("result: {:?}", result);
 
+    let result = queries_v16::get_stake_distribution(client, era)
+        .await
+        .unwrap();
     info!("result: {:?}", result);
 
     client.send_release().await.unwrap();

pallas-codec/src/utils.rs

@@ -829,7 +829,9 @@ where
     }
 }
 
-#[derive(Serialize, Deserialize, Clone, Encode, Decode, Debug, PartialEq, Eq, PartialOrd, Ord)]
+#[derive(
+    Serialize, Deserialize, Clone, Encode, Decode, Debug, PartialEq, Eq, PartialOrd, Ord, Hash,
+)]
 #[cbor(transparent)]
 #[serde(into = "String")]
 #[serde(try_from = "String")]

pallas-network/Cargo.toml

@@ -24,3 +24,4 @@ tracing = "0.1.37"
 tracing-subscriber = "0.3.16"
 tokio = { version = "1", features = ["full"] }
 rand = "0.8.5"
+

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

@@ -1,8 +1,9 @@
 // TODO: this should move to pallas::ledger crate at some point
 
 // required for derive attrs to work
-use pallas_codec::minicbor;
+use pallas_codec::minicbor::{self};
 
+use pallas_codec::utils::{Bytes, KeyValuePairs};
 use pallas_codec::{
     minicbor::{Decode, Encode},
     utils::AnyCbor,
@@ -80,6 +81,30 @@ pub struct SystemStart {
     pub picoseconds_of_day: u64,
 }
 
+#[derive(Debug, Encode, Decode, PartialEq)]
+pub struct StakeDistribution {
+    #[n(0)]
+    pub pools: KeyValuePairs<Bytes, Pool>,
+}
+
+#[derive(Debug, Encode, Decode, PartialEq, Clone)]
+pub struct Pool {
+    #[n(0)]
+    pub stakes: Fraction,
+
+    #[n(1)]
+    pub hashes: Bytes,
+}
+
+#[derive(Debug, Encode, Decode, PartialEq, Clone)]
+pub struct Fraction {
+    #[n(0)]
+    pub num: u64,
+
+    #[n(1)]
+    pub dem: u64,
+}
+
 pub async fn get_chain_point(client: &mut Client) -> Result<Point, ClientError> {
     let query = Request::GetChainPoint;
     let result = client.query(query).await?;
@@ -111,3 +136,15 @@ pub async fn get_block_epoch_number(client: &mut Client, era: u16) -> Result<u32
 
     Ok(result)
 }
+
+pub async fn get_stake_distribution(
+    client: &mut Client,
+    era: u16,
+) -> Result<StakeDistribution, ClientError> {
+    let query = BlockQuery::GetStakeDistribution;
+    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,7 +2,7 @@ use std::fs;
 use std::net::{Ipv4Addr, SocketAddrV4};
 use std::time::Duration;
 
-use pallas_codec::utils::AnyCbor;
+use pallas_codec::utils::{AnyCbor, KeyValuePairs};
 use pallas_network::facades::{NodeClient, PeerClient, PeerServer};
 use pallas_network::miniprotocols::blockfetch::BlockRequest;
 use pallas_network::miniprotocols::chainsync::{ClientRequest, HeaderContent, Tip};
@@ -454,7 +454,6 @@ pub async fn chainsync_server_and_client_happy_path_n2n() {
 }
 
 #[tokio::test]
-#[ignore]
 pub async fn local_state_query_server_and_client_happy_path() {
     let server = tokio::spawn({
         async move {
@@ -503,6 +502,48 @@ pub async fn local_state_query_server_and_client_happy_path() {
 
             assert_eq!(*server.statequery().state(), localstate::State::Acquired);
 
+            // 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:?}"),
+                };
+
+            assert_eq!(
+                query,
+                localstate::queries_v16::Request::LedgerQuery(
+                    localstate::queries_v16::LedgerQuery::BlockQuery(
+                        5,
+                        localstate::queries_v16::BlockQuery::GetStakeDistribution,
+                    ),
+                )
+            );
+            assert_eq!(*server.statequery().state(), localstate::State::Querying);
+
+            let fraction = localstate::queries_v16::Fraction { num: 10, dem: 20 };
+            let pool = localstate::queries_v16::Pool {
+                stakes: fraction.clone(),
+                hashes: b"pool1qv4qgv62s3ha74p0643nexee9zvcdydcyahqqnavhj90zheuykz"
+                    .to_vec()
+                    .into(),
+            };
+
+            let pools = vec![(
+                b"pool1qvfw4r3auysa5mhpr90n7mmdhs55js8gdywh0y2e3sy6568j2wp"
+                    .to_vec()
+                    .into(),
+                pool,
+            )];
+
+            let pools = KeyValuePairs::from(pools);
+
+            let result = AnyCbor::from_encode(localstate::queries_v16::StakeDistribution { pools });
+
+            server.statequery().send_result(result).await.unwrap();
+
+            assert_eq!(*server.statequery().state(), localstate::State::Acquired);
+
             // server receives re-acquire from the client
 
             let maybe_point = match server.statequery().recv_while_acquired().await.unwrap() {
@@ -573,6 +614,42 @@ pub async fn local_state_query_server_and_client_happy_path() {
             }
         );
 
+        let request = AnyCbor::from_encode(localstate::queries_v16::Request::LedgerQuery(
+            localstate::queries_v16::LedgerQuery::BlockQuery(
+                5,
+                localstate::queries_v16::BlockQuery::GetStakeDistribution,
+            ),
+        ));
+
+        client.statequery().send_query(request).await.unwrap();
+
+        let result: localstate::queries_v16::StakeDistribution = client
+            .statequery()
+            .recv_while_querying()
+            .await
+            .unwrap()
+            .into_decode()
+            .unwrap();
+
+        let fraction = localstate::queries_v16::Fraction { num: 10, dem: 20 };
+        let pool = localstate::queries_v16::Pool {
+            stakes: fraction.clone(),
+            hashes: b"pool1qv4qgv62s3ha74p0643nexee9zvcdydcyahqqnavhj90zheuykz"
+                .to_vec()
+                .into(),
+        };
+
+        let pools = vec![(
+            b"pool1qvfw4r3auysa5mhpr90n7mmdhs55js8gdywh0y2e3sy6568j2wp"
+                .to_vec()
+                .into(),
+            pool,
+        )];
+
+        let pools = KeyValuePairs::from(pools);
+
+        assert_eq!(result, localstate::queries_v16::StakeDistribution { pools });
+
         // client sends a ReAquire
 
         client