feat: replace Rollback with StateTransition

SupernaviX · SupernaviX · commit a763b47900b5 · 2025-11-26T13:04:56.000-05:00
diff --git a/common/src/caryatid.rs b/common/src/caryatid.rs
@@ -3,7 +3,7 @@ use std::sync::Arc;
 use anyhow::Result;
 use caryatid_sdk::{async_trait, Context, MessageBounds, Subscription};
 
-use crate::messages::{CardanoMessage, Message};
+use crate::messages::{CardanoMessage, Message, StateTransitionMessage};
 
 #[async_trait]
 pub trait SubscriptionExt<M: MessageBounds> {
@@ -17,7 +17,10 @@ impl SubscriptionExt<Message> for Box<dyn Subscription<Message>> {
             let (stream, message) = self.read().await?;
             if matches!(
                 message.as_ref(),
-                Message::Cardano((_, CardanoMessage::Rollback(_)))
+                Message::Cardano((
+                    _,
+                    CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_))
+                ))
             ) {
                 continue;
             }
@@ -49,7 +52,10 @@ impl RollbackAwarePublisher<Message> {
 
     pub async fn publish(&mut self, message: Arc<Message>) -> Result<()> {
         match message.as_ref() {
-            Message::Cardano((block, CardanoMessage::Rollback(_))) => {
+            Message::Cardano((
+                block,
+                CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+            )) => {
                 if self.last_activity_at.is_some_and(|slot| slot >= block.slot) {
                     self.last_activity_at = None;
                     self.context.publish(&self.topic, message).await?;
diff --git a/common/src/messages.rs b/common/src/messages.rs
@@ -44,9 +44,9 @@ pub struct RawBlockMessage {
 
 /// Rollback message
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
-pub struct RollbackMessage {
-    /// Point which the chain has rolled back to
-    pub point: Point,
+pub enum StateTransitionMessage {
+    /// The chain has been rolled back to a specific point
+    Rollback(Point),
 }
 
 /// Snapshot completion message
@@ -307,7 +307,7 @@ pub struct SPOStateMessage {
 #[allow(clippy::large_enum_variant)]
 pub enum CardanoMessage {
     BlockAvailable(RawBlockMessage),         // Block body available
-    Rollback(RollbackMessage),               // Chain has been rolled back
+    StateTransition(StateTransitionMessage), // Our position on the chain has changed
     BlockValidation(ValidationStatus),       // Result of a block validation
     SnapshotComplete,                        // Mithril snapshot loaded
     ReceivedTxs(RawTxsMessage),              // Transaction available
diff --git a/modules/accounts_state/src/accounts_state.rs b/modules/accounts_state/src/accounts_state.rs
@@ -3,7 +3,7 @@
 
 use acropolis_common::{
     caryatid::SubscriptionExt,
-    messages::{CardanoMessage, Message, StateQuery, StateQueryResponse},
+    messages::{CardanoMessage, Message, StateQuery, StateQueryResponse, StateTransitionMessage},
     queries::accounts::{DrepDelegators, PoolDelegators, DEFAULT_ACCOUNTS_QUERY_TOPIC},
     state_history::{StateHistory, StateHistoryStore},
     BlockInfo, BlockStatus,
@@ -136,7 +136,10 @@ impl AccountsState {
                     current_block = Some(block_info.clone());
                     block_info.new_epoch && block_info.epoch > 0
                 }
-                Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                Message::Cardano((
+                    _,
+                    CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                )) => {
                     drep_publisher.publish_rollback(certs_message.clone()).await?;
                     spo_publisher.publish_rollback(certs_message.clone()).await?;
                     spo_rewards_publisher.publish_rollback(certs_message.clone()).await?;
@@ -296,7 +299,10 @@ impl AccountsState {
                     .await;
                 }
 
-                Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                Message::Cardano((
+                    _,
+                    CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                )) => {
                     // Ignore this, we already handled rollbacks
                 }
 
diff --git a/modules/block_unpacker/src/block_unpacker.rs b/modules/block_unpacker/src/block_unpacker.rs
@@ -1,7 +1,7 @@
 //! Acropolis Block unpacker module for Caryatid
 //! Unpacks block bodies into transactions
 
-use acropolis_common::messages::{CardanoMessage, Message, RawTxsMessage};
+use acropolis_common::messages::{CardanoMessage, Message, RawTxsMessage, StateTransitionMessage};
 use anyhow::Result;
 use caryatid_sdk::{module, Context};
 use config::Config;
@@ -81,7 +81,10 @@ impl BlockUnpacker {
                         }
                     }
 
-                    Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                    Message::Cardano((
+                        _,
+                        CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                    )) => {
                         // forward the rollback downstream
                         context
                             .message_bus
diff --git a/modules/consensus/src/consensus.rs b/modules/consensus/src/consensus.rs
@@ -2,7 +2,7 @@
 //! Maintains a favoured chain based on offered options from multiple sources
 
 use acropolis_common::{
-    messages::{CardanoMessage, Message},
+    messages::{CardanoMessage, Message, StateTransitionMessage},
     validation::ValidationStatus,
 };
 use anyhow::Result;
@@ -134,7 +134,10 @@ impl Consensus {
                         .await;
                     }
 
-                    Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                    Message::Cardano((
+                        _,
+                        CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                    )) => {
                         // Send rollback to all validators and state modules
                         context
                             .message_bus
diff --git a/modules/drep_state/src/drep_state.rs b/modules/drep_state/src/drep_state.rs
@@ -2,6 +2,7 @@
 //! Accepts certificate events and derives the DRep State in memory
 
 use acropolis_common::caryatid::SubscriptionExt;
+use acropolis_common::messages::StateTransitionMessage;
 use acropolis_common::queries::errors::QueryError;
 use acropolis_common::{
     messages::{CardanoMessage, Message, StateQuery, StateQueryResponse},
@@ -91,7 +92,10 @@ impl DRepState {
                     current_block = Some(block_info.clone());
                     block_info.new_epoch && block_info.epoch > 0
                 }
-                Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                Message::Cardano((
+                    _,
+                    CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                )) => {
                     drep_state_publisher.publish_rollback(certs_message.clone()).await?;
                     false
                 }
@@ -153,7 +157,10 @@ impl DRepState {
                     .await;
                 }
 
-                Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                Message::Cardano((
+                    _,
+                    CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                )) => {
                     // Do nothing, we handled the rollback earlier
                 }
 
diff --git a/modules/epochs_state/src/epochs_state.rs b/modules/epochs_state/src/epochs_state.rs
@@ -3,7 +3,7 @@
 
 use acropolis_common::{
     caryatid::SubscriptionExt,
-    messages::{CardanoMessage, Message, StateQuery, StateQueryResponse},
+    messages::{CardanoMessage, Message, StateQuery, StateQueryResponse, StateTransitionMessage},
     queries::{
         epochs::{
             EpochsStateQuery, EpochsStateQueryResponse, LatestEpoch, DEFAULT_EPOCHS_QUERY_TOPIC,
@@ -136,7 +136,10 @@ impl EpochsState {
                     });
                 }
 
-                Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                Message::Cardano((
+                    _,
+                    CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                )) => {
                     // publish epoch activity rollback message
                     epoch_activity_publisher.publish_rollback(message).await.unwrap_or_else(|e| {
                         error!("Failed to publish epoch activity rollback: {e}")
diff --git a/modules/governance_state/src/governance_state.rs b/modules/governance_state/src/governance_state.rs
@@ -2,6 +2,7 @@
 //! Accepts certificate events and derives the Governance State in memory
 
 use acropolis_common::caryatid::SubscriptionExt;
+use acropolis_common::messages::StateTransitionMessage;
 use acropolis_common::queries::errors::QueryError;
 use acropolis_common::{
     messages::{
@@ -223,7 +224,10 @@ impl GovernanceState {
                 Message::Cardano((blk, CardanoMessage::GovernanceProcedures(msg))) => {
                     (blk.clone(), msg.clone())
                 }
-                Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                Message::Cardano((
+                    _,
+                    CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                )) => {
                     let mut state = state.lock().await;
                     state.publish_rollback(message).await?;
                     continue;
diff --git a/modules/parameters_state/src/parameters_state.rs b/modules/parameters_state/src/parameters_state.rs
@@ -1,6 +1,7 @@
 //! Acropolis Parameter State module for Caryatid
 //! Accepts certificate events and derives the Governance State in memory
 
+use acropolis_common::messages::StateTransitionMessage;
 use acropolis_common::queries::errors::QueryError;
 use acropolis_common::{
     messages::{CardanoMessage, Message, ProtocolParamsMessage, StateQuery, StateQueryResponse},
@@ -145,7 +146,10 @@ impl ParametersState {
                     .instrument(span)
                     .await?;
                 }
-                Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                Message::Cardano((
+                    _,
+                    CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                )) => {
                     // forward the rollback downstream
                     config.context.publish(&config.protocol_parameters_topic, message).await?;
                 }
diff --git a/modules/spo_state/src/spo_state.rs b/modules/spo_state/src/spo_state.rs
@@ -2,6 +2,7 @@
 //! Accepts certificate events and derives the SPO state in memory
 
 use acropolis_common::caryatid::SubscriptionExt;
+use acropolis_common::messages::StateTransitionMessage;
 use acropolis_common::queries::errors::QueryError;
 use acropolis_common::{
     ledger_state::SPOState as LedgerSPOState,
@@ -126,7 +127,10 @@ impl SPOState {
                     block_info.new_epoch && block_info.epoch > 0
                 }
 
-                Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                Message::Cardano((
+                    _,
+                    CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                )) => {
                     spo_state_publisher.publish_rollback(certs_message.clone()).await?;
                     false
                 }
@@ -205,7 +209,10 @@ impl SPOState {
                     .await;
                 }
 
-                Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                Message::Cardano((
+                    _,
+                    CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                )) => {
                     // Do nothing, we handled rollback earlier
                 }
 
diff --git a/modules/stake_delta_filter/src/stake_delta_filter.rs b/modules/stake_delta_filter/src/stake_delta_filter.rs
@@ -3,7 +3,7 @@
 
 use acropolis_common::{
     caryatid::SubscriptionExt,
-    messages::{CardanoMessage, Message},
+    messages::{CardanoMessage, Message, StateTransitionMessage},
     NetworkId,
 };
 use anyhow::{anyhow, Result};
@@ -186,7 +186,10 @@ impl StakeDeltaFilter {
                         .await;
                     }
 
-                    Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                    Message::Cardano((
+                        _,
+                        CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                    )) => {
                         publisher
                             .publish_rollback(message)
                             .await
@@ -269,7 +272,10 @@ impl StakeDeltaFilter {
                         .await;
                     }
 
-                    Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                    Message::Cardano((
+                        _,
+                        CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                    )) => {
                         let mut state = state_deltas.lock().await;
                         state
                             .handle_rollback(message)
diff --git a/modules/tx_unpacker/src/tx_unpacker.rs b/modules/tx_unpacker/src/tx_unpacker.rs
@@ -5,7 +5,7 @@ use acropolis_codec::*;
 use acropolis_common::{
     messages::{
         AssetDeltasMessage, BlockTxsMessage, CardanoMessage, GovernanceProceduresMessage, Message,
-        TxCertificatesMessage, UTXODeltasMessage, WithdrawalsMessage,
+        StateTransitionMessage, TxCertificatesMessage, UTXODeltasMessage, WithdrawalsMessage,
     },
     *,
 };
@@ -466,7 +466,7 @@ impl TxUnpacker {
                         }.instrument(span).await;
                     }
 
-                    Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                    Message::Cardano((_, CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)))) => {
                         let mut futures = Vec::new();
                         if let Some(ref topic) = publish_utxo_deltas_topic {
                             futures.push(context.message_bus.publish(topic, message.clone()));
diff --git a/modules/utxo_state/src/utxo_state.rs b/modules/utxo_state/src/utxo_state.rs
@@ -2,7 +2,7 @@
 //! Accepts UTXO events and derives the current ledger state in memory
 
 use acropolis_common::{
-    messages::{CardanoMessage, Message, StateQuery, StateQueryResponse},
+    messages::{CardanoMessage, Message, StateQuery, StateQueryResponse, StateTransitionMessage},
     queries::utxos::{UTxOStateQuery, UTxOStateQueryResponse, DEFAULT_UTXOS_QUERY_TOPIC},
 };
 use caryatid_sdk::{module, Context};
@@ -101,7 +101,10 @@ impl UTXOState {
                         .await;
                     }
 
-                    Message::Cardano((_, CardanoMessage::Rollback(_))) => {
+                    Message::Cardano((
+                        _,
+                        CardanoMessage::StateTransition(StateTransitionMessage::Rollback(_)),
+                    )) => {
                         let mut state = state1.lock().await;
                         state
                             .handle_rollback(message)
