Merge pull request ProvableHQ#3239 from AleoHQ/feat/store_proposals

[HackerOne-2452182] Store `Proposal` and `SignedProposals` to disk and remove proposal expiration

Cargo.toml

@@ -46,7 +46,7 @@ default-features = false
 
 [workspace.dependencies.snarkvm]
 git = "https://github.com/AleoHQ/snarkVM.git"
-rev = "aef07da"
+rev = "57641ca"
 #version = "=0.16.18"
 features = [ "circuit", "console", "rocks" ]
 

cli/src/commands/clean.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use snarkos_node::bft::helpers::proposal_cache_path;
+
 use aleo_std::StorageMode;
 use anyhow::{bail, Result};
 use clap::Parser;
@@ -35,6 +37,13 @@ pub struct Clean {
 impl Clean {
     /// Cleans the snarkOS node storage.
     pub fn parse(self) -> Result<String> {
+        // Remove the current proposal cache file, if it exists.
+        let proposal_cache_path = proposal_cache_path(self.network, self.dev);
+        if proposal_cache_path.exists() {
+            if let Err(err) = std::fs::remove_file(&proposal_cache_path) {
+                bail!("Failed to remove the current proposal cache file at {}: {err}", proposal_cache_path.display());
+            }
+        }
         // Remove the specified ledger from storage.
         Self::remove_ledger(self.network, match self.path {
             Some(path) => StorageMode::Custom(path),

node/bft/src/gateway.rs

@@ -261,6 +261,11 @@ impl<N: Network> Gateway<N> {
         &self.account
     }
 
+    /// Returns the dev identifier of the node.
+    pub const fn dev(&self) -> Option<u16> {
+        self.dev
+    }
+
     /// Returns the IP address of this node.
     pub fn local_ip(&self) -> SocketAddr {
         self.tcp.listening_addr().expect("The TCP listener is not enabled")

node/bft/src/helpers/mod.rs

@@ -30,12 +30,18 @@ pub use pending::*;
 pub mod proposal;
 pub use proposal::*;
 
+pub mod proposal_cache;
+pub use proposal_cache::*;
+
 pub mod ready;
 pub use ready::*;
 
 pub mod resolver;
 pub use resolver::*;
 
+pub mod signed_proposals;
+pub use signed_proposals::*;
+
 pub mod storage;
 pub use storage::*;
 

node/bft/src/helpers/proposal.rs

@@ -22,12 +22,13 @@ use snarkvm::{
         committee::Committee,
         narwhal::{BatchCertificate, BatchHeader, Transmission, TransmissionID},
     },
-    prelude::{bail, ensure, Itertools, Result},
+    prelude::{bail, ensure, error, FromBytes, IoResult, Itertools, Read, Result, ToBytes, Write},
 };
 
 use indexmap::{IndexMap, IndexSet};
 use std::collections::HashSet;
 
+#[derive(Debug, PartialEq, Eq)]
 pub struct Proposal<N: Network> {
     /// The proposed batch header.
     batch_header: BatchHeader<N>,
@@ -167,6 +168,94 @@ impl<N: Network> Proposal<N> {
     }
 }
 
+impl<N: Network> ToBytes for Proposal<N> {
+    fn write_le<W: Write>(&self, mut writer: W) -> IoResult<()> {
+        // Write the batch header.
+        self.batch_header.write_le(&mut writer)?;
+        // Write the number of transmissions.
+        u32::try_from(self.transmissions.len()).map_err(error)?.write_le(&mut writer)?;
+        // Write the transmissions.
+        for (transmission_id, transmission) in &self.transmissions {
+            transmission_id.write_le(&mut writer)?;
+            transmission.write_le(&mut writer)?;
+        }
+        // Write the number of signatures.
+        u32::try_from(self.signatures.len()).map_err(error)?.write_le(&mut writer)?;
+        // Write the signatures.
+        for signature in &self.signatures {
+            signature.write_le(&mut writer)?;
+        }
+        Ok(())
+    }
+}
+
+impl<N: Network> FromBytes for Proposal<N> {
+    fn read_le<R: Read>(mut reader: R) -> IoResult<Self> {
+        // Read the batch header.
+        let batch_header = FromBytes::read_le(&mut reader)?;
+        // Read the number of transmissions.
+        let num_transmissions = u32::read_le(&mut reader)?;
+        // Ensure the number of transmissions is within bounds (this is an early safety check).
+        if num_transmissions as usize > BatchHeader::<N>::MAX_TRANSMISSIONS_PER_BATCH {
+            return Err(error("Invalid number of transmissions in the proposal"));
+        }
+        // Read the transmissions.
+        let mut transmissions = IndexMap::default();
+        for _ in 0..num_transmissions {
+            let transmission_id = FromBytes::read_le(&mut reader)?;
+            let transmission = FromBytes::read_le(&mut reader)?;
+            transmissions.insert(transmission_id, transmission);
+        }
+        // Read the number of signatures.
+        let num_signatures = u32::read_le(&mut reader)?;
+        // Ensure the number of signatures is within bounds (this is an early safety check).
+        if num_signatures as usize > Committee::<N>::MAX_COMMITTEE_SIZE as usize {
+            return Err(error("Invalid number of signatures in the proposal"));
+        }
+        // Read the signatures.
+        let mut signatures = IndexSet::default();
+        for _ in 0..num_signatures {
+            signatures.insert(FromBytes::read_le(&mut reader)?);
+        }
+
+        Ok(Self { batch_header, transmissions, signatures })
+    }
+}
+
+#[cfg(test)]
+pub(crate) mod tests {
+    use super::*;
+    use crate::helpers::storage::tests::sample_transmissions;
+    use snarkvm::{console::network::MainnetV0, utilities::TestRng};
+
+    type CurrentNetwork = MainnetV0;
+
+    const ITERATIONS: usize = 100;
+
+    pub(crate) fn sample_proposal(rng: &mut TestRng) -> Proposal<CurrentNetwork> {
+        let certificate = snarkvm::ledger::narwhal::batch_certificate::test_helpers::sample_batch_certificate(rng);
+        let (_, transmissions) = sample_transmissions(&certificate, rng);
+
+        let transmissions = transmissions.into_iter().map(|(id, (t, _))| (id, t)).collect::<IndexMap<_, _>>();
+        let batch_header = certificate.batch_header().clone();
+        let signatures = certificate.signatures().copied().collect();
+
+        Proposal { batch_header, transmissions, signatures }
+    }
+
+    #[test]
+    fn test_bytes() {
+        let rng = &mut TestRng::default();
+
+        for _ in 0..ITERATIONS {
+            let expected = sample_proposal(rng);
+            // Check the byte representation.
+            let expected_bytes = expected.to_bytes_le().unwrap();
+            assert_eq!(expected, Proposal::read_le(&expected_bytes[..]).unwrap());
+        }
+    }
+}
+
 #[cfg(test)]
 mod prop_tests {
     use crate::helpers::{

node/bft/src/helpers/proposal_cache.rs

@@ -0,0 +1,195 @@
+// Copyright (C) 2019-2023 Aleo Systems Inc.
+// This file is part of the snarkOS library.
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at:
+// http://www.apache.org/licenses/LICENSE-2.0
+
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::helpers::{Proposal, SignedProposals};
+
+use snarkvm::{
+    console::{account::Address, network::Network},
+    prelude::{anyhow, bail, FromBytes, IoResult, Read, Result, ToBytes, Write},
+};
+
+use aleo_std::{aleo_ledger_dir, StorageMode};
+use std::{fs, path::PathBuf};
+
+// Returns the path where a proposal cache file may be stored.
+pub fn proposal_cache_path(network: u16, dev: Option<u16>) -> PathBuf {
+    const PROPOSAL_CACHE_FILE_NAME: &str = "current-proposal-cache";
+
+    // Obtain the path to the ledger.
+    let mut path = aleo_ledger_dir(network, StorageMode::from(dev));
+    // Go to the folder right above the ledger.
+    path.pop();
+    // Append the proposal store's file name.
+    match dev {
+        Some(id) => path.push(&format!(".{PROPOSAL_CACHE_FILE_NAME}-{}-{}", network, id)),
+        None => path.push(&format!("{PROPOSAL_CACHE_FILE_NAME}-{}", network)),
+    }
+
+    path
+}
+
+/// A helper type for the cache of proposal and signed proposals.
+#[derive(Debug, PartialEq, Eq)]
+pub struct ProposalCache<N: Network> {
+    /// The latest round this node was on prior to the reboot.
+    latest_round: u64,
+    /// The latest proposal this node has created.
+    proposal: Option<Proposal<N>>,
+    /// The signed proposals this node has received.
+    signed_proposals: SignedProposals<N>,
+}
+
+impl<N: Network> ProposalCache<N> {
+    /// Initializes a new instance of the proposal cache.
+    pub fn new(latest_round: u64, proposal: Option<Proposal<N>>, signed_proposals: SignedProposals<N>) -> Self {
+        Self { latest_round, proposal, signed_proposals }
+    }
+
+    /// Ensure that the proposal and every signed proposal is associated with the `expected_signer`.
+    pub fn is_valid(&self, expected_signer: Address<N>) -> bool {
+        self.proposal
+            .as_ref()
+            .map(|proposal| {
+                proposal.batch_header().author() == expected_signer && self.latest_round == proposal.round()
+            })
+            .unwrap_or(true)
+            && self.signed_proposals.is_valid(expected_signer)
+    }
+
+    /// Returns `true` if a proposal cache exists for the given network and `dev`.
+    pub fn exists(dev: Option<u16>) -> bool {
+        proposal_cache_path(N::ID, dev).exists()
+    }
+
+    /// Load the proposal cache from the file system and ensure that the proposal cache is valid.
+    pub fn load(expected_signer: Address<N>, dev: Option<u16>) -> Result<Self> {
+        // Construct the proposal cache file system path.
+        let path = proposal_cache_path(N::ID, dev);
+
+        // Deserialize the proposal cache from the file system.
+        let proposal_cache = match fs::read(&path) {
+            Ok(bytes) => match Self::from_bytes_le(&bytes) {
+                Ok(proposal_cache) => proposal_cache,
+                Err(_) => bail!("Couldn't deserialize the proposal stored at {}", path.display()),
+            },
+            Err(_) => bail!("Couldn't read the proposal stored at {}", path.display()),
+        };
+
+        // Ensure the proposal cache is valid.
+        if !proposal_cache.is_valid(expected_signer) {
+            bail!("The proposal cache is invalid for the given address {expected_signer}");
+        }
+
+        info!("Loaded the proposal cache from {} at round {}", path.display(), proposal_cache.latest_round);
+
+        Ok(proposal_cache)
+    }
+
+    /// Store the proposal cache to the file system.
+    pub fn store(&self, dev: Option<u16>) -> Result<()> {
+        let path = proposal_cache_path(N::ID, dev);
+        info!("Storing the proposal cache to {}...", path.display());
+
+        // Serialize the proposal cache.
+        let bytes = self.to_bytes_le()?;
+        // Store the proposal cache to the file system.
+        fs::write(&path, bytes)
+            .map_err(|err| anyhow!("Couldn't write the proposal cache to {} - {err}", path.display()))?;
+
+        Ok(())
+    }
+
+    /// Returns the latest round, proposal and signed proposals.
+    pub fn into(self) -> (u64, Option<Proposal<N>>, SignedProposals<N>) {
+        (self.latest_round, self.proposal, self.signed_proposals)
+    }
+}
+
+impl<N: Network> ToBytes for ProposalCache<N> {
+    fn write_le<W: Write>(&self, mut writer: W) -> IoResult<()> {
+        // Serialize the `latest_round`.
+        self.latest_round.write_le(&mut writer)?;
+        // Serialize the `proposal`.
+        self.proposal.is_some().write_le(&mut writer)?;
+        if let Some(proposal) = &self.proposal {
+            proposal.write_le(&mut writer)?;
+        }
+        // Serialize the `signed_proposals`.
+        self.signed_proposals.write_le(&mut writer)?;
+
+        Ok(())
+    }
+}
+
+impl<N: Network> FromBytes for ProposalCache<N> {
+    fn read_le<R: Read>(mut reader: R) -> IoResult<Self> {
+        // Deserialize `latest_round`.
+        let latest_round = u64::read_le(&mut reader)?;
+        // Deserialize `proposal`.
+        let has_proposal: bool = FromBytes::read_le(&mut reader)?;
+        let proposal = match has_proposal {
+            true => Some(Proposal::read_le(&mut reader)?),
+            false => None,
+        };
+        // Deserialize `signed_proposals`.
+        let signed_proposals = SignedProposals::read_le(&mut reader)?;
+
+        Ok(Self::new(latest_round, proposal, signed_proposals))
+    }
+}
+
+impl<N: Network> Default for ProposalCache<N> {
+    /// Initializes a new instance of the proposal cache.
+    fn default() -> Self {
+        Self::new(0, None, Default::default())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::helpers::{proposal::tests::sample_proposal, signed_proposals::tests::sample_signed_proposals};
+    use snarkvm::{
+        console::{account::PrivateKey, network::MainnetV0},
+        utilities::TestRng,
+    };
+
+    type CurrentNetwork = MainnetV0;
+
+    const ITERATIONS: usize = 100;
+
+    pub(crate) fn sample_proposal_cache(
+        signer: &PrivateKey<CurrentNetwork>,
+        rng: &mut TestRng,
+    ) -> ProposalCache<CurrentNetwork> {
+        let proposal = sample_proposal(rng);
+        let signed_proposals = sample_signed_proposals(signer, rng);
+        let round = proposal.round();
+
+        ProposalCache::new(round, Some(proposal), signed_proposals)
+    }
+
+    #[test]
+    fn test_bytes() {
+        let rng = &mut TestRng::default();
+        let singer_private_key = PrivateKey::<CurrentNetwork>::new(rng).unwrap();
+
+        for _ in 0..ITERATIONS {
+            let expected = sample_proposal_cache(&singer_private_key, rng);
+            // Check the byte representation.
+            let expected_bytes = expected.to_bytes_le().unwrap();
+            assert_eq!(expected, ProposalCache::read_le(&expected_bytes[..]).unwrap());
+        }
+    }
+}