address feedback

consensus/src/dag/bootstrap.rs

@@ -6,11 +6,12 @@ use super::{
     dag_fetcher::{DagFetcherService, FetchRequestHandler},
     dag_handler::NetworkHandler,
     dag_network::TDAGNetworkSender,
+    dag_state_sync::DAG_WINDOW,
     dag_store::Dag,
     order_rule::OrderRule,
     rb_handler::NodeBroadcastHandler,
     storage::DAGStorage,
-    types::DAGMessage, dag_state_sync::DAG_WINDOW,
+    types::DAGMessage,
 };
 use crate::{
     dag::adapter::NotificationAdapter, experimental::buffer_manager::OrderedBlocks,

consensus/src/dag/dag_state_sync.rs

@@ -141,7 +141,10 @@ impl StateSyncManager {
 
         // Create a new DAG store and Fetch blocks
         let target_round = node.round();
-        let start_round = commit_li.commit_info().round().saturating_sub(DAG_WINDOW as Round);
+        let start_round = commit_li
+            .commit_info()
+            .round()
+            .saturating_sub(DAG_WINDOW as Round);
         let sync_dag_store = Arc::new(RwLock::new(Dag::new_empty(
             self.epoch_state.clone(),
             self.storage.clone(),

consensus/src/dag/dag_store.rs

@@ -78,11 +78,16 @@ impl Dag {
         if let Err(e) = storage.delete_certified_nodes(expired) {
             error!("Error deleting expired nodes: {:?}", e);
         }
+        let initial_round = if highest_committed_anchor_round <= dag_window_size_config as Round {
+            1
+        } else {
+            highest_committed_anchor_round.saturating_sub(dag_window_size_config as Round)
+        };
         Self {
             nodes_by_round,
             author_to_index,
             storage,
-            initial_round: highest_committed_anchor_round - (dag_window_size_config as Round),
+            initial_round,
             epoch_state,
             highest_committed_anchor_round,
         }

consensus/src/dag/tests/dag_driver_tests.rs

@@ -6,6 +6,7 @@ use crate::{
         dag_driver::{DagDriver, DagDriverError},
         dag_fetcher::DagFetcherService,
         dag_network::{RpcWithFallback, TDAGNetworkSender},
+        dag_state_sync::DAG_WINDOW,
         dag_store::Dag,
         order_rule::OrderRule,
         tests::{
@@ -78,8 +79,8 @@ async fn test_certified_node_handler() {
     let dag = Arc::new(RwLock::new(Dag::new(
         epoch_state.clone(),
         storage.clone(),
-        1,
         0,
+        DAG_WINDOW,
     )));
 
     let network_sender = Arc::new(MockNetworkSender {});

consensus/src/dag/tests/dag_test.rs

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::dag::{
+    dag_state_sync::DAG_WINDOW,
     dag_store::Dag,
     storage::DAGStorage,
     tests::helpers::new_certified_node,
@@ -106,7 +107,7 @@ fn setup() -> (Vec<ValidatorSigner>, Arc<EpochState>, Dag, Arc<MockStorage>) {
         verifier: validator_verifier,
     });
     let storage = Arc::new(MockStorage::new());
-    let dag = Dag::new(epoch_state.clone(), storage.clone(), 1, 0);
+    let dag = Dag::new(epoch_state.clone(), storage.clone(), 0, DAG_WINDOW);
     (signers, epoch_state, dag, storage)
 }
 
@@ -194,7 +195,7 @@ fn test_dag_recover_from_storage() {
             assert!(dag.add_node(node).is_ok());
         }
     }
-    let new_dag = Dag::new(epoch_state.clone(), storage.clone(), 1, 0);
+    let new_dag = Dag::new(epoch_state.clone(), storage.clone(), 0, DAG_WINDOW);
 
     for metadata in &metadatas {
         assert!(new_dag.exists(metadata));
@@ -205,7 +206,7 @@ fn test_dag_recover_from_storage() {
         verifier: epoch_state.verifier.clone(),
     });
 
-    let _new_epoch_dag = Dag::new(new_epoch_state, storage.clone(), 1, 0);
+    let _new_epoch_dag = Dag::new(new_epoch_state, storage.clone(), 0, DAG_WINDOW);
     assert!(storage.certified_node_data.lock().is_empty());
 }
 

consensus/src/dag/tests/fetcher_test.rs

@@ -3,6 +3,7 @@
 use super::dag_test::MockStorage;
 use crate::dag::{
     dag_fetcher::FetchRequestHandler,
+    dag_state_sync::DAG_WINDOW,
     dag_store::Dag,
     tests::helpers::new_certified_node,
     types::{DagSnapshotBitmask, FetchResponse, RemoteFetchRequest},
@@ -21,7 +22,12 @@ fn test_dag_fetcher_receiver() {
         verifier: validator_verifier,
     });
     let storage = Arc::new(MockStorage::new());
-    let dag = Arc::new(RwLock::new(Dag::new(epoch_state.clone(), storage, 1, 0)));
+    let dag = Arc::new(RwLock::new(Dag::new(
+        epoch_state.clone(),
+        storage,
+        0,
+        DAG_WINDOW,
+    )));
 
     let mut fetcher = FetchRequestHandler::new(dag.clone(), epoch_state);
 

consensus/src/dag/tests/order_rule_tests.rs

@@ -5,6 +5,7 @@ use crate::{
     dag::{
         adapter::Notifier,
         anchor_election::RoundRobinAnchorElection,
+        dag_state_sync::DAG_WINDOW,
         dag_store::Dag,
         order_rule::OrderRule,
         tests::{dag_test::MockStorage, helpers::new_certified_node},
@@ -186,7 +187,7 @@ proptest! {
             epoch: 1,
             verifier: validator_verifier,
         });
-        let mut dag = Dag::new(epoch_state.clone(), Arc::new(MockStorage::new()), 1, 0);
+        let mut dag = Dag::new(epoch_state.clone(), Arc::new(MockStorage::new()), 0, DAG_WINDOW);
         for round_nodes in &nodes {
             for node in round_nodes.iter().flatten() {
                 dag.add_node(node.clone()).unwrap();
@@ -273,7 +274,12 @@ fn test_order_rule_basic() {
         epoch: 1,
         verifier: validator_verifier,
     });
-    let mut dag = Dag::new(epoch_state.clone(), Arc::new(MockStorage::new()), 1, 0);
+    let mut dag = Dag::new(
+        epoch_state.clone(),
+        Arc::new(MockStorage::new()),
+        0,
+        DAG_WINDOW,
+    );
     for round_nodes in &nodes {
         for node in round_nodes.iter().flatten() {
             dag.add_node(node.clone()).unwrap();

consensus/src/dag/tests/rb_handler_tests.rs

@@ -3,6 +3,7 @@
 
 use crate::dag::{
     dag_fetcher::TFetchRequester,
+    dag_state_sync::DAG_WINDOW,
     dag_store::Dag,
     rb_handler::{NodeBroadcastHandleError, NodeBroadcastHandler},
     storage::DAGStorage,
@@ -43,8 +44,8 @@ async fn test_node_broadcast_receiver_succeed() {
     let dag = Arc::new(RwLock::new(Dag::new(
         epoch_state.clone(),
         storage.clone(),
-        1,
         0,
+        DAG_WINDOW,
     )));
 
     let wellformed_node = new_node(1, 10, signers[0].author(), vec![]);
@@ -87,8 +88,8 @@ async fn test_node_broadcast_receiver_failure() {
             let dag = Arc::new(RwLock::new(Dag::new(
                 epoch_state.clone(),
                 storage.clone(),
-                1,
                 0,
+                DAG_WINDOW,
             )));
 
             NodeBroadcastHandler::new(
@@ -163,8 +164,8 @@ fn test_node_broadcast_receiver_storage() {
     let dag = Arc::new(RwLock::new(Dag::new(
         epoch_state.clone(),
         storage.clone(),
-        1,
         0,
+        DAG_WINDOW,
     )));
 
     let node = new_node(1, 10, signers[0].author(), vec![]);