feat(drive-abci): state sync via ABCI snapshots with reduced platform state (protocol v15)

packages/rs-dpp/src/lib.rs

@@ -61,6 +61,8 @@ pub mod core_subsidy;
 pub mod fee;
 pub mod nft;
 pub mod prefunded_specialized_balance;
+/// Reduced platform state stored in replicated state for state sync reconstruction
+pub mod reduced_platform_state;
 pub mod serialization;
 #[cfg(any(
     feature = "message-signing",

packages/rs-dpp/src/reduced_platform_state/mod.rs

@@ -0,0 +1,106 @@
+//! Reduced platform state
+//!
+//! A minimal subset of the Platform state that is stored inside the replicated GroveDB
+//! state (under the Misc tree), allowing a node that syncs via ABCI state sync to
+//! reconstruct the full Platform state. The full Platform state itself is only persisted
+//! to GroveDB aux storage, which is not replicated by GroveDB state sync.
+
+use crate::ProtocolError;
+use bincode::{Decode, Encode};
+use derive_more::From;
+use platform_serialization_derive::{PlatformDeserialize, PlatformSerialize};
+
+pub mod v0;
+
+use v0::ReducedPlatformStateV0;
+
+/// Reduced Platform State (platform-versioned wrapper)
+///
+/// The structure version is the enum discriminant, so it serializes `unversioned` (big
+/// endian, no limit) exactly like the other versioned platform types. These bytes are
+/// covered by the app hash, so the encoding is consensus-fixed.
+#[derive(Clone, Debug, PartialEq, Encode, Decode, PlatformSerialize, PlatformDeserialize, From)]
+#[platform_serialize(unversioned)]
+pub enum ReducedPlatformState {
+    /// Version 0
+    V0(ReducedPlatformStateV0),
+}
+
+#[cfg(test)]
+mod tests {
+    use super::v0::{
+        ReducedBlockInfoV0, ReducedPlatformStateV0, ReducedPreviousQuorumsV0,
+        ReducedVerificationQuorumV0,
+    };
+    use super::*;
+    use crate::block::block_info::BlockInfo;
+    use crate::serialization::{PlatformDeserializable, PlatformSerializable};
+
+    #[test]
+    fn should_roundtrip_reduced_platform_state_serialization() {
+        let state = ReducedPlatformState::V0(ReducedPlatformStateV0 {
+            last_committed_block_info: Some(ReducedBlockInfoV0 {
+                basic_info: BlockInfo::default_with_time(1_700_000_000_000),
+                quorum_hash: [1u8; 32].into(),
+                proposer_pro_tx_hash: [2u8; 32].into(),
+            }),
+            current_protocol_version_in_consensus: 15,
+            next_epoch_protocol_version: 15,
+            current_validator_set_quorum_hash: [4u8; 32].into(),
+            next_validator_set_quorum_hash: Some([5u8; 32].into()),
+            previous_fee_versions: [(0u16, 1u32)].into_iter().collect(),
+            quorum_positions: vec![[4u8; 32].into(), [5u8; 32].into()],
+            proposed_core_chain_locked_height: 1000,
+            previous_chain_lock_quorums: Some(ReducedPreviousQuorumsV0 {
+                quorums: vec![ReducedVerificationQuorumV0 {
+                    quorum_hash: [6u8; 32].into(),
+                    public_key: [7u8; 48],
+                    index: None,
+                }],
+                last_active_core_height: 990,
+                updated_at_core_height: 995,
+                previous_change_height: Some(900),
+            }),
+            previous_instant_lock_quorums: Some(ReducedPreviousQuorumsV0 {
+                quorums: vec![ReducedVerificationQuorumV0 {
+                    quorum_hash: [8u8; 32].into(),
+                    public_key: [9u8; 48],
+                    index: Some(3),
+                }],
+                last_active_core_height: 991,
+                updated_at_core_height: 996,
+                previous_change_height: None,
+            }),
+        });
+
+        let bytes = state.serialize_to_bytes().expect("should serialize");
+        let restored =
+            ReducedPlatformState::deserialize_from_bytes(&bytes).expect("should deserialize");
+
+        assert_eq!(state, restored);
+    }
+
+    /// The reduced state is encoded like every other versioned platform type: big
+    /// endian. Pin it so the app-hash-covered encoding cannot drift silently.
+    #[test]
+    fn should_encode_big_endian_like_other_platform_types() {
+        let state = ReducedPlatformState::V0(ReducedPlatformStateV0 {
+            last_committed_block_info: None,
+            current_protocol_version_in_consensus: 0x0102_0304,
+            next_epoch_protocol_version: 0,
+            current_validator_set_quorum_hash: [0u8; 32].into(),
+            next_validator_set_quorum_hash: None,
+            previous_fee_versions: Default::default(),
+            quorum_positions: vec![],
+            proposed_core_chain_locked_height: 0,
+            previous_chain_lock_quorums: None,
+            previous_instant_lock_quorums: None,
+        });
+
+        let bytes = state.serialize_to_bytes().expect("should serialize");
+        // discriminant 0, then `None`, then the protocol version as a big-endian varint
+        // (bincode's varint marker 0xfc precedes a u32 payload)
+        assert_eq!(&bytes[..2], &[0u8, 0u8]);
+        assert_eq!(&bytes[2..7], &[0xfc, 0x01, 0x02, 0x03, 0x04]);
+    }
+}

