Merge pull request #164 from alpenlabs/90-finalized-blocks

duty_executor: figure out which blocks are newly finalized

Author: delbonis

Cargo.lock

@@ -58,6 +58,7 @@ impl BlockSigningDuty {
 pub struct DutyTracker {
     next_id: u64,
     duties: Vec<DutyEntry>,
+    finalized_block: Option<L2BlockId>,
 }
 
 impl DutyTracker {
@@ -66,6 +67,7 @@ impl DutyTracker {
         Self {
             next_id: 1,
             duties: Vec::new(),
+            finalized_block: None,
         }
     }
 
@@ -78,6 +80,10 @@ impl DutyTracker {
     pub fn update(&mut self, update: &StateUpdate) -> usize {
         let mut kept_duties = Vec::new();
 
+        if update.latest_finalized_block.is_some() {
+            self.set_finalized_block(update.latest_finalized_block);
+        }
+
         for d in self.duties.drain(..) {
             match d.duty.expiry() {
                 Expiry::NextBlock => {
@@ -120,6 +126,14 @@ impl DutyTracker {
         }));
     }
 
+    pub fn set_finalized_block(&mut self, blkid: Option<L2BlockId>) {
+        self.finalized_block = blkid;
+    }
+
+    pub fn get_finalized_block(&self) -> Option<L2BlockId> {
+        self.finalized_block
+    }
+
     /// Returns the slice of duties we're keeping around.
     pub fn duties(&self) -> &[DutyEntry] {
         &self.duties
@@ -162,6 +176,9 @@ pub struct StateUpdate {
 
     /// Newly finalized blocks, must be sorted.
     newly_finalized_blocks: Vec<L2BlockId>,
+
+    /// Latest finalized block.
+    latest_finalized_block: Option<L2BlockId>,
 }
 
 impl StateUpdate {
@@ -170,11 +187,16 @@ impl StateUpdate {
         cur_timestamp: time::Instant,
         mut newly_finalized_blocks: Vec<L2BlockId>,
     ) -> Self {
+        // Extract latest finalized block before sorting
+        let latest_finalized_block = newly_finalized_blocks.first().cloned();
+
         newly_finalized_blocks.sort();
+
         Self {
             last_block_slot,
             cur_timestamp,
             newly_finalized_blocks,
+            latest_finalized_block,
         }
     }
 

crates/consensus-logic/src/duties.rs

@@ -54,9 +54,17 @@ pub fn duty_tracker_task<D: Database>(
     batch_queue: broadcast::Sender<DutyBatch>,
     ident: Identity,
     database: Arc<D>,
-) {
+) -> Result<(), Error> {
     let mut duties_tracker = duties::DutyTracker::new_empty();
 
+    let idx = database.client_state_provider().get_last_checkpoint_idx()?;
+    let last_checkpoint_state = database.client_state_provider().get_state_checkpoint(idx)?;
+    let last_finalized_blk = match last_checkpoint_state {
+        Some(state) => state.sync().map(|sync| *sync.finalized_blkid()),
+        None => None,
+    };
+    duties_tracker.set_finalized_block(last_finalized_blk);
+
     loop {
         let update = match cupdate_rx.blocking_recv() {
             Ok(u) => u,
@@ -87,6 +95,8 @@ pub fn duty_tracker_task<D: Database>(
     }
 
     info!("duty extractor task exiting");
+
+    Ok(())
 }
 
 fn update_tracker<D: Database>(
@@ -111,9 +121,15 @@ fn update_tracker<D: Database>(
     let block_idx = block.header().blockidx();
     let ts = time::Instant::now(); // FIXME XXX use .timestamp()!!!
 
-    // TODO figure out which blocks were finalized
-    let newly_finalized = Vec::new();
-    let tracker_update = duties::StateUpdate::new(block_idx, ts, newly_finalized);
+    // Figure out which blocks were finalized
+    let new_finalized = state.sync().map(|sync| *sync.finalized_blkid());
+    let newly_finalized_blocks: Vec<L2BlockId> = get_finalized_blocks(
+        tracker.get_finalized_block(),
+        l2prov.as_ref(),
+        new_finalized,
+    )?;
+
+    let tracker_update = duties::StateUpdate::new(block_idx, ts, newly_finalized_blocks);
     let n_evicted = tracker.update(&tracker_update);
     trace!(%n_evicted, "evicted old duties from new consensus state");
 
@@ -123,6 +139,35 @@ fn update_tracker<D: Database>(
     Ok(())
 }
 
+fn get_finalized_blocks(
+    last_finalized_block: Option<L2BlockId>,
+    l2prov: &impl L2DataProvider,
+    finalized: Option<L2BlockId>,
+) -> Result<Vec<L2BlockId>, Error> {
+    // Figure out which blocks were finalized
+    let mut newly_finalized_blocks: Vec<L2BlockId> = Vec::new();
+    let mut new_finalized = finalized;
+
+    while let Some(finalized) = new_finalized {
+        // If the last finalized block is equal to the new finalized block,
+        // it means that no new blocks are finalized
+        if last_finalized_block == Some(finalized) {
+            break;
+        }
+
+        // else loop till we reach to the last finalized block or go all the way
+        // as long as we get some block data
+        match l2prov.get_block_data(finalized)? {
+            Some(block) => new_finalized = Some(*block.header().parent()),
+            None => break,
+        }
+
+        newly_finalized_blocks.push(finalized);
+    }
+
+    Ok(newly_finalized_blocks)
+}
+
 pub fn duty_dispatch_task<
     D: Database + Sync + Send + 'static,
     E: ExecEngineCtl + Sync + Send + 'static,
@@ -377,3 +422,74 @@ fn poll_status_loop<E: ExecEngineCtl>(
 
     Ok(None)
 }
+
+#[cfg(test)]
+mod tests {
+    use alpen_express_db::traits::{Database, L2DataStore};
+    use alpen_express_state::header::L2Header;
+    use alpen_test_utils::l2::gen_l2_chain;
+
+    use super::get_finalized_blocks;
+
+    #[test]
+    fn test_get_finalized_blocks() {
+        let db = alpen_test_utils::get_common_db();
+        let chain = gen_l2_chain(None, 5);
+
+        for block in chain.clone() {
+            db.as_ref()
+                .l2_store()
+                .as_ref()
+                .put_block_data(block)
+                .unwrap();
+        }
+
+        let block_ids: Vec<_> = chain.iter().map(|b| b.header().get_blockid()).collect();
+
+        {
+            let last_finalized_block = Some(block_ids[0]);
+            let new_finalized = Some(block_ids[4]);
+            let finalized_blocks = get_finalized_blocks(
+                last_finalized_block,
+                db.l2_provider().as_ref(),
+                new_finalized,
+            )
+            .unwrap();
+
+            let expected_finalized_blocks: Vec<_> =
+                block_ids[1..=4].iter().rev().cloned().collect();
+
+            assert_eq!(finalized_blocks, expected_finalized_blocks);
+        }
+
+        {
+            let last_finalized_block = None;
+            let new_finalized = Some(block_ids[4]);
+            let finalized_blocks = get_finalized_blocks(
+                last_finalized_block,
+                db.l2_provider().as_ref(),
+                new_finalized,
+            )
+            .unwrap();
+
+            let expected_finalized_blocks: Vec<_> =
+                block_ids[0..=4].iter().rev().cloned().collect();
+
+            assert_eq!(finalized_blocks, expected_finalized_blocks);
+        }
+
+        {
+            let last_finalized_block = None;
+            let new_finalized = None;
+            let finalized_blocks = get_finalized_blocks(
+                last_finalized_block,
+                db.l2_provider().as_ref(),
+                new_finalized,
+            )
+            .unwrap();
+
+            let expected_finalized_blocks: Vec<_> = vec![];
+            assert_eq!(finalized_blocks, expected_finalized_blocks);
+        }
+    }
+}

crates/consensus-logic/src/duty_executor.rs

@@ -315,50 +315,11 @@ mod tests {
     use std::collections::HashSet;
 
     use alpen_express_db::traits::{Database, L2DataProvider, L2DataStore};
-    use alpen_express_state::{
-        block::{L2Block, L2BlockAccessory, L2BlockBody, L2BlockBundle},
-        header::{L2BlockHeader, L2Header, SignedL2BlockHeader},
-        id::L2BlockId,
-    };
-    use alpen_test_utils::ArbitraryGenerator;
+    use alpen_express_state::{header::L2Header, id::L2BlockId};
+    use alpen_test_utils::l2::gen_l2_chain;
 
     use crate::unfinalized_tracker;
 
-    fn get_genesis_block() -> L2BlockBundle {
-        let arb = ArbitraryGenerator::new();
-        let gen_header: SignedL2BlockHeader = arb.generate();
-        let body: L2BlockBody = arb.generate();
-        let accessory: L2BlockAccessory = arb.generate();
-
-        let empty_hash = L2BlockId::default();
-        let header = L2BlockHeader::new(
-            0,
-            gen_header.timestamp(),
-            empty_hash,
-            &body,
-            *gen_header.state_root(),
-        );
-        let signed_header = SignedL2BlockHeader::new(header, *gen_header.sig());
-        L2BlockBundle::new(L2Block::new(signed_header, body), accessory)
-    }
-
-    fn get_mock_block_with_parent(parent: &SignedL2BlockHeader) -> L2BlockBundle {
-        let arb = ArbitraryGenerator::new();
-        let gen_header: SignedL2BlockHeader = arb.generate();
-        let body: L2BlockBody = arb.generate();
-        let accessory: L2BlockAccessory = arb.generate();
-
-        let header = L2BlockHeader::new(
-            parent.blockidx() + 1,
-            gen_header.timestamp(),
-            parent.get_blockid(),
-            &body,
-            *gen_header.state_root(),
-        );
-        let signed_header = SignedL2BlockHeader::new(header, *gen_header.sig());
-        L2BlockBundle::new(L2Block::new(signed_header, body), accessory)
-    }
-
     fn setup_test_chain(l2_prov: &impl L2DataStore) -> [L2BlockId; 7] {
         // Chain A: g -> a1 -> a2 -> a3
         // Chain B: g -> a1 -> b2 -> b3
@@ -375,43 +336,27 @@ mod tests {
         // g
         // |
 
-        let genesis = get_genesis_block();
-        let genesis_header = genesis.header().clone();
-
-        let block_a1 = get_mock_block_with_parent(genesis.header());
-        let block_a1_header = block_a1.header().clone();
-
-        let block_c1 = get_mock_block_with_parent(genesis.header());
-        let block_c1_header = block_c1.header().clone();
-
-        let block_a2 = get_mock_block_with_parent(block_a1.header());
-        let block_a2_header = block_a2.header().clone();
-
-        let block_b2 = get_mock_block_with_parent(block_a1.header());
-        let block_b2_header = block_b2.header().clone();
-
-        let block_a3 = get_mock_block_with_parent(block_a2.header());
-        let block_a3_header = block_a3.header().clone();
-
-        let block_b3 = get_mock_block_with_parent(block_b2.header());
-        let block_b3_header = block_b3.header().clone();
-
-        l2_prov.put_block_data(genesis.clone()).unwrap();
-        l2_prov.put_block_data(block_a1.clone()).unwrap();
-        l2_prov.put_block_data(block_c1.clone()).unwrap();
-        l2_prov.put_block_data(block_a2.clone()).unwrap();
-        l2_prov.put_block_data(block_b2.clone()).unwrap();
-        l2_prov.put_block_data(block_a3.clone()).unwrap();
-        l2_prov.put_block_data(block_b3.clone()).unwrap();
+        let a_chain = gen_l2_chain(None, 3);
+        let b_chain = gen_l2_chain(Some(a_chain[1].header().clone()), 2);
+        let c_chain = gen_l2_chain(Some(a_chain[0].header().clone()), 1);
+
+        for b in a_chain
+            .clone()
+            .into_iter()
+            .chain(b_chain.clone())
+            .chain(c_chain.clone())
+        {
+            l2_prov.put_block_data(b).unwrap();
+        }
 
         [
-            genesis_header.get_blockid(),
-            block_a1_header.get_blockid(),
-            block_c1_header.get_blockid(),
-            block_a2_header.get_blockid(),
-            block_b2_header.get_blockid(),
-            block_a3_header.get_blockid(),
-            block_b3_header.get_blockid(),
+            a_chain[0].header().get_blockid(),
+            a_chain[1].header().get_blockid(),
+            c_chain[0].header().get_blockid(),
+            a_chain[2].header().get_blockid(),
+            b_chain[0].header().get_blockid(),
+            a_chain[3].header().get_blockid(),
+            b_chain[1].header().get_blockid(),
         ]
     }