Revert "Update shared-sequencer/movement-sequencer to match main"

This reverts commit 66a2072.

Author: 0xmovses

shared-sequencer/movement-sequencer/Cargo.toml

@@ -29,6 +29,7 @@ serde_json = "1.0.113" # https://github.com/serde-rs/json/releases
 serde_with = { version = "3.6.1", features = ["hex"] }
 tokio = { version = "1.36.0", features = ["fs", "rt-multi-thread"] }
 tonic = { version = "0.11.0", features = ["gzip"] }
+tempfile = "3.2.0"
 futures = "0.3.30" 
 
 # new deps

shared-sequencer/movement-sequencer/src/block/mod.rs

@@ -21,16 +21,33 @@ use serde_with::serde_as;
 
 #[serde_as]
 #[derive(Serialize, Deserialize, Clone, Derivative, Default)]
-#[derivative(Debug, PartialEq, Eq)]
+#[derivative(Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
 pub struct Transaction {
+    // the id of the rollup chain that the transaction is for
     #[serde_as(as = "Hex0xBytes")]
     pub consumer_id : Vec<u8>,
+    // the data of the transaction
     #[serde_as(as = "Hex0xBytes")]
     pub data : Vec<u8>, 
 }
 
 impl Transaction {
 
+    pub fn new(consumer_id: Vec<u8>, data: Vec<u8>) -> Self {
+        Self {
+            consumer_id,
+            data,
+        }
+    }
+
+    /// Creates a transaction for testing
+    pub fn test() -> Self {
+        Self {
+            consumer_id: vec![0, 1, 2, 3, 4],
+            data: vec![5, 6, 7, 8, 9],
+        }
+    }
+
     #[must_use]
     pub fn id(&self) -> ids::Id {
         let data_and_consumer_id = [self.data.clone(), self.consumer_id.clone()].concat();
@@ -72,6 +89,28 @@ pub struct Block {
 
 
 impl Block {
+
+    /// Creates a transaction for testing
+    pub fn test() -> Self {
+        let consumer_id = vec![0, 1, 2, 3, 4];
+        let data = vec![5, 6, 7, 8, 9];
+        let transactions = vec![Transaction::new(consumer_id, data)];
+        let parent_id = ids::Id::empty();
+        let height = 0;
+        let timestamp = Utc::now().timestamp() as u64;
+        let status = choices::status::Status::default();
+        Self {
+            parent_id,
+            height,
+            timestamp,
+            transactions,
+            status,
+            bytes: vec![],
+            id: ids::Id::empty(),
+            state: state::State::default(),
+        }
+    }
+
     /// Can fail if the block can't be serialized to JSON.
     /// # Errors
     /// Will fail if the block can't be serialized to JSON.

shared-sequencer/movement-sequencer/src/mempool/mod.rs

@@ -1,29 +1,84 @@
+use serde::{Deserialize, Serialize};
+
 use crate::block::{
     Transaction,
     Block
 };
+use std::cmp::Ordering;
 pub mod rocksdb;
 
 #[tonic::async_trait]
 pub trait MempoolTransactionOperations {
 
-    /// Checks whether a transaction exists in the mempool.
-    async fn has_transaction(&self, transaction_id : avalanche_types::ids::Id) -> Result<bool, anyhow::Error>;
+    // todo: move mempool_transaction methods into separate trait
+
+    /// Checks whether a mempool transaction exists in the mempool.
+    async fn has_mempool_transaction(&self, transaction_id : avalanche_types::ids::Id) -> Result<bool, anyhow::Error>;
+
+    /// Adds a mempool transaction to the mempool.
+    async fn add_mempool_transaction(&self, tx: MempoolTransaction) -> Result<(), anyhow::Error>;
+
+    /// Removes a mempool transaction from the mempool.
+    async fn remove_mempool_transaction(&self, transaction_id: avalanche_types::ids::Id) -> Result<(), anyhow::Error>;
+
+    /// Pops mempool transaction from the mempool.
+    async fn pop_mempool_transaction(&self) -> Result<Option<MempoolTransaction>, anyhow::Error>;
+ 
+    /// Gets a mempool transaction from the mempool.
+    async fn get_mempool_transaction(&self, transaction_id: avalanche_types::ids::Id) -> Result<Option<MempoolTransaction>, anyhow::Error>;
+
+    /// Pops the next n mempool transactions from the mempool.
+    async fn pop_mempool_transactions(&self, n: usize) -> Result<Vec<MempoolTransaction>, anyhow::Error> {
+        let mut mempool_transactions = Vec::new();
+        for _ in 0..n {
+            if let Some(mempool_transaction) = self.pop_mempool_transaction().await? {
+                mempool_transactions.push(mempool_transaction);
+            } else {
+                break;
+            }
+        }
+        Ok(mempool_transactions)
+    }
+
+    /// Checks whether the mempool has the transaction.
+    async fn has_transaction(&self, transaction_id: avalanche_types::ids::Id) -> Result<bool, anyhow::Error> {
+        self.has_mempool_transaction(transaction_id).await
+    }
 
     /// Adds a transaction to the mempool.
-    async fn add_transaction(&self, tx: Transaction) -> Result<(), anyhow::Error>;
+    async fn add_transaction(&self, tx: Transaction) -> Result<(), anyhow::Error> {
+
+        if self.has_transaction(tx.id()).await? {
+            return Ok(());
+        }
+
+        let mempool_transaction = MempoolTransaction::slot_now(tx);
+        self.add_mempool_transaction(mempool_transaction).await
+
+    }
 
     /// Removes a transaction from the mempool.
-    async fn remove_transaction(&self, transaction_id: avalanche_types::ids::Id) -> Result<(), anyhow::Error>;
+    async fn remove_transaction(&self, transaction_id: avalanche_types::ids::Id) -> Result<(), anyhow::Error> {
+        self.remove_mempool_transaction(transaction_id).await
+    }
 
     /// Pops transaction from the mempool.
-    async fn pop_transaction(&self, transaction_id: avalanche_types::ids::Id) -> Result<Transaction, anyhow::Error>;
- 
+    async fn pop_transaction(&self) -> Result<Option<Transaction>, anyhow::Error> {
+        let mempool_transaction = self.pop_mempool_transaction().await?;
+        Ok(mempool_transaction.map(|mempool_transaction| mempool_transaction.transaction))
+    }
+
     /// Gets a transaction from the mempool.
-    async fn get_transaction(&self, transaction_id: avalanche_types::ids::Id) -> Result<Transaction, anyhow::Error>;
+    async fn get_transaction(&self, transaction_id: avalanche_types::ids::Id) -> Result<Option<Transaction>, anyhow::Error> {
+        let mempool_transaction = self.get_mempool_transaction(transaction_id).await?;
+        Ok(mempool_transaction.map(|mempool_transaction| mempool_transaction.transaction))
+    }
 
-    /// Provides well-ordered transaction iterable
-    async fn iter(&self) -> Result<impl Iterator<Item = Transaction>, anyhow::Error>;
+    /// Pops the next n transactions from the mempool.
+    async fn pop_transactions(&self, n: usize) -> Result<Vec<Transaction>, anyhow::Error> {
+        let mempool_transactions = self.pop_mempool_transactions(n).await?;
+        Ok(mempool_transactions.into_iter().map(|mempool_transaction| mempool_transaction.transaction).collect())
+    }
 
 }
 
@@ -38,15 +93,91 @@ pub trait MempoolBlockOperations {
 
     /// Removes a block from the mempool.
     async fn remove_block(&self, block_id: avalanche_types::ids::Id) -> Result<(), anyhow::Error>;
-
-    /// Pops block from the mempool.
-    async fn pop_block(&self, block_id: avalanche_types::ids::Id) -> Result<Block, anyhow::Error>;
 
     /// Gets a block from the mempool.
-    async fn get_block(&self, block_id: avalanche_types::ids::Id) -> Result<Block, anyhow::Error>;
+    async fn get_block(&self, block_id: avalanche_types::ids::Id) -> Result<Option<Block>, anyhow::Error>;
+
+}
 
+/// Wraps a transaction with a timestamp for help ordering.
+#[derive(Debug, Clone, Hash, PartialEq, Eq, Serialize, Deserialize)]
+pub struct MempoolTransaction {
+    pub transaction: Transaction,
+    pub timestamp: u64,
+    pub slot_seconds : u64
 }
 
+impl PartialOrd for MempoolTransaction {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+/// Ordered first by slot_seconds, then by transaction.
+/// This allows us to use a BTreeSet to order transactions by slot_seconds, and then by transaction and pop them off in order.
+impl Ord for MempoolTransaction {
+    fn cmp(&self, other: &Self) -> Ordering {
+        // First, compare by slot_seconds
+        match self.slot_seconds.cmp(&other.slot_seconds) {
+            Ordering::Equal => {}
+            non_equal => return non_equal,
+        }
+        // If slot_seconds are equal, then compare by transaction
+        self.transaction.cmp(&other.transaction)
+    }
+}
+
+impl MempoolTransaction {
+
+    const SLOT_SECONDS : u64 = 2;
+
+    /// Creates a test MempoolTransaction.
+    pub fn test() -> Self {
+        Self {
+            transaction: Transaction::test(),
+            timestamp: 0,
+            slot_seconds: Self::SLOT_SECONDS
+        }
+    }
+
+    pub fn at_time(transaction: Transaction, timestamp: u64) -> Self {
+        let floor = (
+            timestamp / Self::SLOT_SECONDS
+        ) * Self::SLOT_SECONDS;
+        Self {
+            transaction,
+            timestamp : floor,
+            slot_seconds: Self::SLOT_SECONDS
+        }
+    }
+
+    pub fn new(transaction: Transaction, timestamp: u64, slot_seconds: u64) -> Self {
+        Self {
+            transaction,
+            timestamp,
+            slot_seconds
+        }
+    }
+
+    /// Creates a new MempoolTransaction with the current timestamp floored to the nearest slot.
+    /// todo: probably want to move this out to a factory.
+    pub fn slot_now(transaction : Transaction) -> MempoolTransaction {
+        
+        let timestamp = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs();
+    
+        Self::at_time(transaction, timestamp)
+        
+    }
+
+    pub fn id(&self) -> avalanche_types::ids::Id {
+        self.transaction.id()
+    }
+    
+
+}
+
+
+/// Combines RocksdbMempool with InMemoryMempool.
 #[derive(Debug, Clone)]
 pub struct Mempool {