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

Cargo.toml

@@ -311,6 +311,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 }
 
 # Common executor between `sandbox-host` and `lazy-pages-fuzzer`
 wasmi = { package = "wasmi", version = "0.38" }

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)]
+pub struct SignedTransaction {
+    pub signature: Vec<u8>,
+    pub transaction: Transaction,
+}
+
+impl SignedTransaction {
+    /// Ethexe transaction blake2b256 hash.
+    pub fn tx_hash(&self) -> H256 {
+        gear_core::ids::hash(&self.encode()).into()
+    }
+
+    /// Ethexe transaction reference block hash
+    ///
+    /// Reference block hash is used for a transcation mortality check.
+    pub fn reference_block_hash(&self) -> H256 {
+        self.transaction.reference_block
+    }
+}
+
+impl fmt::Debug for SignedTransaction {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("SignedEthexeTransaction")
+            .field("signature", &hex::encode(&self.signature))
+            .field("transaction", &self.transaction)
+            .finish()
+    }
+}
+
+impl fmt::Display for SignedTransaction {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "SignedEthexeTransaction {{ signature: 0x{}, transaction: {} }}",
+            hex::encode(&self.signature),
+            self.transaction
+        )
+    }
+}
+
+/// Ethexe transaction with a reference block for mortality.
+#[derive(Debug, Clone, Encode, Decode, PartialEq, Eq)]
+pub struct Transaction {
+    pub raw: RawTransacton,
+    pub reference_block: H256,
+}
+
+impl fmt::Display for Transaction {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "EthexeTransaction {{ raw: {}, reference_block: {} }}",
+            self.raw, self.reference_block
+        )
+    }
+}
+
+/// Raw ethexe transaction.
+///
+/// A particular job to be processed without external specifics.
+#[derive(Debug, Clone, Encode, Decode, PartialEq, Eq)]
+pub enum RawTransacton {
+    SendMessage {
+        program_id: H160,
+        payload: Vec<u8>,
+        value: u128,
+    },
+}
+
+impl RawTransacton {
+    /// Gets the program id of the transaction.
+    pub fn program_id(&self) -> H160 {
+        match self {
+            RawTransacton::SendMessage { program_id, .. } => *program_id,
+        }
+    }
+
+    /// Gets the payload of the transaction.
+    pub fn payload(&self) -> &[u8] {
+        match self {
+            RawTransacton::SendMessage { payload, .. } => payload,
+        }
+    }
+
+    /// Gets the value of the transaction.
+    pub fn value(&self) -> u128 {
+        match self {
+            RawTransacton::SendMessage { value, .. } => *value,
+        }
+    }
+}
+
+impl fmt::Display for RawTransacton {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            RawTransacton::SendMessage {
+                program_id,
+                payload,
+                value,
+            } => f
+                .debug_struct("SendMessage")
+                .field("program_id", program_id)
+                .field("payload", &hex::encode(payload))
+                .field("value", value)
+                .finish(),
+        }
+    }
+}

ethexe/db/src/database.rs

