feat(ethexe-tx-pool): Introduce basic tx-pool (#4366)

Cargo.toml

@@ -302,6 +302,7 @@ ethexe-prometheus = { path = "ethexe/prometheus", default-features = false }
 ethexe-validator = { path = "ethexe/validator", default-features = false }
 ethexe-rpc = { path = "ethexe/rpc", default-features = false }
 ethexe-common = { path = "ethexe/common", default-features = false }
+ethexe-tx-pool = { path = "ethexe/tx-pool", default-features = false }
 ethexe-compute = { path = "ethexe/compute", default-features = false }
 
 # Common executor between `sandbox-host` and `lazy-pages-fuzzer`

core/src/lib.rs

@@ -43,6 +43,34 @@ pub mod program;
 pub mod reservation;
 pub mod str;
 pub mod tasks;
+pub mod utils {
+    //! Utility functions.
+
+    use blake2::{digest::typenum::U32, Blake2b, Digest};
+
+    /// BLAKE2b-256 hasher state.
+    type Blake2b256 = Blake2b<U32>;
+
+    /// Creates a unique identifier by passing given argument to blake2b hash-function.
+    ///
+    /// # SAFETY: DO NOT ADJUST HASH FUNCTION, BECAUSE MESSAGE ID IS SENSITIVE FOR IT.
+    pub fn hash(data: &[u8]) -> [u8; 32] {
+        let mut ctx = Blake2b256::new();
+        ctx.update(data);
+        ctx.finalize().into()
+    }
+
+    /// Creates a unique identifier by passing given argument to blake2b hash-function.
+    ///
+    /// # SAFETY: DO NOT ADJUST HASH FUNCTION, BECAUSE MESSAGE ID IS SENSITIVE FOR IT.
+    pub fn hash_of_array<T: AsRef<[u8]>, const N: usize>(array: [T; N]) -> [u8; 32] {
+        let mut ctx = Blake2b256::new();
+        for data in array {
+            ctx.update(data);
+        }
+        ctx.finalize().into()
+    }
+}
 
 // This allows all casts from u32 into usize be safe.
 const _: () = assert!(size_of::<u32>() <= size_of::<usize>());

ethexe/cli/src/commands/tx.rs

@@ -218,6 +218,7 @@ impl TxCommand {
 }
 
 // TODO (breathx): impl reply, value claim and exec balance top up with watch.
+// TODO (breathx) submit offchain txs
 /// Available transaction to submit.
 #[derive(Debug, Subcommand)]
 pub enum TxSubcommand {

ethexe/common/src/lib.rs

@@ -25,6 +25,7 @@ extern crate alloc;
 pub mod db;
 pub mod events;
 pub mod gear;
+pub mod tx_pool;
 
 pub use gear_core;
 pub use gprimitives;

ethexe/common/src/tx_pool.rs

@@ -0,0 +1,134 @@
+// This file is part of Gear.
+//
+// Copyright (C) 2025 Gear Technologies Inc.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+//! ethexe tx pool types
+
+use alloc::vec::Vec;
+use core::fmt;
+use gprimitives::{H160, H256};
+use parity_scale_codec::{Decode, Encode};
+
+/// Ethexe transaction with a signature.
+#[derive(Clone, Encode, Decode, PartialEq, Eq)]
+#[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
+pub struct SignedOffchainTransaction {
+    pub signature: Vec<u8>,
+    pub transaction: OffchainTransaction,
+}
+
+impl SignedOffchainTransaction {
+    /// Ethexe transaction blake2b256 hash.
+    pub fn tx_hash(&self) -> H256 {
+        gear_core::utils::hash(&self.encode()).into()
+    }
+
+    /// Ethexe transaction reference block hash
+    ///
+    /// Reference block hash is used for a transaction mortality check.
+    pub fn reference_block(&self) -> H256 {
+        self.transaction.reference_block
+    }
+}
+
+impl fmt::Debug for SignedOffchainTransaction {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("SignedOffchainTransaction")
+            .field("signature", &hex::encode(&self.signature))
+            .field("transaction", &self.transaction)
+            .finish()
+    }
+}
+
+impl fmt::Display for SignedOffchainTransaction {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "SignedOffchainTransaction {{ signature: 0x{}, transaction: {} }}",
+            hex::encode(&self.signature),
+            self.transaction
+        )
+    }
+}
+
+/// Ethexe offchain transaction with a reference block for mortality.
+#[derive(Debug, Clone, Encode, Decode, PartialEq, Eq)]
+#[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
+pub struct OffchainTransaction {
+    pub raw: RawOffchainTransaction,
+    pub reference_block: H256,
+}
+
+impl OffchainTransaction {
+    /// Recent block hashes window size used to check transaction mortality.
+    ///
+    /// ### Rationale
+    /// The constant could have been defined in the `ethexe-db`,
+    /// but defined here to ease upgrades without invalidation of the transactions
+    /// stores.
+    pub const BLOCK_HASHES_WINDOW_SIZE: u32 = 32;
+}
+
+impl fmt::Display for OffchainTransaction {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "OffchainTransaction {{ raw: {}, reference_block: {} }}",
+            self.raw, self.reference_block
+        )
+    }
+}
+
+/// Raw ethexe offchain transaction.
+///
+/// A particular job to be processed without external specifics.
+#[derive(Debug, Clone, Encode, Decode, PartialEq, Eq)]
+#[cfg_attr(feature = "std", derive(serde::Serialize, serde::Deserialize))]
+pub enum RawOffchainTransaction {
+    SendMessage { program_id: H160, payload: Vec<u8> },
+}
+
+impl RawOffchainTransaction {
+    /// Gets the program id of the transaction.
+    pub fn program_id(&self) -> H160 {
+        match self {
+            RawOffchainTransaction::SendMessage { program_id, .. } => *program_id,
+        }
+    }
+
+    /// Gets the payload of the transaction.
+    pub fn payload(&self) -> &[u8] {
+        match self {
+            RawOffchainTransaction::SendMessage { payload, .. } => payload,
+        }
+    }
+}
+
+impl fmt::Display for RawOffchainTransaction {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            RawOffchainTransaction::SendMessage {
+                program_id,
+                payload,
+            } => f
+                .debug_struct("SendMessage")
+                .field("program_id", program_id)
+                .field("payload", &hex::encode(payload))
+                .finish(),
+        }
+    }
+}

