dashpay · ogabrielides · Jan 8, 2025 · Jan 9, 2025 · Jan 13, 2025 · Jan 13, 2025
@@ -552,19 +552,24 @@ LABEL description="Drive ABCI Rust"
 RUN apk add --no-cache libgcc libstdc++
 
 ENV DB_PATH=/var/lib/dash/rs-drive-abci/db
+ENV CHECKPOINTS_PATH=/var/lib/dash/rs-drive-abci/db-checkpoints
 ENV REJECTIONS_PATH=/var/log/dash/rejected
 
-RUN mkdir -p /var/log/dash \
-    /var/lib/dash/rs-drive-abci/db \
-    ${REJECTIONS_PATH}
-
 COPY --from=build-drive-abci /artifacts/drive-abci /usr/bin/drive-abci
 COPY packages/rs-drive-abci/.env.mainnet /var/lib/dash/rs-drive-abci/.env
 
 # Create a volume
 VOLUME /var/lib/dash/rs-drive-abci/db
 VOLUME /var/log/dash
 
+# Ensure required paths do exist
+# TODO: remove /var/lib/dash-platform/data/checkpoints when drive-abci is fixed
+RUN mkdir -p /var/log/dash \
+    ${DB_PATH} \
+    ${CHECKPOINTS_PATH} \
+    ${REJECTIONS_PATH} \
+    /var/lib/dash-platform/data/checkpoints
+
 # Double-check that we don't have missing deps
 RUN ldd /usr/bin/drive-abci
 
@@ -574,9 +579,10 @@ RUN ldd /usr/bin/drive-abci
 ARG USERNAME=dash
 ARG USER_UID=1000
 ARG USER_GID=$USER_UID
+# TODO: remove /var/lib/dash-platform/data/checkpoints when drive-abci is fixed
 RUN addgroup -g $USER_GID $USERNAME && \
     adduser -D -u $USER_UID -G $USERNAME -h /var/lib/dash/rs-drive-abci $USERNAME && \
-    chown -R $USER_UID:$USER_GID /var/lib/dash/rs-drive-abci /var/log/dash
+    chown -R $USER_UID:$USER_GID /var/lib/dash/rs-drive-abci /var/log/dash /var/lib/dash-platform/data/checkpoints
 
 USER $USERNAME
 

