fix(bft/sync): pause bft block advancement during sync

Author: kaimast

Cargo.lock

@@ -48,7 +48,7 @@ default-features = false
 [workspace.dependencies.snarkvm]
 #path = "../snarkVM"
 git = "https://github.com/ProvableHQ/snarkVM.git"
-rev = "d8e736636dd"
+rev = "cf10c9df8"
 #version = "=4.4.0"
 default-features = false
 

Cargo.toml

@@ -98,21 +98,22 @@ fn check_locktick_imports<P: AsRef<Path>>(path: P) {
 
             // Modify the lock balance based on the type of the relevant import.
             if [ImportOfInterest::ParkingLot, ImportOfInterest::Tokio].contains(&ioi) {
-                if line.contains("Mutex") {
-                    lock_balance += 1;
-                }
-                if line.contains("RwLock") {
-                    lock_balance += 1;
+                lock_balance += line.matches("Mutex").count() as i8;
+                lock_balance += line.matches("RwLock").count() as i8;
+
+                // Correct for `tokio::sync::MutexGuard` not having a locktick counterpart.
+                if line.contains("MutexGuard") {
+                    lock_balance -= 1;
                 }
+
+                // Correction for `use tokio::Mutex as TMutex` and `use tokio::MutexGuard as TMutexGuard`.
+                lock_balance -= line.matches("TMutex").count() as i8;
             } else if ioi == ImportOfInterest::Locktick {
                 // Use `matches` instead of just `contains` here, as more than a single
                 // lock type entry is possible in a locktick import.
-                for _hit in line.matches("Mutex") {
-                    lock_balance -= 1;
-                }
-                for _hit in line.matches("RwLock") {
-                    lock_balance -= 1;
-                }
+                lock_balance -= line.matches("Mutex").count() as i8;
+                lock_balance -= line.matches("RwLock").count() as i8;
+
                 // A correction in case of the `use tokio::Mutex as TMutex` convention.
                 if line.contains("TMutex") {
                     lock_balance += 1;

build.rs

@@ -78,6 +78,10 @@ workspace = true
 features = [ "macros", "rt-multi-thread" ]
 optional = true
 
+[dependencies.thiserror]
+workspace = true
+
 [dependencies.tracing]
 workspace = true
 optional = true
+features = [ "log" ]

node/bft/ledger-service/Cargo.toml

@@ -13,13 +13,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::{LedgerService, fmt_id, spawn_blocking};
+use crate::{BeginLedgerUpdateError, LedgerService, LedgerUpdateService, fmt_id, spawn_blocking};
 
 use snarkos_utilities::Stoppable;
 
 use snarkvm::{
     ledger::{
         Block,
+        CheckBlockError,
         Ledger,
         PendingBlock,
         Transaction,
@@ -48,9 +49,13 @@ use snarkvm::{
 use anyhow::ensure;
 use indexmap::IndexMap;
 #[cfg(feature = "locktick")]
-use locktick::parking_lot::RwLock;
+use locktick::{
+    LockGuard,
+    parking_lot::{Mutex, RwLock},
+};
+use parking_lot::MutexGuard;
 #[cfg(not(feature = "locktick"))]
-use parking_lot::RwLock;
+use parking_lot::{Mutex, RwLock};
 #[cfg(not(feature = "serial"))]
 use rayon::prelude::*;
 
@@ -62,12 +67,73 @@ pub struct CoreLedgerService<N: Network, C: ConsensusStorage<N>> {
     ledger: Ledger<N, C>,
     latest_leader: Arc<RwLock<Option<(u64, Address<N>)>>>,
     stoppable: Arc<dyn Stoppable>,
+    update_lock: Arc<Mutex<()>>,
+}
+
+/// A transactional update to the ledger.
+#[cfg(feature = "ledger-write")]
+pub struct LedgerUpdate<'a, N: Network, C: ConsensusStorage<N>> {
+    ledger: Ledger<N, C>,
+    #[cfg(feature = "locktick")]
+    _lock: LockGuard<MutexGuard<'a, ()>>,
+    #[cfg(not(feature = "locktick"))]
+    _lock: MutexGuard<'a, ()>,
+}
+
+#[cfg(feature = "ledger-write")]
+impl<'a, N: Network, C: ConsensusStorage<N>> LedgerUpdateService<N> for LedgerUpdate<'a, N, C> {
+    fn check_block_subdag(
+        &self,
+        block: Block<N>,
+        prefix: &[PendingBlock<N>],
+    ) -> Result<PendingBlock<N>, CheckBlockError<N>> {
+        self.ledger.check_block_subdag(block, prefix)
+    }
+
+    fn check_block_content(&self, block: PendingBlock<N>) -> Result<Block<N>, CheckBlockError<N>> {
+        self.ledger.check_block_content(block, &mut rand::thread_rng())
+    }
+
+    /// Checks the given block is valid next block.
+    fn check_next_block(&self, block: &Block<N>) -> Result<()> {
+        self.ledger.check_next_block(block, &mut rand::thread_rng())
+    }
+
+    /// Returns a candidate for the next block in the ledger, using a committed subdag and its transmissions.
+    fn prepare_advance_to_next_quorum_block(
+        &self,
+        subdag: Subdag<N>,
+        transmissions: IndexMap<TransmissionID<N>, Transmission<N>>,
+    ) -> Result<Block<N>> {
+        self.ledger.prepare_advance_to_next_quorum_block(subdag, transmissions, &mut rand::thread_rng())
+    }
+
+    /// Adds the given block as the next block in the ledger.
+    fn advance_to_next_block(&self, block: &Block<N>) -> Result<()> {
+        // Advance to the next block.
+        self.ledger.advance_to_next_block(block)?;
+        // Update BFT metrics.
+        #[cfg(feature = "metrics")]
+        {
+            let num_sol = block.solutions().len();
+            let num_tx = block.transactions().len();
+
+            metrics::gauge(metrics::bft::HEIGHT, block.height() as f64);
+            metrics::gauge(metrics::bft::LAST_COMMITTED_ROUND, block.round() as f64);
+            metrics::increment_gauge(metrics::blocks::SOLUTIONS, num_sol as f64);
+            metrics::increment_gauge(metrics::blocks::TRANSACTIONS, num_tx as f64);
+            metrics::update_block_metrics(block);
+        }
+
+        tracing::info!("\n\nAdvanced to block {} at round {} - {}\n", block.height(), block.round(), block.hash());
+        Ok(())
+    }
 }
 
 impl<N: Network, C: ConsensusStorage<N>> CoreLedgerService<N, C> {
     /// Initializes a new core ledger service.
     pub fn new(ledger: Ledger<N, C>, stoppable: Arc<dyn Stoppable>) -> Self {
-        Self { ledger, latest_leader: Default::default(), stoppable }
+        Self { ledger, latest_leader: Default::default(), stoppable, update_lock: Default::default() }
     }
 }
 
@@ -338,53 +404,26 @@ impl<N: Network, C: ConsensusStorage<N>> LedgerService<N> for CoreLedgerService<
         spawn_blocking!(ledger.check_transaction_basic(&transaction, None, &mut rand::thread_rng()))
     }
 
-    fn check_block_subdag(&self, block: Block<N>, prefix: &[PendingBlock<N>]) -> Result<PendingBlock<N>> {
-        self.ledger.check_block_subdag(block, prefix)
-    }
-
-    fn check_block_content(&self, block: PendingBlock<N>) -> Result<Block<N>> {
-        self.ledger.check_block_content(block, &mut rand::thread_rng())
-    }
-
-    /// Checks the given block is valid next block.
-    fn check_next_block(&self, block: &Block<N>) -> Result<()> {
-        self.ledger.check_next_block(block, &mut rand::thread_rng())
-    }
-
-    /// Returns a candidate for the next block in the ledger, using a committed subdag and its transmissions.
-    #[cfg(feature = "ledger-write")]
-    fn prepare_advance_to_next_quorum_block(
+    fn check_block_subdag(
         &self,
-        subdag: Subdag<N>,
-        transmissions: IndexMap<TransmissionID<N>, Transmission<N>>,
-    ) -> Result<Block<N>> {
-        self.ledger.prepare_advance_to_next_quorum_block(subdag, transmissions, &mut rand::thread_rng())
+        block: Block<N>,
+        prefix: &[PendingBlock<N>],
+    ) -> Result<PendingBlock<N>, CheckBlockError<N>> {
+        self.ledger.check_block_subdag(block, prefix)
     }
 
-    /// Adds the given block as the next block in the ledger.
-    #[cfg(feature = "ledger-write")]
-    fn advance_to_next_block(&self, block: &Block<N>) -> Result<()> {
-        // If the Ctrl-C handler registered the signal, then skip advancing to the next block.
+    /// Begins a ledger update.i
+    ///
+    /// # Returns
+    /// - `Ok(Some(LedgerUpdate<N, C>))` if the ledger update was successfully started.
+    /// - `Ok(None)` if the node is stopped.
+    /// - `Err(anyhow::Error)` if we failed to acquire the update lock.
+    fn begin_ledger_update<'a>(&'a self) -> Result<Box<dyn LedgerUpdateService<N> + 'a>, BeginLedgerUpdateError> {
         if self.stoppable.is_stopped() {
-            bail!("Skipping advancing to block {} - The node is shutting down", block.height());
-        }
-        // Advance to the next block.
-        self.ledger.advance_to_next_block(block)?;
-        // Update BFT metrics.
-        #[cfg(feature = "metrics")]
-        {
-            let num_sol = block.solutions().len();
-            let num_tx = block.transactions().len();
-
-            metrics::gauge(metrics::bft::HEIGHT, block.height() as f64);
-            metrics::gauge(metrics::bft::LAST_COMMITTED_ROUND, block.round() as f64);
-            metrics::increment_gauge(metrics::blocks::SOLUTIONS, num_sol as f64);
-            metrics::increment_gauge(metrics::blocks::TRANSACTIONS, num_tx as f64);
-            metrics::update_block_metrics(block);
+            return Err(BeginLedgerUpdateError::ShuttingDown);
         }
 
-        tracing::info!("\n\nAdvanced to block {} at round {} - {}\n", block.height(), block.round(), block.hash());
-        Ok(())
+        Ok(Box::new(LedgerUpdate { ledger: self.ledger.clone(), _lock: self.update_lock.lock() }))
     }
 
     /// Returns the spend for a transaction in microcredits.

node/bft/ledger-service/src/ledger.rs

@@ -13,20 +13,20 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::{LedgerService, fmt_id};
+use crate::{BeginLedgerUpdateError, LedgerService, LedgerUpdateService, fmt_id};
 use snarkvm::{
     ledger::{
         Block,
+        CheckBlockError,
         PendingBlock,
         Transaction,
         committee::Committee,
-        narwhal::{BatchCertificate, Data, Subdag, Transmission, TransmissionID},
+        narwhal::{BatchCertificate, Data, Transmission, TransmissionID},
         puzzle::{Solution, SolutionID},
     },
-    prelude::{Address, ConsensusVersion, Field, Network, Result, Zero, bail, consensus_config_value, ensure},
+    prelude::{Address, ConsensusVersion, Field, Network, Result, Zero, bail, consensus_config_value},
 };
 
-use indexmap::IndexMap;
 #[cfg(feature = "locktick")]
 use locktick::parking_lot::Mutex;
 #[cfg(not(feature = "locktick"))]
@@ -213,42 +213,19 @@ impl<N: Network> LedgerService<N> for MockLedgerService<N> {
         Ok(())
     }
 
-    fn check_block_subdag(&self, _block: Block<N>, _prefix: &[PendingBlock<N>]) -> Result<PendingBlock<N>> {
+    fn check_block_subdag(
+        &self,
+        _block: Block<N>,
+        _prefix: &[PendingBlock<N>],
+    ) -> Result<PendingBlock<N>, CheckBlockError<N>> {
         unimplemented!();
     }
 
-    fn check_block_content(&self, _block: PendingBlock<N>) -> Result<Block<N>> {
+    /// Begins a ledger update.
+    fn begin_ledger_update<'a>(&'a self) -> Result<Box<dyn LedgerUpdateService<N> + 'a>, BeginLedgerUpdateError> {
         unimplemented!();
     }
 
-    /// Checks the given block is valid next block.
-    fn check_next_block(&self, _block: &Block<N>) -> Result<()> {
-        Ok(())
-    }
-
-    /// Returns a candidate for the next block in the ledger, using a committed subdag and its transmissions.
-    #[cfg(feature = "ledger-write")]
-    fn prepare_advance_to_next_quorum_block(
-        &self,
-        _subdag: Subdag<N>,
-        _transmissions: IndexMap<TransmissionID<N>, Transmission<N>>,
-    ) -> Result<Block<N>> {
-        unreachable!("MockLedgerService does not support prepare_advance_to_next_quorum_block")
-    }
-
-    /// Adds the given block as the next block in the ledger.
-    #[cfg(feature = "ledger-write")]
-    fn advance_to_next_block(&self, block: &Block<N>) -> Result<()> {
-        ensure!(
-            block.height() == self.latest_block_height() + 1,
-            "Tried to advance to block {} from block {}",
-            block.height(),
-            self.latest_block_height()
-        );
-        self.height_to_round_and_hash.lock().insert(block.height(), (block.round(), block.hash()));
-        Ok(())
-    }
-
     /// Returns the spent cost for a transaction in microcredits.
     fn transaction_spend_in_microcredits(
         &self,

node/bft/ledger-service/src/mock.rs

@@ -13,24 +13,24 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::LedgerService;
+use crate::{BeginLedgerUpdateError, LedgerService, LedgerUpdateService};
 use snarkvm::{
     ledger::{
         Block,
+        CheckBlockError,
         PendingBlock,
         Transaction,
         committee::Committee,
-        narwhal::{BatchCertificate, Data, Subdag, Transmission, TransmissionID},
+        narwhal::{BatchCertificate, Data, Transmission, TransmissionID},
         puzzle::{Solution, SolutionID},
     },
     prelude::{Address, ConsensusVersion, Field, Network, Result, Zero, bail},
 };
 
-use indexmap::IndexMap;
 use std::ops::Range;
 
 /// A ledger service for a prover.
-#[derive(Clone, Debug, Default)]
+#[derive(Clone, Default, Debug)]
 pub struct ProverLedgerService<N: Network> {
     _network: std::marker::PhantomData<N>,
 }
@@ -168,33 +168,18 @@ impl<N: Network> LedgerService<N> for ProverLedgerService<N> {
         Ok(())
     }
 
-    fn check_block_subdag(&self, _block: Block<N>, _prefix: &[PendingBlock<N>]) -> Result<PendingBlock<N>> {
-        bail!("Cannot check block subDAG in prover")
-    }
-
-    fn check_block_content(&self, _pending_block: PendingBlock<N>) -> Result<Block<N>> {
-        bail!("Cannot check block content in prover")
-    }
-
-    /// Checks the given block is valid next block.
-    fn check_next_block(&self, _block: &Block<N>) -> Result<()> {
-        Ok(())
-    }
-
-    /// Returns a candidate for the next block in the ledger, using a committed subdag and its transmissions.
-    #[cfg(feature = "ledger-write")]
-    fn prepare_advance_to_next_quorum_block(
+    fn check_block_subdag(
         &self,
-        _subdag: Subdag<N>,
-        _transmissions: IndexMap<TransmissionID<N>, Transmission<N>>,
-    ) -> Result<Block<N>> {
-        bail!("Cannot prepare advance to next quorum block in prover")
+        _block: Block<N>,
+        _prefix: &[PendingBlock<N>],
+    ) -> Result<PendingBlock<N>, CheckBlockError<N>> {
+        unimplemented!()
     }
 
     /// Adds the given block as the next block in the ledger.
     #[cfg(feature = "ledger-write")]
-    fn advance_to_next_block(&self, block: &Block<N>) -> Result<()> {
-        bail!("Cannot advance to next block in prover - {block}")
+    fn begin_ledger_update<'a>(&'a self) -> Result<Box<dyn LedgerUpdateService<N> + 'a>, BeginLedgerUpdateError> {
+        Err(BeginLedgerUpdateError::NotSupportedForProver)
     }
 
     /// Returns the spent cost for a transaction in microcredits.