feat(dan_layer/core): track checkpoint number for each checkpoint sub…

…mitted (#4268)

Description
---
- tracks the checkpoint number for a contract
- adds general-purpose metadata table to chain db
- adds `MetadataBackendAdapter` that specified interface for databases that support getting and setting metadata 

Motivation and Context
---
This PR adds functionality to persist the checkpoint number each time a checkpoint is submitted so that sequential checkpoints are submitted.

How Has This Been Tested?
---
Manually - VN submits sequential checkpoints

applications/tari_console_wallet/src/grpc/wallet_grpc_server.rs

@@ -884,7 +884,7 @@ impl wallet_server::Wallet for WalletGrpcServer {
             .await
             .map_err(|e| Status::internal(e.to_string()))?;
 
-        let message = format!("Sidechain state checkpoint for {}", contract_id);
+        let message = format!("Checkpoint #{} for {}", checkpoint_number, contract_id);
         transaction_service
             .submit_transaction(tx_id, transaction, 10.into(), message)
             .await

applications/tari_validator_node/src/grpc/services/wallet_client.rs

@@ -23,7 +23,6 @@
 use std::net::SocketAddr;
 
 use async_trait::async_trait;
-use log::*;
 use tari_app_grpc::{
     tari_rpc as grpc,
     tari_rpc::{
@@ -39,7 +38,7 @@ use tari_crypto::tari_utilities::ByteArray;
 use tari_dan_core::{services::WalletClient, DigitalAssetError};
 use tari_dan_engine::state::models::StateRoot;
 
-const LOG_TARGET: &str = "tari::dan::wallet_grpc";
+const _LOG_TARGET: &str = "tari::dan::wallet_grpc";
 
 type Inner = grpc::wallet_client::WalletClient<tonic::transport::Channel>;
 
@@ -80,8 +79,6 @@ impl WalletClient for GrpcWalletClient {
             signatures: checkpoint_signatures.into_iter().map(Into::into).collect(),
         };
 
-        info!(target: LOG_TARGET, "✅ Creating checkpoint #{}", checkpoint_number);
-
         if checkpoint_number == 0 {
             let request = CreateInitialAssetCheckpointRequest {
                 contract_id: contract_id.to_vec(),

dan_layer/core/src/services/checkpoint_manager.rs

@@ -33,6 +33,7 @@ const LOG_TARGET: &str = "tari::dan::checkpoint_manager";
 pub trait CheckpointManager {
     async fn create_checkpoint(
         &mut self,
+        checkpoint_number: u64,
         state_root: StateRoot,
         signature: Vec<SignerSignature>,
     ) -> Result<(), DigitalAssetError>;
@@ -42,17 +43,13 @@ pub trait CheckpointManager {
 pub struct ConcreteCheckpointManager<TWallet: WalletClient> {
     asset_definition: AssetDefinition,
     wallet: TWallet,
-    num_calls: u32,
-    checkpoint_interval: u32,
 }
 
 impl<TWallet: WalletClient> ConcreteCheckpointManager<TWallet> {
     pub fn new(asset_definition: AssetDefinition, wallet: TWallet) -> Self {
         Self {
             asset_definition,
             wallet,
-            num_calls: 0,
-            checkpoint_interval: 100,
         }
     }
 }
@@ -61,26 +58,23 @@ impl<TWallet: WalletClient> ConcreteCheckpointManager<TWallet> {
 impl<TWallet: WalletClient + Sync + Send> CheckpointManager for ConcreteCheckpointManager<TWallet> {
     async fn create_checkpoint(
         &mut self,
+        checkpoint_number: u64,
         state_root: StateRoot,
         signatures: Vec<SignerSignature>,
     ) -> Result<(), DigitalAssetError> {
-        if self.num_calls == 0 || self.num_calls >= self.checkpoint_interval {
-            // TODO: fetch and increment checkpoint number
-            let checkpoint_number = u64::from(self.num_calls / self.checkpoint_interval);
-            info!(
-                target: LOG_TARGET,
-                "Creating checkpoint for contract {}", self.asset_definition.contract_id
-            );
-            self.wallet
-                .create_new_checkpoint(
-                    &self.asset_definition.contract_id,
-                    &state_root,
-                    checkpoint_number,
-                    signatures,
-                )
-                .await?;
-        }
-        self.num_calls += 1;
+        info!(
+            target: LOG_TARGET,
+            "✅ Creating checkpoint #{} for contract {}", checkpoint_number, self.asset_definition.contract_id
+        );
+
+        self.wallet
+            .create_new_checkpoint(
+                &self.asset_definition.contract_id,
+                &state_root,
+                checkpoint_number,
+                signatures,
+            )
+            .await?;
         Ok(())
     }
 }

dan_layer/core/src/services/service_specification.rs

@@ -39,7 +39,13 @@ use crate::{
         SigningService,
         ValidatorNodeClientFactory,
     },
-    storage::{chain::ChainDbBackendAdapter, global::GlobalDbBackendAdapter, ChainStorageService, DbFactory},
+    storage::{
+        chain::{ChainDbBackendAdapter, ChainDbMetadataKey},
+        global::GlobalDbBackendAdapter,
+        ChainStorageService,
+        DbFactory,
+        MetadataBackendAdapter,
+    },
 };
 
 /// A trait to describe a specific configuration of services. This type allows other services to
@@ -50,7 +56,7 @@ pub trait ServiceSpecification: Default + Clone {
     type AssetProcessor: AssetProcessor + Clone;
     type AssetProxy: AssetProxy + Clone;
     type BaseNodeClient: BaseNodeClient + Clone;
-    type ChainDbBackendAdapter: ChainDbBackendAdapter;
+    type ChainDbBackendAdapter: ChainDbBackendAdapter + MetadataBackendAdapter<ChainDbMetadataKey>;
     type ChainStorageService: ChainStorageService<Self::Payload>;
     type CheckpointManager: CheckpointManager;
     type CommitteeManager: CommitteeManager<Self::Addr>;

dan_layer/core/src/storage/atomic_db.rs

@@ -0,0 +1,32 @@
+//  Copyright 2022. The Tari Project
+//
+//  Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+//  following conditions are met:
+//
+//  1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+//  disclaimer.
+//
+//  2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
+//  following disclaimer in the documentation and/or other materials provided with the distribution.
+//
+//  3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
+//  products derived from this software without specific prior written permission.
+//
+//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+//  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+//  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+//  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+//  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+//  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+//  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+use crate::storage::StorageError;
+
+pub trait AtomicDb {
+    type Error: Into<StorageError>;
+    type DbTransaction;
+
+    fn create_transaction(&self) -> Result<Self::DbTransaction, Self::Error>;
+
+    fn commit(&self, transaction: &Self::DbTransaction) -> Result<(), Self::Error>;
+}

dan_layer/core/src/storage/chain/chain_db.rs

@@ -24,12 +24,13 @@
 use crate::{
     models::{Node, QuorumCertificate, SideChainBlock, TreeNodeHash},
     storage::{
-        chain::{chain_db_unit_of_work::ChainDbUnitOfWorkImpl, ChainDbBackendAdapter},
+        chain::{chain_db_unit_of_work::ChainDbUnitOfWorkImpl, ChainDbBackendAdapter, ChainDbMetadataKey},
+        MetadataBackendAdapter,
         StorageError,
     },
 };
 
-pub struct ChainDb<TBackendAdapter: ChainDbBackendAdapter> {
+pub struct ChainDb<TBackendAdapter> {
     adapter: TBackendAdapter,
 }
 
@@ -107,6 +108,44 @@ impl<TBackendAdapter: ChainDbBackendAdapter> ChainDb<TBackendAdapter> {
     }
 }
 
+impl<TBackendAdapter> ChainDb<TBackendAdapter>
+where TBackendAdapter: MetadataBackendAdapter<ChainDbMetadataKey>
+{
+    pub fn get_current_checkpoint_number(&self) -> Result<u64, StorageError> {
+        let tx = self
+            .adapter
+            .create_transaction()
+            .map_err(TBackendAdapter::Error::into)?;
+        let number = self
+            .adapter
+            .get_metadata(&ChainDbMetadataKey::CheckpointNumber, &tx)
+            .map_err(TBackendAdapter::Error::into)?
+            .unwrap_or(0);
+        Ok(number)
+    }
+
+    /// Increments checkpoint number and returns the incremented value. If the key did not previously exist, it
+    /// is created and set to 1.
+    pub fn increment_checkpoint_number(&self) -> Result<u64, StorageError> {
+        let tx = self
+            .adapter
+            .create_transaction()
+            .map_err(TBackendAdapter::Error::into)?;
+        const KEY: ChainDbMetadataKey = ChainDbMetadataKey::CheckpointNumber;
+        let n = self
+            .adapter
+            .get_metadata::<u64>(&KEY, &tx)
+            .map_err(TBackendAdapter::Error::into)?
+            .unwrap_or(0);
+        let next = n + 1;
+        self.adapter
+            .set_metadata(KEY, next, &tx)
+            .map_err(TBackendAdapter::Error::into)?;
+        self.adapter.commit(&tx).map_err(TBackendAdapter::Error::into)?;
+        Ok(next)
+    }
+}
+
 impl<TBackendAdapter: ChainDbBackendAdapter + Clone + Send + Sync> ChainDb<TBackendAdapter> {
     pub fn new_unit_of_work(&self) -> ChainDbUnitOfWorkImpl<TBackendAdapter> {
         ChainDbUnitOfWorkImpl::new(self.adapter.clone())

dan_layer/core/src/storage/chain/chain_db_backend_adapter.rs

@@ -26,33 +26,20 @@ use crate::{
     models::{Payload, QuorumCertificate, TreeNodeHash},
     storage::{
         chain::{DbInstruction, DbNode, DbQc},
-        StorageError,
+        AtomicDb,
     },
 };
 
-pub trait ChainDbBackendAdapter: Send + Sync + Clone {
-    type BackendTransaction;
-    type Error: Into<StorageError>;
+pub trait ChainDbBackendAdapter: AtomicDb + Send + Sync + Clone {
     type Id: Copy + Send + Sync + Debug + PartialEq;
     type Payload: Payload;
 
     fn is_empty(&self) -> Result<bool, Self::Error>;
-    fn create_transaction(&self) -> Result<Self::BackendTransaction, Self::Error>;
     fn node_exists(&self, node_hash: &TreeNodeHash) -> Result<bool, Self::Error>;
     fn get_tip_node(&self) -> Result<Option<DbNode>, Self::Error>;
-    fn insert_node(&self, item: &DbNode, transaction: &Self::BackendTransaction) -> Result<(), Self::Error>;
-    fn update_node(
-        &self,
-        id: &Self::Id,
-        item: &DbNode,
-        transaction: &Self::BackendTransaction,
-    ) -> Result<(), Self::Error>;
-    fn insert_instruction(
-        &self,
-        item: &DbInstruction,
-        transaction: &Self::BackendTransaction,
-    ) -> Result<(), Self::Error>;
-    fn commit(&self, transaction: &Self::BackendTransaction) -> Result<(), Self::Error>;
+    fn insert_node(&self, item: &DbNode, transaction: &Self::DbTransaction) -> Result<(), Self::Error>;
+    fn update_node(&self, id: &Self::Id, item: &DbNode, transaction: &Self::DbTransaction) -> Result<(), Self::Error>;
+    fn insert_instruction(&self, item: &DbInstruction, transaction: &Self::DbTransaction) -> Result<(), Self::Error>;
     fn locked_qc_id(&self) -> Self::Id;
     fn prepare_qc_id(&self) -> Self::Id;
     fn find_highest_prepared_qc(&self) -> Result<QuorumCertificate, Self::Error>;
@@ -61,6 +48,6 @@ pub trait ChainDbBackendAdapter: Send + Sync + Clone {
     fn find_node_by_hash(&self, node_hash: &TreeNodeHash) -> Result<Option<(Self::Id, DbNode)>, Self::Error>;
     fn find_node_by_parent_hash(&self, parent_hash: &TreeNodeHash) -> Result<Option<(Self::Id, DbNode)>, Self::Error>;
     fn find_all_instructions_by_node(&self, node_id: Self::Id) -> Result<Vec<DbInstruction>, Self::Error>;
-    fn update_prepare_qc(&self, item: &DbQc, transaction: &Self::BackendTransaction) -> Result<(), Self::Error>;
-    fn update_locked_qc(&self, locked_qc: &DbQc, transaction: &Self::BackendTransaction) -> Result<(), Self::Error>;
+    fn update_prepare_qc(&self, item: &DbQc, transaction: &Self::DbTransaction) -> Result<(), Self::Error>;
+    fn update_locked_qc(&self, locked_qc: &DbQc, transaction: &Self::DbTransaction) -> Result<(), Self::Error>;
 }

dan_layer/core/src/storage/chain/metadata_key.rs

@@ -0,0 +1,44 @@
+//  Copyright 2022. The Tari Project
+//
+//  Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+//  following conditions are met:
+//
+//  1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+//  disclaimer.
+//
+//  2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
+//  following disclaimer in the documentation and/or other materials provided with the distribution.
+//
+//  3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
+//  products derived from this software without specific prior written permission.
+//
+//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+//  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+//  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+//  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+//  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+//  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+//  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+use std::fmt::{Display, Formatter};
+
+use crate::storage::metadata_backend_adapter::AsKeyBytes;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ChainDbMetadataKey {
+    CheckpointNumber,
+}
+
+impl AsKeyBytes for ChainDbMetadataKey {
+    fn as_key_bytes(&self) -> &[u8] {
+        match self {
+            ChainDbMetadataKey::CheckpointNumber => b"checkpoint-number",
+        }
+    }
+}
+
+impl Display for ChainDbMetadataKey {
+    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", String::from_utf8_lossy(self.as_key_bytes()))
+    }
+}

dan_layer/core/src/storage/chain/mod.rs

@@ -26,9 +26,12 @@ mod chain_db_unit_of_work;
 mod db_instruction;
 mod db_node;
 mod db_qc;
+mod metadata_key;
+
 pub use chain_db::ChainDb;
 pub use chain_db_backend_adapter::ChainDbBackendAdapter;
 pub use chain_db_unit_of_work::ChainDbUnitOfWork;
 pub use db_instruction::DbInstruction;
 pub use db_node::DbNode;
 pub use db_qc::DbQc;
+pub use metadata_key::ChainDbMetadataKey;

dan_layer/core/src/storage/metadata_backend_adapter.rs

@@ -0,0 +1,47 @@
+//  Copyright 2022. The Tari Project
+//
+//  Redistribution and use in source and binary forms, with or without modification, are permitted provided that the
+//  following conditions are met:
+//
+//  1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following
+//  disclaimer.
+//
+//  2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the
+//  following disclaimer in the documentation and/or other materials provided with the distribution.
+//
+//  3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote
+//  products derived from this software without specific prior written permission.
+//
+//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES,
+//  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+//  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+//  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+//  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+//  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
+//  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+// TODO: probably want to use something like bors or consensus encoding
+use tari_utilities::message_format::MessageFormat;
+
+use crate::storage::AtomicDb;
+
+pub trait MetadataBackendAdapter<K: AsKeyBytes>: AtomicDb + Send + Sync + Clone {
+    fn get_metadata<T: MessageFormat>(
+        &self,
+        key: &K,
+        transaction: &Self::DbTransaction,
+    ) -> Result<Option<T>, Self::Error>;
+
+    fn set_metadata<T: MessageFormat>(
+        &self,
+        key: K,
+        value: T,
+        transaction: &Self::DbTransaction,
+    ) -> Result<(), Self::Error>;
+
+    fn metadata_key_exists(&self, key: &K, transaction: &Self::DbTransaction) -> Result<bool, Self::Error>;
+}
+
+pub trait AsKeyBytes {
+    fn as_key_bytes(&self) -> &[u8];
+}