BFT-474: Implement the new PersistentBatchStore methods for the inmem…

…ory version

node/libs/roles/src/attester/messages/batch.rs

@@ -8,13 +8,21 @@ use zksync_consensus_crypto::{keccak256::Keccak256, ByteFmt, Text, TextFmt};
 use zksync_consensus_utils::enum_util::Variant;
 
 /// A batch of L2 blocks used for the peers to fetch and keep in sync.
+///
+/// The use case for this is for peers to be able to catch up with L1
+/// batches they missed during periods of being offline. These will
+/// come with a proof of having been included on L1, rather than an
+/// attester quorum certificate.
 #[derive(Clone, Debug, PartialEq, Eq, Default)]
 pub struct SyncBatch {
     /// The number of the batch.
     pub number: BatchNumber,
     /// The payloads of the blocks the batch contains.
     pub payloads: Vec<Payload>,
     /// The proof of the batch.
+    ///
+    /// It is going to be a Merkle proof the the batch has been included
+    /// in the state hash of the L1.
     pub proof: Vec<u8>,
 }
 

node/libs/storage/Cargo.toml

@@ -9,6 +9,7 @@ license.workspace = true
 [dependencies]
 zksync_concurrency.workspace = true
 zksync_consensus_roles.workspace = true
+zksync_consensus_crypto.workspace = true
 zksync_protobuf.workspace = true
 
 anyhow.workspace = true

node/libs/storage/src/batch_store.rs

@@ -56,18 +56,42 @@ pub trait PersistentBatchStore: 'static + fmt::Debug + Send + Sync {
     /// Returns `None` if no batches have been created yet.
     async fn last_batch(&self, ctx: &ctx::Ctx) -> ctx::Result<Option<attester::BatchNumber>>;
 
+    /// Get the numbers of L1 batches which are missing the corresponding L1 batch quorum certificates
+    /// and potentially need to be signed by attesters.
+    ///
+    /// A replica might never have a complete history of L1 batch QCs; once the L1 batch is included on L1,
+    /// the replicas might use the [attester::SyncBatch] route to obtain them, in which case they will not
+    /// have a QC and no reason to get them either. The store will have sufficient information to decide
+    /// where it's still necessary to gossip votes; for example the main node will want to have a QC on
+    /// every batch while it's the one submitting them to L1, while replicas can ask the L1 what is considered
+    /// final.
+    async fn unsigned_batch_numbers(
+        &self,
+        ctx: &ctx::Ctx,
+    ) -> ctx::Result<Vec<attester::BatchNumber>>;
+
     /// Get the L1 batch QC from storage with the highest number.
     ///
     /// Returns `None` if we don't have a QC for any of the batches yet.
     async fn last_batch_qc(&self, ctx: &ctx::Ctx) -> ctx::Result<Option<attester::BatchQC>>;
 
-    /// Returns the batch with the given number.
+    /// Returns the [attester::SyncBatch] with the given number, which is used by peers
+    /// to catch up with L1 batches that they might have missed if they went offline.
     async fn get_batch(
         &self,
         ctx: &ctx::Ctx,
         number: attester::BatchNumber,
     ) -> ctx::Result<Option<attester::SyncBatch>>;
 