ethexe/db/src/database.rs

@@ -22,10 +22,12 @@ use crate::{
     overlay::{CASOverlay, KVOverlay},
     CASDatabase, KVDatabase,
 };
+use anyhow::{bail, Result};
 use ethexe_common::{
     db::{BlockHeader, BlockMetaStorage, CodeInfo, CodesStorage, Schedule},
     events::BlockRequestEvent,
     gear::StateTransition,
+    tx_pool::{OffchainTransaction, SignedOffchainTransaction},
 };
 use ethexe_runtime_common::state::{
     Allocations, DispatchStash, HashOf, Mailbox, MemoryPages, MemoryPagesRegion, MessageQueue,
@@ -58,6 +60,7 @@ enum KeyPrefix {
     CodeValid = 10,
     BlockStartSchedule = 11,
     BlockEndSchedule = 12,
+    SignedTransaction = 13,
 }
 
 impl KeyPrefix {
@@ -444,6 +447,60 @@ impl Database {
         self.cas.write(data)
     }
 
+    pub fn get_offchain_transaction(&self, tx_hash: H256) -> Option<SignedOffchainTransaction> {
+        self.kv
+            .get(&KeyPrefix::SignedTransaction.one(tx_hash))
+            .map(|data| {
+                Decode::decode(&mut data.as_slice())
+                    .expect("failed to data into `SignedTransaction`")
+            })
+    }
+
+    pub fn set_offchain_transaction(&self, tx: SignedOffchainTransaction) {
+        let tx_hash = tx.tx_hash();
+        self.kv
+            .put(&KeyPrefix::SignedTransaction.one(tx_hash), tx.encode());
+    }
+
+    pub fn check_within_recent_blocks(&self, reference_block_hash: H256) -> Result<bool> {
+        let Some((latest_valid_block_hash, latest_valid_block_header)) = self.latest_valid_block()
+        else {
+            bail!("No latest valid block found");
+        };
+        let Some(reference_block_header) = self.block_header(reference_block_hash) else {
+            bail!("No reference block found");
+        };
+
+        // If reference block is far away from the latest valid block, it's not in the window.
+        let Some(actual_window) = latest_valid_block_header
+            .height
+            .checked_sub(reference_block_header.height)
+        else {
+            bail!("Can't calculate actual window: reference block hash doesn't suit actual blocks state");
+        };
+
+        if actual_window > OffchainTransaction::BLOCK_HASHES_WINDOW_SIZE {
+            return Ok(false);
+        }
+
+        // Check against reorgs.
+        let mut block_hash = latest_valid_block_hash;
+        for _ in 0..OffchainTransaction::BLOCK_HASHES_WINDOW_SIZE {
+            if block_hash == reference_block_hash {
+                return Ok(true);
+            }
+
+            let Some(block_header) = self.block_header(block_hash) else {
+                bail!(
+                    "Block with {block_hash} hash not found in the window. Possibly reorg happened"
+                );
+            };
+            block_hash = block_header.parent_hash;
+        }
+
+        Ok(false)
+    }
+
     fn block_small_meta(&self, block_hash: H256) -> Option<BlockSmallMetaInfo> {
         self.kv
             .get(&KeyPrefix::BlockSmallMeta.two(self.router_address, block_hash))

ethexe/db/src/lib.rs

@@ -18,7 +18,7 @@
 
 //! Database library for ethexe.
 
-use gear_core::ids;
+use gear_core::utils;
 use gprimitives::H256;
 
 mod database;
@@ -32,7 +32,7 @@ pub use mem::MemDb;
 pub use rocks::RocksDatabase;
 
 pub fn hash(data: &[u8]) -> H256 {
-    ids::hash(data).into()
+    utils::hash(data).into()
 }
 
 /// Content-addressable storage database.

ethexe/db/src/overlay.rs

@@ -17,7 +17,7 @@
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
 use crate::{CASDatabase, KVDatabase, MemDb};
-use gear_core::ids::hash;
+use gear_core::utils;
 use gprimitives::H256;
 use std::collections::HashSet;
 
@@ -97,8 +97,8 @@ impl KVDatabase for KVOverlay {
 
         let mut known_keys = HashSet::new();
 
-        let filtered_iter =
-            full_iter.filter_map(move |(k, v)| known_keys.insert(hash(&k)).then_some((k, v)));
+        let filtered_iter = full_iter
+            .filter_map(move |(k, v)| known_keys.insert(utils::hash(&k)).then_some((k, v)));
 
         Box::new(filtered_iter)
     }