feat: introduce StakeAddressDiffsMessage and StakeAddressDiff structures for tracking stake address changes

- Added StakeAddressDiffsMessage to encapsulate a vector of StakeAddressDiffs.
- Implemented StakeAddressDiff to represent individual stake address changes, including registration, deregistration, and delegation.
- Updated AccountsState to handle and publish stake address diffs during state transitions.
- Refactored state handling methods to return StakeAddressDiffs for various operations, enhancing tracking of stake address changes.

Author: golddydev

common/src/messages.rs

@@ -102,6 +102,14 @@ pub struct StakeAddressDeltasMessage {
     pub deltas: Vec<StakeAddressDelta>,
 }
 
+/// Stake address Diffs message
+/// How stake address changes
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+pub struct StakeAddressDiffsMessage {
+    /// Stake Address Changes
+    pub diffs: Vec<StakeAddressDiff>,
+}
+
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct BlockFeesMessage {
     /// Total fees
@@ -237,6 +245,7 @@ pub enum CardanoMessage {
     SPOStakeDistribution(SPOStakeDistributionMessage),   // SPO delegation distribution (SPDD)
     SPORewards(SPORewardsMessage),                       // SPO rewards distribution (SPRD)
     StakeAddressDeltas(StakeAddressDeltasMessage),       // Stake part of address deltas
+    StakeAddressDiffs(StakeAddressDiffsMessage),         // Stake Address Diffs
 }
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]

common/src/types.rs

@@ -140,6 +140,23 @@ pub struct StakeAddressDelta {
     pub delta: i64,
 }
 