@@ -26,6 +26,7 @@ use ethexe_common::{
     db::{BlockHeader, BlockMetaStorage, CodeInfo, CodesStorage, Schedule},
     events::BlockRequestEvent,
     gear::StateTransition,
+    tx_pool::SignedTransaction,
 };
 use ethexe_runtime_common::state::{
     Allocations, DispatchStash, HashOf, Mailbox, MemoryPages, MemoryPagesRegion, MessageQueue,
@@ -42,6 +43,8 @@ use parity_scale_codec::{Decode, Encode};
 use std::collections::{BTreeMap, BTreeSet, VecDeque};
 
 const LOG_TARGET: &str = "ethexe-db";
+/// Recent block hashes window size used to check transaction mortality.
+const BLOCK_HASHES_WINDOW_SIZE: u32 = 30;
 
 #[repr(u64)]
 enum KeyPrefix {
@@ -58,6 +61,7 @@ enum KeyPrefix {
     CodeValid = 10,
     BlockStartSchedule = 11,
     BlockEndSchedule = 12,
+    Transaction = 13,
 }
 
 impl KeyPrefix {
@@ -444,6 +448,53 @@ impl Database {
         self.cas.write(data)
     }
 
+    pub fn validated_transaction(&self, tx_hash: H256) -> Option<SignedTransaction> {
+        self.kv
+            .get(&KeyPrefix::Transaction.one(tx_hash))
+            .map(|data| {
+                Decode::decode(&mut data.as_slice())
+                    .expect("failed to data into `SignedTransaction`")
+            })
+    }
+
+    pub fn set_validated_transaction(&self, tx: SignedTransaction) {
+        let tx_hash = tx.tx_hash();
+        self.kv
+            .put(&KeyPrefix::Transaction.one(tx_hash), tx.encode());
+    }
+
+    pub fn check_within_recent_blocks(&self, reference_block_hash: H256) -> bool {
+        let Some((latest_valid_block_hash, latest_valid_block_header)) = self.latest_valid_block()
+        else {
+            return false;
+        };
+        let Some(reference_block_header) = self.block_header(reference_block_hash) else {
+            return false;
+        };
+
+        // If reference block is far away from the latest valid block, it's not in the window.
+        if latest_valid_block_header.height - reference_block_header.height
+            > BLOCK_HASHES_WINDOW_SIZE
+        {
+            return false;
+        }
+
+        // Check against reorgs.
+        let mut block_hash = latest_valid_block_hash;
+        for _ in 0..BLOCK_HASHES_WINDOW_SIZE {
+            if block_hash == reference_block_hash {
+                return true;
+            }
+
+            let Some(block_header) = self.block_header(block_hash) else {
+                return false;
+            };
+            block_hash = block_header.parent_hash;
+        }
+
+        false
+    }
+
     fn block_small_meta(&self, block_hash: H256) -> Option<BlockSmallMetaInfo> {
         self.kv
             .get(&KeyPrefix::BlockSmallMeta.two(self.router_address, block_hash))

ethexe/network/src/lib.rs

@@ -367,7 +367,7 @@ impl NetworkService {
                         topic,
                     },
                 ..
-            }) if gpu_commitments_topic().hash() == topic => {
+            }) if gpu_commitments_topic().hash() == topic || tx_topic().hash() == topic => {
                 return Some(NetworkEvent::Message { source, data });
             }
             BehaviourEvent::Gossipsub(gossipsub::Event::GossipsubNotSupported { peer_id }) => {
@@ -416,7 +416,18 @@ impl NetworkService {
             .gossipsub
             .publish(gpu_commitments_topic(), data)
         {
-            log::debug!("gossipsub publishing failed: {e}")
+            log::error!("gossipsub publishing failed: {e}")
+        }
+    }
+
+    pub fn publish_transaction(&mut self, data: Vec<u8>) {
+        if let Err(e) = self
+            .swarm
+            .behaviour_mut()
+            .gossipsub
+            .publish(tx_topic(), data)
+        {
+            log::error!("gossipsub publishing failed: {e}")
         }
     }
 
@@ -533,6 +544,7 @@ impl Behaviour {
             .map_err(|e| anyhow!("`gossipsub` scoring parameters error: {e}"))?;
 
         gossipsub.subscribe(&gpu_commitments_topic())?;
+        gossipsub.subscribe(&tx_topic())?;
 
         let db_sync = db_sync::Behaviour::new(db_sync::Config::default(), peer_score_handle, db);
 
@@ -555,6 +567,10 @@ fn gpu_commitments_topic() -> gossipsub::IdentTopic {
     gossipsub::IdentTopic::new("gpu-commitments")
 }
 
+fn tx_topic() -> gossipsub::IdentTopic {
+    gossipsub::IdentTopic::new("tx")
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

ethexe/rpc/Cargo.toml

@@ -10,7 +10,7 @@ repository.workspace = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-tokio = { workspace = true }
+tokio = { workspace = true, features = ["sync"] }
 anyhow.workspace = true
 futures.workspace = true
 gprimitives = { workspace = true, features = ["serde", "ethexe"] }

ethexe/rpc/src/apis/mod.rs

@@ -20,8 +20,10 @@ mod block;
 mod code;
 mod dev;
 mod program;
+mod tx_pool;
 
 pub use block::{BlockApi, BlockServer};
 pub use code::{CodeApi, CodeServer};
 pub use dev::{DevApi, DevServer};
 pub use program::{ProgramApi, ProgramServer};
+pub use tx_pool::{TransactionPoolApi, TransactionPoolServer};