packages/rs-dpp/src/reduced_platform_state/v0/mod.rs

@@ -0,0 +1,96 @@
+use crate::block::block_info::BlockInfo;
+use crate::fee::default_costs::EpochIndexFeeVersionsForStorage;
+use crate::util::deserializer::ProtocolVersion;
+use bincode::{Decode, Encode};
+use platform_value::Bytes32;
+
+/// Block information persisted as part of the reduced platform state.
+///
+/// Only what the block header fixes goes in here. The reduced state is written during
+/// block execution and covered by the app hash, so anything that can differ between two
+/// proposals of the same block (the consensus round, the app hash itself, the block id
+/// hash and the signature) must stay out: a re-proposal of the same header at a later
+/// round has to produce the same app hash. A state-synced node takes the app hash from
+/// the snapshot and learns the rest with the next finalized block.
+#[derive(Clone, Debug, PartialEq, Encode, Decode)]
+pub struct ReducedBlockInfoV0 {
+    /// Basic block info (height, core height, time, epoch)
+    pub basic_info: BlockInfo,
+    /// The quorum that signed (or will sign) this block
+    pub quorum_hash: Bytes32,
+    /// The block proposer's pro tx hash
+    pub proposer_pro_tx_hash: Bytes32,
+}
+
+/// One quorum of a signature-verification quorum set, as persisted in the reduced
+/// platform state.
+#[derive(Clone, Debug, PartialEq, Encode, Decode)]
+pub struct ReducedVerificationQuorumV0 {
+    /// The quorum hash
+    pub quorum_hash: Bytes32,
+    /// The quorum's threshold BLS public key, compressed (48 bytes)
+    pub public_key: [u8; 48],
+    /// The DIP24 rotation index, for rotating quorum types
+    pub index: Option<u32>,
+}
+
+/// The superseded quorums of a signature-verification quorum set, together with the core
+/// heights that define the window they are still authoritative for.
+///
+/// This history CANNOT be recovered from Core: `get_quorum_listextended` answers "which
+/// quorums exist at height h", not "when did this node observe the set change". It is
+/// nonetheless consensus-relevant — `select_quorums` picks the previous set for locks
+/// signed within `SIGN_OFFSET` core blocks of a change — so it has to travel with the
+/// snapshot. Without it a restored node would judge an instant lock against a different
+/// quorum than a node that replayed the chain, and reject a state transition the network
+/// accepted.
+#[derive(Clone, Debug, PartialEq, Encode, Decode)]
+pub struct ReducedPreviousQuorumsV0 {
+    /// The superseded quorums
+    pub quorums: Vec<ReducedVerificationQuorumV0>,
+    /// The core height at which these quorums were last active
+    pub last_active_core_height: u32,
+    /// The core height at which the quorums were changed
+    pub updated_at_core_height: u32,
+    /// The core height at which the set before these became active
+    pub previous_change_height: Option<u32>,
+}
+
+/// Reduced Platform State V0.
+///
+/// This minimal version of the Platform state is written into GroveDB (under the Misc
+/// tree, hence below the root hash) on every block proposal. Because it is part of the
+/// replicated state, a freshly state-synced node can read it back and reconstruct the
+/// full in-memory Platform state, which is otherwise only persisted to non-replicated
+/// GroveDB aux storage.
+#[derive(Clone, Debug, PartialEq, Encode, Decode)]
+pub struct ReducedPlatformStateV0 {
+    /// Info about the block that was being processed when this state was written
+    /// (it becomes the last committed block once the block finalizes)
+    pub last_committed_block_info: Option<ReducedBlockInfoV0>,
+    /// Current protocol version in consensus
+    pub current_protocol_version_in_consensus: ProtocolVersion,
+    /// Upcoming protocol version
+    pub next_epoch_protocol_version: ProtocolVersion,
+    /// Current validator set quorum hash
+    pub current_validator_set_quorum_hash: Bytes32,
+    /// Next validator set quorum hash
+    pub next_validator_set_quorum_hash: Option<Bytes32>,
+    /// Fee versions of previous epochs, stored by fee version number so they can be
+    /// restored faithfully on reconstruction
+    pub previous_fee_versions: EpochIndexFeeVersionsForStorage,
+    /// Ordered list of quorum hashes reflecting validator set quorum positions
+    // TODO: optimize this to not store the whole quorum hash, but only some index
+    pub quorum_positions: Vec<Bytes32>,
+    /// Core chain locked height, as provided in RequestProcessProposal ABCI message;
+    /// note this can differ from the one in RequestPrepareProposal, as it can be
+    /// modified by the proposer.
+    pub proposed_core_chain_locked_height: u32,
+    /// The superseded chain lock validating quorums, if any. The CURRENT set is
+    /// re-derived from Core during reconstruction (it is exactly the quorum list at
+    /// `proposed_core_chain_locked_height`); only the history has to be carried.
+    pub previous_chain_lock_quorums: Option<ReducedPreviousQuorumsV0>,
+    /// The superseded instant lock validating quorums, if any. See
+    /// [`ReducedPreviousQuorumsV0`] for why this cannot be left to reconstruction.
+    pub previous_instant_lock_quorums: Option<ReducedPreviousQuorumsV0>,
+}