+/// Stake Address State Change
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
+pub struct StakeAddressDiff {
+    pub hash: KeyHash,
+    pub change: StakeAddressChange,
+}
+
+/// Stake Address State Change
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
+pub enum StakeAddressChange {
+    StakeAddressRegistered,
+    StakeAddressDeregistered,
+    DelegatedToSPO(KeyHash),
+    DelegatedToDRep(DRepChoice),
+    ValueDiff(LovelaceDelta),
+}
+
 /// Value (lovelace + multiasset)
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct Value {
@@ -698,7 +715,7 @@ pub struct Deregistration {
 }
 
 /// DRepChoice (=CDDL drep, badly named)
-#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
 pub enum DRepChoice {
     /// Address key
     Key(KeyHash),

modules/accounts_state/src/accounts_state.rs

@@ -6,7 +6,7 @@ use acropolis_common::{
     queries::accounts::{PoolsLiveStakes, DEFAULT_ACCOUNTS_QUERY_TOPIC},
     rest_helper::handle_rest,
     state_history::{StateHistory, StateHistoryStore},
-    BlockInfo, BlockStatus,
+    BlockInfo, BlockStatus, StakeAddressDiff,
 };
 use anyhow::Result;
 use caryatid_sdk::{message_bus::Subscription, module, Context, Module};
@@ -21,7 +21,9 @@ mod spo_distribution_publisher;
 use spo_distribution_publisher::SPODistributionPublisher;
 mod spo_rewards_publisher;
 use spo_rewards_publisher::SPORewardsPublisher;
+mod stake_diffs_publisher;
 mod state;
+use stake_diffs_publisher::StakeDiffsPublisher;
 use state::State;
 mod monetary;
 mod rest;
@@ -43,6 +45,7 @@ const DEFAULT_DREP_DISTRIBUTION_TOPIC: &str = "cardano.drep.distribution";
 const DEFAULT_SPO_DISTRIBUTION_TOPIC: &str = "cardano.spo.distribution";
 const DEFAULT_SPO_REWARDS_TOPIC: &str = "cardano.spo.rewards";
 const DEFAULT_PROTOCOL_PARAMETERS_TOPIC: &str = "cardano.protocol.parameters";
+const DEFAULT_STAKE_DIFFS_TOPIC: &str = "cardano.stake.diffs";
 
 const DEFAULT_HANDLE_POTS_TOPIC: (&str, &str) = ("handle-topic-pots", "rest.get.pots");
 
@@ -61,6 +64,7 @@ impl AccountsState {
         mut drep_publisher: DRepDistributionPublisher,
         mut spo_publisher: SPODistributionPublisher,
         mut spo_rewards_publisher: SPORewardsPublisher,
+        mut stake_diffs_publisher: StakeDiffsPublisher,
         mut spos_subscription: Box<dyn Subscription<Message>>,
         mut ea_subscription: Box<dyn Subscription<Message>>,
         mut certs_subscription: Box<dyn Subscription<Message>>,
@@ -113,6 +117,7 @@ impl AccountsState {
             let stake_message_f = stake_subscription.read();
             let withdrawals_message_f = withdrawals_subscription.read();
             let mut current_block: Option<BlockInfo> = None;
+            let mut stake_diffs = Vec::<StakeAddressDiff>::new();
 
             // Use certs_message as the synchroniser, but we have to handle it after the
             // epoch things, because they apply to the new epoch, not the last
@@ -202,13 +207,14 @@ impl AccountsState {
                                 .inspect_err(|e| error!("EpochActivity handling error: {e:#}"))
                                 .ok();
                             // SPO rewards is for previous epoch
-                            if let Some(spo_rewards) = spo_rewards {
+                            if let Some((spo_rewards, diffs)) = spo_rewards {
                                 if let Err(e) = spo_rewards_publisher
                                     .publish_spo_rewards(block_info, spo_rewards)
                                     .await
                                 {
                                     error!("Error publishing SPO rewards: {e:#}")
                                 }
+                                stake_diffs.extend(diffs);
                             }
                         }
                         .instrument(span)
@@ -258,10 +264,13 @@ impl AccountsState {
                     let span = info_span!("account_state.handle_certs", block = block_info.number);
                     async {
                         Self::check_sync(&current_block, &block_info);
-                        state
+                        let diffs = state
                             .handle_tx_certificates(tx_certs_msg)
                             .inspect_err(|e| error!("TxCertificates handling error: {e:#}"))
                             .ok();
+                        if let Some(diffs) = diffs {
+                            stake_diffs.extend(diffs);
+                        }
                     }
                     .instrument(span)
                     .await;
@@ -280,10 +289,13 @@ impl AccountsState {
                     );
                     async {
                         Self::check_sync(&current_block, &block_info);
-                        state
+                        let diffs = state
                             .handle_withdrawals(withdrawals_msg)
                             .inspect_err(|e| error!("Withdrawals handling error: {e:#}"))
                             .ok();
+                        if let Some(diffs) = diffs {
+                            stake_diffs.extend(diffs);
+                        }
                     }
                     .instrument(span)
                     .await;
@@ -302,10 +314,13 @@ impl AccountsState {
                     );
                     async {
                         Self::check_sync(&current_block, &block_info);
-                        state
+                        let diffs = state
                             .handle_stake_deltas(deltas_msg)
                             .inspect_err(|e| error!("StakeAddressDeltas handling error: {e:#}"))
                             .ok();
+                        if let Some(diffs) = diffs {
+                            stake_diffs.extend(diffs);
+                        }
                     }
                     .instrument(span)
                     .await;
@@ -316,6 +331,13 @@ impl AccountsState {
 
             // Commit the new state
             if let Some(block_info) = current_block {
+                // publish stake address diffs message
+                if let Err(e) =
+                    stake_diffs_publisher.publish_stake_diffs(&block_info, stake_diffs).await
+                {
+                    error!("Error publishing stake diffs: {e:#}")
+                }
+
                 history.lock().await.commit(block_info.number, state);
             }
         }
@@ -387,6 +409,11 @@ impl AccountsState {
             .get_string("publish-spo-rewards-topic")
             .unwrap_or(DEFAULT_SPO_REWARDS_TOPIC.to_string());
 
+        let stake_diffs_topic = config
+            .get_string("publish-stake-diffs-topic")
+            .unwrap_or(DEFAULT_STAKE_DIFFS_TOPIC.to_string());
+        info!("Creating stake diffs subscriber on '{stake_diffs_topic}'");
+
         // REST handler topics
         let handle_pots_topic = config
             .get_string(DEFAULT_HANDLE_POTS_TOPIC.0)
@@ -519,6 +546,7 @@ impl AccountsState {
             DRepDistributionPublisher::new(context.clone(), drep_distribution_topic);
         let spo_publisher = SPODistributionPublisher::new(context.clone(), spo_distribution_topic);
         let spo_rewards_publisher = SPORewardsPublisher::new(context.clone(), spo_rewards_topic);
+        let stake_diffs_publisher = StakeDiffsPublisher::new(context.clone(), stake_diffs_topic);
 
         // Subscribe
         let spos_subscription = context.subscribe(&spo_state_topic).await?;
@@ -537,6 +565,7 @@ impl AccountsState {
                 drep_publisher,
                 spo_publisher,
                 spo_rewards_publisher,
+                stake_diffs_publisher,
                 spos_subscription,
                 ea_subscription,
                 certs_subscription,

modules/accounts_state/src/stake_diffs_publisher.rs

@@ -0,0 +1,40 @@
+use acropolis_common::messages::{CardanoMessage, Message, StakeAddressDiffsMessage};
+use acropolis_common::{BlockInfo, StakeAddressDiff};
+use caryatid_sdk::Context;
+use std::sync::Arc;
+
+/// Message publisher for Stake Address Diffs
+pub struct StakeDiffsPublisher {
+    /// Module context
+    context: Arc<Context<Message>>,
+
+    /// Topic to publish on
+    topic: String,
+}
+
+impl StakeDiffsPublisher {
+    /// Construct with context and topic to publish on
+    pub fn new(context: Arc<Context<Message>>, topic: String) -> Self {
+        Self { context, topic }
+    }
+
+    /// Publish the Stake Diffs
+    pub async fn publish_stake_diffs(
+        &mut self,
+        block: &BlockInfo,
+        stake_diffs: Vec<StakeAddressDiff>,
+    ) -> anyhow::Result<()> {
+        self.context
+            .message_bus
+            .publish(
+                &self.topic,
+                Arc::new(Message::Cardano((
+                    block.clone(),
+                    CardanoMessage::StakeAddressDiffs(StakeAddressDiffsMessage {
+                        diffs: stake_diffs,
+                    }),
+                ))),
+            )
+            .await
+    }
+}