@@ -42,8 +42,8 @@ tonic = { version = "0.12.3", features = [
 serde = { version = "1.0.197", optional = true, features = ["derive"] }
 serde_bytes = { version = "0.11.12", optional = true }
 serde_json = { version = "1.0", optional = true }
-tenderdash-proto = { git = "https://github.com/dashpay/rs-tenderdash-abci", version = "1.2.1", tag = "v1.2.1+1.3.0", default-features = false, features = [
-  "grpc",
+tenderdash-proto  = { git = "https://github.com/dashpay/rs-tenderdash-abci", rev="47141571a9df6cda83d67bbf68091f3263799eed", default-features = false, features = [
+  "grpc", "serde",
 ] }
 dapi-grpc-macros = { path = "../rs-dapi-grpc-macros" }
 platform-version = { path = "../rs-platform-version" }

@@ -309,7 +309,7 @@ export default function getBaseConfigFactory() {
           tenderdash: {
             mode: 'full',
             docker: {
-              image: 'dashpay/tenderdash:1',
+              image: 'dashpay/tenderdash:feat-statesync-integration',
             },
             p2p: {
               host: '0.0.0.0',

@@ -81,7 +81,7 @@ filter-peers = false
 # Example for routed multi-app setup:
 #   abci = "routed"
 #   address = "Info:socket:unix:///tmp/socket.1,Info:socket:unix:///tmp/socket.2,CheckTx:socket:unix:///tmp/socket.1,*:socket:unix:///tmp/socket.3"
-address = "CheckTx:grpc:drive_abci:26670,*:socket:tcp://drive_abci:26658"
+address = "ListSnapshots:grpc:drive_abci:26670,LoadSnapshotChunk:grpc:drive_abci:26670,CheckTx:grpc:drive_abci:26670,*:socket:tcp://drive_abci:26658"
 # Transport mechanism to connect to the ABCI application: socket | grpc | routed
 transport = "routed"
 # Maximum number of simultaneous connections to the ABCI application
@@ -97,6 +97,10 @@ transport = "routed"
 #]
 grpc-concurrency = [
   { "check_tx" = {{= it.platform.drive.tenderdash.mempool.maxConcurrentCheckTx }} },
+  { "list_snapshots" = {{= it.platform.drive.tenderdash.mempool.maxConcurrentCheckTx }} },
+  { "load_snapshot_chunk" = {{= it.platform.drive.tenderdash.mempool.maxConcurrentCheckTx }} },
+  { "offer_snapshot" = 1 },
+  { "apply_snapshot_chunk" = 1 },
 ]
 
 
@@ -414,26 +418,17 @@ ttl-num-blocks = {{=it.platform.drive.tenderdash.mempool.ttlNumBlocks}}
 # the network to take and serve state machine snapshots. State sync is not attempted if the node
 # has any local state (LastBlockHeight > 0). The node will have a truncated block history,
 # starting from the height of the snapshot.
-enable = false
+enable = true
 
 # State sync uses light client verification to verify state. This can be done either through the
 # P2P layer or RPC layer. Set this to true to use the P2P layer. If false (default), RPC layer
 # will be used.
-use-p2p = false
+use-p2p = true
 
 # If using RPC, at least two addresses need to be provided. They should be compatible with net.Dial,
 # for example: "host.example.com:2125"
 rpc-servers = ""
 
-# The hash and height of a trusted block. Must be within the trust-period.
-trust-height = 0
-trust-hash = ""
-
-# The trust period should be set so that Tendermint can detect and gossip misbehavior before
-# it is considered expired. For chains based on the Cosmos SDK, one day less than the unbonding
-# period should suffice.
-trust-period = "168h0m0s"
-
 # Time to spend discovering snapshots before initiating a restore.
 discovery-time = "15s"
 

@@ -12,6 +12,12 @@ ABCI_LOG_STDOUT_FORMAT=pretty
 ABCI_LOG_STDOUT_COLOR=true
 
 DB_PATH=/tmp/db
+
+CHECKPOINTS_PATH=${DB_PATH}/checkpoints
+
+# GroveDB database file
+GROVEDB_LATEST_FILE=${DB_PATH}/latest_state
+
 REJECTIONS_PATH=/tmp/rejected
 
 # Cache size for Data Contracts

@@ -49,8 +49,8 @@ tracing-subscriber = { version = "0.3.16", default-features = false, features =
   "registry",
   "tracing-log",
 ], optional = false }
-tenderdash-abci = { git = "https://github.com/dashpay/rs-tenderdash-abci", version = "1.2.1", tag = "v1.2.1+1.3.0", features = [
-  "grpc",
+tenderdash-abci = { git = "https://github.com/dashpay/rs-tenderdash-abci", rev="47141571a9df6cda83d67bbf68091f3263799eed", features = [
+  "crypto", "grpc", "serde",
 ] }
 lazy_static = "1.4.0"
 itertools = { version = "0.13" }

@@ -1,7 +1,8 @@
-use crate::abci::app::PlatformApplication;
+use crate::abci::app::{PlatformApplication, SnapshotManagerApplication};
 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,8 @@ where
     /// Platform
     platform: Arc<Platform<C>>,
     core_rpc: Arc<C>,
+    /// Snapshot manager
+    snapshot_manager: SnapshotManager,
 }
 
 impl<C> PlatformApplication<C> for CheckTxAbciApplication<C>
@@ -33,13 +36,31 @@ where
     }
 }
 
+impl<C> SnapshotManagerApplication for CheckTxAbciApplication<C>
+where
+    C: CoreRPCLike + Send + Sync + 'static,
+{
+    fn snapshot_manager(&self) -> &SnapshotManager {
+        &self.snapshot_manager
+    }
+}
+
 impl<C> CheckTxAbciApplication<C>
 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 }
+        let snapshot_manager = SnapshotManager::new(
+            platform.config.state_sync_config.checkpoints_path.clone(),
+            platform.config.state_sync_config.max_num_snapshots,
+            platform.config.state_sync_config.snapshots_frequency,
+        );
+        Self {
+            platform,
+            core_rpc,
+            snapshot_manager,
+        }
     }
 }
 
@@ -92,6 +113,24 @@ 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> {
+        handler::list_snapshots(self, request.into_inner())
+            .map(tonic::Response::new)
+            .map_err(|e| tonic::Status::internal(format!("list_snapshots failed: {}", e)))
+    }
+
+    async fn load_snapshot_chunk(
+        &self,
+        request: tonic::Request<proto::RequestLoadSnapshotChunk>,
+    ) -> Result<tonic::Response<proto::ResponseLoadSnapshotChunk>, tonic::Status> {
+        handler::load_snapshot_chunk(self, request.into_inner())
+            .map(tonic::Response::new)
+            .map_err(|e| tonic::Status::internal(format!("load_snapshot_chunk failed: {}", e)))
+    }
 }
 
 pub fn error_into_status(error: Error) -> tonic::Status {

@@ -1,11 +1,16 @@
-use crate::abci::app::{BlockExecutionApplication, PlatformApplication, TransactionalApplication};
+use crate::abci::app::{
+    BlockExecutionApplication, PlatformApplication, SnapshotFetchingApplication,
+    SnapshotManagerApplication, TransactionalApplication,
+};
 use crate::abci::handler;
 use crate::abci::handler::error::error_into_exception;
 use crate::error::execution::ExecutionError;
 use crate::error::Error;
 use crate::execution::types::block_execution_context::BlockExecutionContext;
 use crate::platform_types::platform::Platform;
+use crate::platform_types::snapshot::{SnapshotFetchingSession, SnapshotManager};
 use crate::rpc::core::CoreRPCLike;
+use dapi_grpc::tonic;
 use dpp::version::PlatformVersion;
 use drive::grovedb::Transaction;
 use std::fmt::Debug;
@@ -16,46 +21,74 @@ use tenderdash_abci::proto::abci as proto;
 ///
 /// AbciApp implements logic that should be triggered when Tenderdash performs various operations, like
 /// creating new proposal or finalizing new block.
-pub struct ConsensusAbciApplication<'a, C> {
+/// 'p: 'tx, means that Platform must outlive the transaction
+pub struct ConsensusAbciApplication<'p, C> {
     /// Platform
-    platform: &'a Platform<C>,
+    platform: &'p Platform<C>,
     /// The current GroveDb transaction
-    transaction: RwLock<Option<Transaction<'a>>>,
+    transaction: RwLock<Option<Transaction<'p>>>,
     /// The current block execution context
     block_execution_context: RwLock<Option<BlockExecutionContext>>,
+    /// The State sync session
+    snapshot_fetching_session: RwLock<Option<SnapshotFetchingSession<'p>>>,
+    /// The snapshot manager
+    snapshot_manager: SnapshotManager,
 }
 
-impl<'a, C> ConsensusAbciApplication<'a, C> {
+impl<'p, C> ConsensusAbciApplication<'p, C> {
     /// Create new ABCI app
-    pub fn new(platform: &'a Platform<C>) -> Self {
+    pub fn new(platform: &'p Platform<C>) -> Self {
+        let snapshot_manager = SnapshotManager::new(
+            platform.config.state_sync_config.checkpoints_path.clone(),
+            platform.config.state_sync_config.max_num_snapshots,
+            platform.config.state_sync_config.snapshots_frequency,
+        );
         Self {
             platform,
             transaction: Default::default(),
             block_execution_context: Default::default(),
+            snapshot_fetching_session: Default::default(),
+            snapshot_manager,
         }
     }
 }
 
-impl<'a, C> PlatformApplication<C> for ConsensusAbciApplication<'a, C> {
+impl<'p, C> PlatformApplication<C> for ConsensusAbciApplication<'p, C> {
     fn platform(&self) -> &Platform<C> {
         self.platform
     }
 }
 
-impl<'a, C> BlockExecutionApplication for ConsensusAbciApplication<'a, C> {
+impl<'p, C> SnapshotManagerApplication for ConsensusAbciApplication<'p, C> {
+    fn snapshot_manager(&self) -> &SnapshotManager {
+        &self.snapshot_manager
+    }
+}
+
+impl<'p, C> SnapshotFetchingApplication<'p, C> for ConsensusAbciApplication<'p, C> {
+    fn snapshot_fetching_session(&self) -> &RwLock<Option<SnapshotFetchingSession<'p>>> {
+        &self.snapshot_fetching_session
+    }
+
+    fn platform(&self) -> &'p Platform<C> {
+        self.platform
+    }
+}
+
+impl<'p, C> BlockExecutionApplication for ConsensusAbciApplication<'p, C> {
     fn block_execution_context(&self) -> &RwLock<Option<BlockExecutionContext>> {
         &self.block_execution_context
     }
 }
 
-impl<'a, C> TransactionalApplication<'a> for ConsensusAbciApplication<'a, C> {
+impl<'p, C> TransactionalApplication<'p> for ConsensusAbciApplication<'p, C> {
     /// create and store a new transaction
     fn start_transaction(&self) {
         let transaction = self.platform.drive.grove.start_transaction();
         self.transaction.write().unwrap().replace(transaction);
     }
 
-    fn transaction(&self) -> &RwLock<Option<Transaction<'a>>> {
+    fn transaction(&self) -> &RwLock<Option<Transaction<'p>>> {
         &self.transaction
     }
 
@@ -77,13 +110,13 @@ impl<'a, C> TransactionalApplication<'a> for ConsensusAbciApplication<'a, C> {
     }
 }
 
