fix: Handle historical bug where messages were not correctly revoked (#503)

Fix for processing blocks when
#484 was active.


Also supports future protocol version upgrades via EngineVersion.

Author: sanjayprabhu

src/consensus/proposer.rs

@@ -1,6 +1,5 @@
-use crate::core::types::{
-    proto, Address, Height, ShardHash, ShardId, SnapchainShard, FARCASTER_EPOCH,
-};
+use crate::core::types::{proto, Address, Height, ShardHash, ShardId, SnapchainShard};
+use crate::core::util::FarcasterTime;
 use crate::proto::{
     full_proposal, Block, BlockHeader, Commits, FullProposal, ShardChunk, ShardChunkWitness,
     ShardHeader, ShardWitness,
@@ -9,6 +8,7 @@ use crate::storage::store::engine::{BlockEngine, ShardEngine, ShardStateChange};
 use crate::storage::store::stores::Stores;
 use crate::storage::store::BlockStorageError;
 use crate::utils::statsd_wrapper::StatsdClientWrapper;
+use crate::version::version::EngineVersion;
 use informalsystems_malachitebft_core_types::{Round, Validity};
 use prost::Message;
 use std::collections::{BTreeMap, HashMap};
@@ -23,14 +23,6 @@ pub const PROTOCOL_VERSION: u32 = 1;
 pub const GENESIS_MESSAGE: &str =
     "It occurs to me that our survival may depend upon our talking to one another.";
 
-pub fn current_time() -> u64 {
-    std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)
-        .unwrap()
-        .as_secs()
-        - (FARCASTER_EPOCH / 1000)
-}
-
 #[allow(async_fn_in_trait)] // TODO
 pub trait Proposer {
     // Create a new block/shard chunk for the given height that will be proposed for confirmation to the other validators
@@ -167,12 +159,12 @@ impl Proposer for ShardProposer {
             None => vec![0, 32],
         };
 
-        let state_change = self
-            .engine
-            .propose_state_change(self.shard_id.shard_id(), messages);
+        let state_change =
+            self.engine
+                .propose_state_change(self.shard_id.shard_id(), messages, None);
         let shard_header = ShardHeader {
             parent_hash,
-            timestamp: current_time(),
+            timestamp: state_change.timestamp.into(),
             height: Some(height.clone()),
             shard_root: state_change.new_state_root.clone(),
         };
@@ -205,7 +197,11 @@ impl Proposer for ShardProposer {
             let height = header.height.unwrap();
             self.proposed_chunks
                 .add_proposed_value(full_proposal.clone());
-            let receive_delay = current_time().saturating_sub(header.timestamp);
+            let timestamp = FarcasterTime::new(header.timestamp);
+            let receive_delay = FarcasterTime::current()
+                .to_u64()
+                .saturating_sub(timestamp.to_u64());
+            let version = EngineVersion::version_for(&timestamp);
             self.statsd_client.gauge_with_shard(
                 self.shard_id.shard_id(),
                 "proposer.receive_delay",
@@ -225,6 +221,8 @@ impl Proposer for ShardProposer {
 
             let state = ShardStateChange {
                 shard_id: height.shard_index,
+                timestamp,
+                version,
                 new_state_root: header.shard_root.clone(),
                 transactions: chunk.transactions.clone(),
                 events: vec![],
@@ -469,7 +467,7 @@ impl Proposer for BlockProposer {
             parent_hash,
             chain_id: self.network as i32,
             version: PROTOCOL_VERSION,
-            timestamp: current_time(),
+            timestamp: FarcasterTime::current().into(),
             height: Some(height.clone()),
             shard_witnesses_hash: witness_hash,
         };

src/core/util.rs

@@ -1,6 +1,45 @@
 use crate::core::error::HubError;
 use crate::core::types::FARCASTER_EPOCH;
 
+#[derive(Clone, Debug)]
+pub struct FarcasterTime {
+    time: u64, // seconds since the farcaster epoch
+}
+
+impl FarcasterTime {
+    pub fn current() -> Self {
+        let time = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap()
+            .as_secs()
+            - (FARCASTER_EPOCH / 1000);
+        FarcasterTime { time }
+    }
+
+    pub fn new(time: u64) -> Self {
+        FarcasterTime { time }
+    }
+
+    pub fn from_unix_seconds(time: u64) -> Self {
+        let time = time - (FARCASTER_EPOCH / 1000);
+        FarcasterTime { time }
+    }
+
+    pub fn to_unix_seconds(&self) -> u64 {
+        self.time + (FARCASTER_EPOCH / 1000)
+    }
+
+    pub fn to_u64(&self) -> u64 {
+        self.time
+    }
+}
+
+impl Into<u64> for FarcasterTime {
+    fn into(self) -> u64 {
+        self.time
+    }
+}
+
 #[allow(dead_code)]
 pub fn to_farcaster_time(time_ms: u64) -> Result<u64, HubError> {
     if time_ms < FARCASTER_EPOCH {
@@ -26,10 +65,6 @@ pub fn from_farcaster_time(time: u64) -> u64 {
     time * 1000 + FARCASTER_EPOCH
 }
 
-pub fn farcaster_time_to_unix_seconds(time: u64) -> u64 {
-    time + (FARCASTER_EPOCH / 1000)
-}
-
 #[allow(dead_code)]
 pub fn get_farcaster_time() -> Result<u64, HubError> {
     let now = std::time::SystemTime::now()

src/lib.rs

@@ -9,6 +9,7 @@ pub mod node;
 pub mod perf;
 pub mod storage;
 pub mod utils;
+pub mod version;
 
 mod tests;
 

src/mempool/mempool.rs

@@ -11,7 +11,7 @@ use std::{
 use tokio::sync::{broadcast, mpsc, oneshot};
 
 use crate::core::error::HubError;
-use crate::core::util::farcaster_time_to_unix_seconds;
+use crate::core::util::FarcasterTime;
 use crate::proto::OnChainEventType;
 use crate::{
     core::types::SnapchainValidatorContext,
@@ -171,7 +171,7 @@ impl proto::Message {
             // TODO: Consider revisiting choice of timestamp here as backdated messages currently are prioritized.
             return MempoolKey::new(
                 MempoolMessageKind::UserMessage,
-                farcaster_time_to_unix_seconds(data.timestamp as u64),
+                FarcasterTime::new(data.timestamp as u64).to_unix_seconds(),
                 self.hex_hash(),
             );
         }

src/perf/engine_only_perftest.rs

@@ -99,7 +99,7 @@ pub async fn run() -> Result<(), Box<dyn Error>> {
         }
 
         let messages = engine.pull_messages(Duration::from_millis(50)).await?;
-        let state_change = engine.propose_state_change(1, messages);
+        let state_change = engine.propose_state_change(1, messages, None);
 
         let valid = engine.validate_state_change(&state_change);
         assert!(valid);

src/perf/perftest.rs

@@ -1,12 +1,13 @@
+use crate::core::util::FarcasterTime;
 use crate::perf::gen_single::SingleUser;
 use crate::perf::generate::{new_generator, GeneratorTypes};
 use crate::perf::{gen_single, generate};
 use crate::proto;
+use crate::proto::admin_service_client::AdminServiceClient;
 use crate::proto::hub_service_client::HubServiceClient;
 use crate::proto::Block;
 use crate::utils::cli::follow_blocks;
 use crate::utils::cli::send_on_chain_event;
-use crate::{consensus::proposer::current_time, proto::admin_service_client::AdminServiceClient};
 use clap::Parser;
 use figment::{
     providers::{Env, Format, Toml},
@@ -181,7 +182,7 @@ pub async fn run() -> Result<(), Box<dyn Error>> {
 
     let start = Instant::now();
     let mut stats_calculation_timer = time::interval(cfg.stats_calculation_interval);
-    let start_time = current_time();
+    let start_time = FarcasterTime::current().to_u64();
     let mut block_count = 0;
     let mut num_messages_confirmed = 0;
     let mut num_messages_submitted = 0;

src/storage/store/engine.rs

@@ -2,6 +2,7 @@ use super::account::UsernameProofStore;
 use super::account::{IntoU8, OnchainEventStorageError, UserDataStore};
 use crate::core::error::HubError;
 use crate::core::types::Height;
+use crate::core::util::FarcasterTime;
 use crate::core::validations;
 use crate::core::validations::verification;
 use crate::mempool::mempool::MempoolMessagesRequest;
@@ -17,6 +18,7 @@ use crate::storage::store::BlockStore;
 use crate::storage::trie;
 use crate::storage::trie::merkle_trie;
 use crate::utils::statsd_wrapper::StatsdClientWrapper;
+use crate::version::version::{EngineVersion, ProtocolFeature};
 use informalsystems_malachitebft_core_types::Round;
 use itertools::Itertools;
 use merkle_trie::TrieKey;
@@ -122,9 +124,11 @@ impl MempoolMessage {
 #[derive(Clone)]
 pub struct ShardStateChange {
     pub shard_id: u32,
+    pub timestamp: FarcasterTime,
     pub new_state_root: Vec<u8>,
     pub transactions: Vec<Transaction>,
     pub events: Vec<HubEvent>,
+    pub version: EngineVersion,
 }
 
 #[derive(Clone)]
@@ -279,18 +283,21 @@ impl ShardEngine {
         txn_batch: &mut RocksDbTransactionBatch,
         shard_id: u32,
         messages: Vec<MempoolMessage>,
+        timestamp: &FarcasterTime,
     ) -> Result<ShardStateChange, EngineError> {
         self.count("prepare_proposal.recv_messages", messages.len() as u64);
 
         let mut snapchain_txns = self.create_transactions_from_mempool(messages)?;
         let mut events = vec![];
         let mut validation_error_count = 0;
+        let version = EngineVersion::version_for(timestamp);
         for snapchain_txn in &mut snapchain_txns {
             let (account_root, txn_events, validation_errors) = self.replay_snapchain_txn(
                 trie_ctx,
                 &snapchain_txn,
                 txn_batch,
                 ProposalSource::Propose,
+                version,
             )?;
             snapchain_txn.account_root = account_root;
             events.extend(txn_events);
@@ -310,6 +317,8 @@ impl ShardEngine {
         let new_root_hash = self.stores.trie.root_hash()?;
         let result = ShardStateChange {
             shard_id,
+            timestamp: timestamp.clone(),
+            version,
             new_state_root: new_root_hash.clone(),
             transactions: snapchain_txns,
             events,
@@ -375,6 +384,7 @@ impl ShardEngine {
         &mut self,
         shard: u32,
         messages: Vec<MempoolMessage>,
+        timestamp: Option<FarcasterTime>,
     ) -> ShardStateChange {
         let now = std::time::Instant::now();
         let mut txn = RocksDbTransactionBatch::new();
@@ -386,12 +396,14 @@ impl ShardEngine {
             count_fn("trie.mem_get_count.total", read_count.1);
             count_fn("trie.mem_get_count.for_propose", read_count.1);
         };
+        let timestamp = timestamp.unwrap_or_else(FarcasterTime::current);
         let result = self
             .prepare_proposal(
                 &merkle_trie::Context::with_callback(count_callback),
                 &mut txn,
                 shard,
                 messages,
+                &timestamp,
             )
             .unwrap(); //TODO: don't unwrap()
 
@@ -470,6 +482,7 @@ impl ShardEngine {
         transactions: &[Transaction],
         shard_root: &[u8],
         source: ProposalSource,
+        version: EngineVersion,
     ) -> Result<Vec<HubEvent>, EngineError> {
         let now = std::time::Instant::now();
         let mut events = vec![];
@@ -500,8 +513,13 @@ impl ShardEngine {
         }
 
         for snapchain_txn in transactions {
-            let (account_root, txn_events, _) =
-                self.replay_snapchain_txn(trie_ctx, snapchain_txn, txn_batch, source.clone())?;
+            let (account_root, txn_events, _) = self.replay_snapchain_txn(
+                trie_ctx,
+                snapchain_txn,
+                txn_batch,
+                source.clone(),
+                version,
+            )?;
             // Reject early if account roots fail to match (shard roots will definitely fail)
             if &account_root != &snapchain_txn.account_root {
                 warn!(
@@ -544,6 +562,7 @@ impl ShardEngine {
         snapchain_txn: &Transaction,
         txn_batch: &mut RocksDbTransactionBatch,
         source: ProposalSource,
+        version: EngineVersion,
     ) -> Result<(Vec<u8>, Vec<HubEvent>, Vec<MessageValidationError>), EngineError> {
         let now = std::time::Instant::now();
         let total_user_messages = snapchain_txn.user_messages.len();
@@ -577,8 +596,7 @@ impl ShardEngine {
                         system_messages_count += 1;
                         match &onchain_event.body {
                             Some(proto::on_chain_event::Body::SignerEventBody(signer_event)) => {
-                                if signer_event.event_type == proto::SignerEventType::Remove as i32
-                                {
+                                if Self::should_revoke_signer(&signer_event, version) {
                                     revoked_signers.insert(signer_event.key.clone());
                                 }
                             }
@@ -1145,6 +1163,7 @@ impl ShardEngine {
             transactions,
             shard_root,
             ProposalSource::Validate,
+            shard_state_change.version,
         );
 
         match proposal_result {
@@ -1279,20 +1298,17 @@ impl ShardEngine {
                 "No valid cached transaction to apply. Replaying proposal"
             );
             // If we need to replay, reset the sequence number on the event id generator, just in case
-            let block_number = &shard_chunk
-                .header
-                .as_ref()
-                .unwrap()
-                .height
-                .unwrap()
-                .block_number;
-            self.stores.event_handler.set_current_height(*block_number);
+            let header = &shard_chunk.header.as_ref().unwrap();
+            let block_number = header.height.unwrap().block_number;
+            self.stores.event_handler.set_current_height(block_number);
+            let version = EngineVersion::version_for(&FarcasterTime::new(header.timestamp));
             match self.replay_proposal(
                 trie_ctx,
                 &mut txn,
                 transactions,
                 shard_root,
                 ProposalSource::Commit,
+                version,
             ) {
                 Err(err) => {
                     error!("State change commit failed: {}", err);
@@ -1316,11 +1332,13 @@ impl ShardEngine {
             system_messages: vec![],
             user_messages: vec![message.clone()],
         };
+        let version = EngineVersion::version_for(&FarcasterTime::current());
         let result = self.replay_snapchain_txn(
             &merkle_trie::Context::new(),
             &snapchain_txn,
             &mut txn,
             ProposalSource::Simulate,
+            version,
         );
 
         match result {
@@ -1490,6 +1508,14 @@ impl ShardEngine {
     pub fn trie_num_items(&mut self) -> usize {
         self.stores.trie.items().unwrap()
     }
+
+    fn should_revoke_signer(signer_event: &proto::SignerEventBody, version: EngineVersion) -> bool {
+        // When this bug was active, we did not revoke any signers, so, always return false
+        if version.is_enabled(ProtocolFeature::SignerRevokeBug) {
+            return false;
+        }
+        signer_event.event_type == proto::SignerEventType::Remove as i32
+    }
 }
 
 pub struct BlockEngine {