[consensus][dag] dag integration helpers

consensus/src/experimental/ordering_state_computer.rs

@@ -19,11 +19,13 @@ use aptos_crypto::HashValue;
 use aptos_executor_types::{ExecutorError, ExecutorResult, StateComputeResult};
 use aptos_logger::prelude::*;
 use aptos_types::{epoch_state::EpochState, ledger_info::LedgerInfoWithSignatures};
+use async_trait::async_trait;
 use fail::fail_point;
 use futures::{
     channel::{mpsc::UnboundedSender, oneshot},
     SinkExt,
 };
+use futures_channel::mpsc::unbounded;
 use std::sync::Arc;
 
 /// Ordering-only execution proxy
@@ -133,3 +135,73 @@ impl StateComputer for OrderingStateComputer {
 
     fn end_epoch(&self) {}
 }
+
+// TODO: stop using state computer for DAG state sync
+pub struct DagStateSyncComputer {
+    ordering_state_computer: OrderingStateComputer,
+}
+
+impl DagStateSyncComputer {
+    pub fn new(
+        state_computer_for_sync: Arc<dyn StateComputer>,
+        reset_event_channel_tx: UnboundedSender<ResetRequest>,
+    ) -> Self {
+        // note: this channel is unused
+        let (sender_tx, _) = unbounded();
+        Self {
+            ordering_state_computer: OrderingStateComputer {
+                executor_channel: sender_tx,
+                state_computer_for_sync,
+                reset_event_channel_tx,
+            },
+        }
+    }
+}
+
+#[async_trait]
+impl StateComputer for DagStateSyncComputer {
+    async fn compute(
+        &self,
+        // The block that will be computed.
+        _block: &Block,
+        // The parent block root hash.
+        _parent_block_id: HashValue,
+    ) -> ExecutorResult<StateComputeResult> {
+        unimplemented!("method not supported")
+    }
+
+    /// Send a successful commit. A future is fulfilled when the state is finalized.
+    async fn commit(
+        &self,
+        _blocks: &[Arc<ExecutedBlock>],
+        _finality_proof: LedgerInfoWithSignatures,
+        _callback: StateComputerCommitCallBackType,
+    ) -> ExecutorResult<()> {
+        unimplemented!("method not supported")
+    }
+
+    /// Best effort state synchronization to the given target LedgerInfo.
+    /// In case of success (`Result::Ok`) the LI of storage is at the given target.
+    /// In case of failure (`Result::Error`) the LI of storage remains unchanged, and the validator
+    /// can assume there were no modifications to the storage made.
+    async fn sync_to(&self, target: LedgerInfoWithSignatures) -> Result<(), StateSyncError> {
+        self.ordering_state_computer.sync_to(target).await
+    }
+
+    // Reconfigure to execute transactions for a new epoch.
+    fn new_epoch(
+        &self,
+        _epoch_state: &EpochState,
+        _payload_manager: Arc<PayloadManager>,
+        _transaction_shuffler: Arc<dyn TransactionShuffler>,
+        _block_gas_limit: Option<u64>,
+        _transaction_deduper: Arc<dyn TransactionDeduper>,
+    ) {
+        unimplemented!("method not supported");
+    }
+
+    // Reconfigure to clear epoch state at end of epoch.
+    fn end_epoch(&self) {
+        unimplemented!("method not supported")
+    }
+}

consensus/src/persistent_liveness_storage.rs

@@ -51,6 +51,9 @@ pub trait PersistentLivenessStorage: Send + Sync {
 
     /// Returns a handle of the aptosdb.
     fn aptos_db(&self) -> Arc<dyn DbReader>;
+
+    // Returns a handle of the consensus db
+    fn consensus_db(&self) -> Arc<ConsensusDB>;
 }
 
 #[derive(Clone)]
@@ -444,4 +447,8 @@ impl PersistentLivenessStorage for StorageWriteProxy {
     fn aptos_db(&self) -> Arc<dyn DbReader> {
         self.aptos_db.clone()
     }
+
+    fn consensus_db(&self) -> Arc<ConsensusDB> {
+        self.db.clone()
+    }    
 }

consensus/src/test_utils/mock_storage.rs

@@ -233,6 +233,10 @@ impl PersistentLivenessStorage for MockStorage {
     fn aptos_db(&self) -> Arc<dyn DbReader> {
         unimplemented!()
     }
+
+    fn consensus_db(&self) -> Arc<crate::consensusdb::ConsensusDB> {
+        unimplemented!()
+    }    
 }
 
 /// A storage that ignores any requests, used in the tests that don't care about the storage.
@@ -301,4 +305,8 @@ impl PersistentLivenessStorage for EmptyStorage {
     fn aptos_db(&self) -> Arc<dyn DbReader> {
         unimplemented!()
     }
+
+    fn consensus_db(&self) -> Arc<crate::consensusdb::ConsensusDB> {
+        unimplemented!()
+    }    
 }