packages/rs-drive-abci/.env.local

@@ -90,3 +90,10 @@ GROVEDB_VISUALIZER_ENABLED=false
 GROVEDB_VISUALIZER_ADDRESS=127.0.0.1:8083
 
 NETWORK=regtest
+
+# ABCI state sync snapshots (serving side; disabled by default)
+SNAPSHOTS_ENABLED=false
+SNAPSHOTS_FREQUENCY_SECONDS=600
+MAX_NUM_SNAPSHOTS=3
+# CHECKPOINTS_PATH defaults to <db_path>/checkpoints when unset
+#CHECKPOINTS_PATH=

packages/rs-drive-abci/.env.mainnet

@@ -92,3 +92,10 @@ GROVEDB_VISUALIZER_ADDRESS=127.0.0.1:8083
 PROPOSER_TX_PROCESSING_TIME_LIMIT=5000
 
 NETWORK=mainnet
+
+# ABCI state sync snapshots (serving side; disabled by default)
+SNAPSHOTS_ENABLED=false
+SNAPSHOTS_FREQUENCY_SECONDS=600
+MAX_NUM_SNAPSHOTS=3
+# CHECKPOINTS_PATH defaults to <db_path>/checkpoints when unset
+#CHECKPOINTS_PATH=

packages/rs-drive-abci/.env.testnet

@@ -92,3 +92,10 @@ GROVEDB_VISUALIZER_ADDRESS=127.0.0.1:8083
 PROPOSER_TX_PROCESSING_TIME_LIMIT=5000
 
 NETWORK=testnet
+
+# ABCI state sync snapshots (serving side; disabled by default)
+SNAPSHOTS_ENABLED=false
+SNAPSHOTS_FREQUENCY_SECONDS=600
+MAX_NUM_SNAPSHOTS=3
+# CHECKPOINTS_PATH defaults to <db_path>/checkpoints when unset
+#CHECKPOINTS_PATH=

packages/rs-drive-abci/src/abci/app/check_tx.rs

@@ -2,6 +2,7 @@ use crate::abci::app::PlatformApplication;
 use crate::abci::handler;
 use crate::error::Error;
 use crate::platform_types::platform::Platform;
+use crate::platform_types::snapshot::SnapshotManager;
 use crate::rpc::core::CoreRPCLike;
 use crate::utils::spawn_blocking_task_with_name_if_supported;
 use async_trait::async_trait;
@@ -22,6 +23,12 @@ where
     /// Platform
     platform: Arc<Platform<C>>,
     core_rpc: Arc<C>,