+    /// Returns the [attester::Batch] with the given number, which is the `message` that
+    /// appears in [attester::BatchQC], and represents the content that needs to be signed
+    /// by the attesters.
+    async fn get_batch_to_sign(
+        &self,
+        ctx: &ctx::Ctx,
+        number: attester::BatchNumber,
+    ) -> ctx::Result<Option<attester::Batch>>;
+
     /// Returns the QC of the batch with the given number.
     async fn get_batch_qc(
         &self,
@@ -240,18 +264,28 @@ impl BatchStore {
     /// before the earliest in these numbers that isn't signed, but it would be futile to sign them any more.
     pub async fn unsigned_batch_numbers(
         &self,
-        _ctx: &ctx::Ctx,
+        ctx: &ctx::Ctx,
     ) -> ctx::Result<Vec<attester::BatchNumber>> {
-        todo!("retrieve unsigned numbers from persistent store")
+        let unsigned = self
+            .persistent
+            .unsigned_batch_numbers(ctx)
+            .await
+            .context("persistent.get_batch_to_sign()")?;
+        Ok(unsigned)
     }
 
     /// Retrieve a batch to be signed.
     pub async fn batch_to_sign(
         &self,
-        _ctx: &ctx::Ctx,
-        _number: attester::BatchNumber,
+        ctx: &ctx::Ctx,
+        number: attester::BatchNumber,
     ) -> ctx::Result<Option<attester::Batch>> {
-        todo!("retrieve the batch commitment from persistent store")
+        let batch = self
+            .persistent
+            .get_batch_to_sign(ctx, number)
+            .await
+            .context("persistent.get_batch_to_sign()")?;
+        Ok(batch)
     }
 
     /// Append batch to a queue to be persisted eventually.

node/libs/storage/src/testonly/in_memory.rs

@@ -22,8 +22,7 @@ struct BlockStoreInner {
 struct BatchStoreInner {
     persisted: sync::watch::Sender<BatchStoreState>,
     batches: Mutex<VecDeque<attester::SyncBatch>>,
-    // This field was only added to be able to implement the PersistentBatchStore trait correctly.
-    qcs: Mutex<HashMap<attester::BatchNumber, attester::BatchQC>>,
+    certs: Mutex<HashMap<attester::BatchNumber, attester::BatchQC>>,
 }
 
 /// In-memory block store.
@@ -74,7 +73,7 @@ impl BatchStore {
         Self(Arc::new(BatchStoreInner {
             persisted: sync::watch::channel(BatchStoreState { first, last: None }).0,
             batches: Mutex::default(),
-            qcs: Mutex::default(),
+            certs: Mutex::default(),
         }))
     }
 }
@@ -137,22 +136,56 @@ impl PersistentBatchStore for BatchStore {
     }
 
     async fn last_batch_qc(&self, _ctx: &ctx::Ctx) -> ctx::Result<Option<attester::BatchQC>> {
-        let qcs = self.0.qcs.lock().unwrap();
-        let last_batch_number = qcs.keys().max().unwrap();
-        Ok(qcs.get(last_batch_number).cloned())
+        let certs = self.0.certs.lock().unwrap();
+        let last_batch_number = certs.keys().max().unwrap();
+        Ok(certs.get(last_batch_number).cloned())
+    }
+
+    async fn unsigned_batch_numbers(
+        &self,
+        _ctx: &ctx::Ctx,
+    ) -> ctx::Result<Vec<attester::BatchNumber>> {
+        let batches = self.0.batches.lock().unwrap();
+        let certs = self.0.certs.lock().unwrap();
+
+        Ok(batches
+            .iter()
+            .map(|b| b.number)
+            .filter(|n| !certs.contains_key(n))
+            .collect())
+    }
+
+    async fn get_batch_to_sign(
+        &self,
+        ctx: &ctx::Ctx,
+        number: attester::BatchNumber,
+    ) -> ctx::Result<Option<attester::Batch>> {
+        // Here we just produce some deterministic mock hash. The real hash is available in the database.
+        // and contains a commitment to the data submitted to L1. It is *not* over SyncBatch.
+        let Some(batch) = self.get_batch(ctx, number).await? else {
+            return Ok(None);
+        };
+
+        let bz = zksync_protobuf::canonical(&batch);
+        let hash = zksync_consensus_crypto::keccak256::Keccak256::new(&bz);
+
+        Ok(Some(attester::Batch {
+            number,
+            hash: attester::BatchHash(hash),
+        }))
     }
 
     async fn get_batch_qc(
         &self,
         _ctx: &ctx::Ctx,
         number: attester::BatchNumber,
     ) -> ctx::Result<Option<attester::BatchQC>> {
-        let qcs = self.0.qcs.lock().unwrap();
-        Ok(qcs.get(&number).cloned())
+        let certs = self.0.certs.lock().unwrap();
+        Ok(certs.get(&number).cloned())
     }
 
     async fn store_qc(&self, _ctx: &ctx::Ctx, qc: attester::BatchQC) -> ctx::Result<()> {
-        self.0.qcs.lock().unwrap().insert(qc.message.number, qc);
+        self.0.certs.lock().unwrap().insert(qc.message.number, qc);
         Ok(())
     }