-impl<'a, C> Debug for ConsensusAbciApplication<'a, C> {
+impl<'p, C> Debug for ConsensusAbciApplication<'p, C> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "<ConsensusAbciApplication>")
     }
 }
 
-impl<'a, C> tenderdash_abci::Application for ConsensusAbciApplication<'a, C>
+impl<'p, C> tenderdash_abci::Application for ConsensusAbciApplication<'p, C>
 where
     C: CoreRPCLike,
 {
@@ -149,4 +182,25 @@ where
     ) -> Result<proto::ResponseVerifyVoteExtension, proto::ResponseException> {
         handler::verify_vote_extension(self, request).map_err(error_into_exception)
     }
+
+    fn offer_snapshot(
+        &self,
+        request: proto::RequestOfferSnapshot,
+    ) -> Result<proto::ResponseOfferSnapshot, proto::ResponseException> {
+        handler::offer_snapshot(self, request).map_err(error_into_exception)
+    }
+
+    fn apply_snapshot_chunk(
+        &self,
+        request: proto::RequestApplySnapshotChunk,
+    ) -> Result<proto::ResponseApplySnapshotChunk, proto::ResponseException> {
+        handler::apply_snapshot_chunk(self, request).map_err(error_into_exception)
+    }
+
+    fn finalize_snapshot(
+        &self,
+        request: proto::RequestFinalizeSnapshot,
+    ) -> Result<proto::ResponseFinalizeSnapshot, proto::ResponseException> {
+        handler::finalize_snapshot(self, request).map_err(error_into_exception)
+    }
 }