+    /// The snapshot manager, pinning checkpoints that are being served to peers.
+    ///
+    /// Shared (`Arc`) rather than owned: this application never sees blocks, so it
+    /// cannot expire the pins of abandoned transfers itself — `server::start` keeps a
+    /// clone and sweeps expired pins on a timer.
+    snapshot_manager: Arc<SnapshotManager>,
 }
 
 impl<C> PlatformApplication<C> for CheckTxAbciApplication<C>
@@ -38,8 +45,16 @@ where
     C: CoreRPCLike + Send + Sync + 'static,
 {
     /// Create new ABCI app
-    pub fn new(platform: Arc<Platform<C>>, core_rpc: Arc<C>) -> Self {
-        Self { platform, core_rpc }
+    pub fn new(
+        platform: Arc<Platform<C>>,
+        core_rpc: Arc<C>,
+        snapshot_manager: Arc<SnapshotManager>,
+    ) -> Self {
+        Self {
+            platform,
+            core_rpc,
+            snapshot_manager,
+        }
     }
 }
 
@@ -92,6 +107,47 @@ where
         .await
         .map_err(|error| tonic::Status::internal(format!("check tx panics: {}", error)))?
     }
+
+    async fn list_snapshots(
+        &self,
+        request: tonic::Request<proto::RequestListSnapshots>,
+    ) -> Result<tonic::Response<proto::ResponseListSnapshots>, tonic::Status> {
+        // Checkpoint metadata reads are synchronous rocksdb work; requests are
+        // peer-controlled, so keep them off the async workers (same pattern as check_tx)
+        let platform = Arc::clone(&self.platform);
+        let proto_request = request.into_inner();
+
+        spawn_blocking_task_with_name_if_supported("list_snapshots", move || {
+            handler::list_snapshots(platform.as_ref(), proto_request)
+                .map(tonic::Response::new)
+                .map_err(error_into_status)
+        })?
+        .await
+        .map_err(|error| tonic::Status::internal(format!("list snapshots panics: {}", error)))?
+    }
+
+    async fn load_snapshot_chunk(
+        &self,
+        request: tonic::Request<proto::RequestLoadSnapshotChunk>,
+    ) -> Result<tonic::Response<proto::ResponseLoadSnapshotChunk>, tonic::Status> {
+        // Chunk generation traverses the checkpoint's grovedb and encodes a replication
+        // chunk — synchronous, potentially large, and peer-controlled. Run it on the
+        // blocking pool so concurrent snapshot consumers cannot occupy the async workers
+        // and delay unrelated gRPC traffic (same pattern as check_tx).
+        let platform = Arc::clone(&self.platform);
+        let snapshot_manager = Arc::clone(&self.snapshot_manager);
+        let proto_request = request.into_inner();
+
+        spawn_blocking_task_with_name_if_supported("load_snapshot_chunk", move || {
+            handler::load_snapshot_chunk(platform.as_ref(), &snapshot_manager, proto_request)
+                .map(tonic::Response::new)
+                .map_err(error_into_status)
+        })?
+        .await
+        .map_err(|error| {
+            tonic::Status::internal(format!("load snapshot chunk panics: {}", error))
+        })?
+    }
 }
 
 pub fn error_into_status(error: Error) -> tonic::Status {
@@ -146,7 +202,11 @@ mod tests {
 
         let core_rpc = MockCoreRPCLike::new();
 
-        let app = CheckTxAbciApplication::new(Arc::new(platform.platform), Arc::new(core_rpc));
+        let app = CheckTxAbciApplication::new(
+            Arc::new(platform.platform),
+            Arc::new(core_rpc),
+            Arc::new(SnapshotManager::new()),
+        );
 
         let debug_str = format!("{:?}", app);
         assert_eq!(debug_str, "<CheckTxAbciApplication>");
@@ -159,7 +219,11 @@ mod tests {
 
         let core_rpc = MockCoreRPCLike::new();
 
-        let app = CheckTxAbciApplication::new(Arc::new(platform.platform), Arc::new(core_rpc));
+        let app = CheckTxAbciApplication::new(
+            Arc::new(platform.platform),
+            Arc::new(core_rpc),
+            Arc::new(SnapshotManager::new()),
+        );
 
         // Just verify we can call platform() without panicking
         let _platform_ref = app.platform();