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

Cargo.toml

@@ -309,6 +309,7 @@ ethexe-utils = { path = "ethexe/utils", default-features = false }
 ethexe-validator = { path = "ethexe/validator", default-features = false }
 ethexe-rpc = { path = "ethexe/rpc", default-features = false }
 ethexe-common = { path = "ethexe/common" }
+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/db/src/database.rs

@@ -42,6 +42,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 +60,7 @@ enum KeyPrefix {
     CodeValid = 10,
     BlockStartSchedule = 11,
     BlockEndSchedule = 12,
+    Transaction = 13,
 }
 
 impl KeyPrefix {
@@ -444,6 +447,46 @@ impl Database {
         self.cas.write(data)
     }
 
+    pub fn validated_transaction(&self, tx_hash: H256) -> Option<Vec<u8>> {
+        self.kv.get(&KeyPrefix::Transaction.one(tx_hash))
+    }
+
+    pub fn set_validated_transaction(&self, tx_hash: H256, tx: Vec<u8>) {
+        self.kv.put(&KeyPrefix::Transaction.one(tx_hash), tx);
+    }
+
+    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

@@ -33,7 +33,8 @@ use libp2p::{
     connection_limits,
     core::{muxing::StreamMuxerBox, upgrade},
     futures::StreamExt,
-    gossipsub, identify, identity, kad, mdns,
+    gossipsub::{self, IdentTopic},
+    identify, identity, kad, mdns,
     multiaddr::Protocol,
     ping,
     swarm::{
@@ -120,7 +121,7 @@ impl NetworkService {
 
 #[derive(Debug)]
 enum NetworkSenderEvent {
-    PublishMessage { data: Vec<u8> },
+    PublishMessage { data: Vec<u8>, topic: IdentTopic },
     RequestDbData(db_sync::Request),
     RequestValidated(Result<db_sync::ValidatingResponse, db_sync::ValidatingResponse>),
 }
@@ -136,13 +137,21 @@ impl NetworkSender {
         (Self { tx }, rx)
     }
 
+    pub fn publish_transaction(&self, data: impl Into<Vec<u8>>) {
+        let _res = self.tx.send(NetworkSenderEvent::PublishMessage {
+            data: data.into(),
+            topic: tx_topic(),
+        });
+    }
+
     // TODO: consider to append salt here to be sure that message is unique.
     // This is important for the cases of malfunctions in ethexe, when the same message
     // needs to be sent again #4255
-    pub fn publish_message(&self, data: impl Into<Vec<u8>>) {
-        let _res = self
-            .tx
-            .send(NetworkSenderEvent::PublishMessage { data: data.into() });
+    pub fn publish_commitment(&self, data: impl Into<Vec<u8>>) {
+        let _res = self.tx.send(NetworkSenderEvent::PublishMessage {
+            data: data.into(),
+            topic: gpu_commitments_topic(),
+        });
     }
 
     pub fn request_db_data(&self, request: db_sync::Request) {
@@ -431,7 +440,7 @@ impl NetworkEventLoop {
                         topic,
                     },
                 ..
-            }) if gpu_commitments_topic().hash() == topic => {
+            }) if gpu_commitments_topic().hash() == topic || tx_topic().hash() == topic => {
                 let _res = self
                     .external_tx
                     .send(NetworkReceiverEvent::Message { source, data });
@@ -475,13 +484,8 @@ impl NetworkEventLoop {
 
     fn handle_network_rx_event(&mut self, event: NetworkSenderEvent) {
         match event {
-            NetworkSenderEvent::PublishMessage { data } => {
-                if let Err(e) = self
-                    .swarm
-                    .behaviour_mut()
-                    .gossipsub
-                    .publish(gpu_commitments_topic(), data)
-                {
+            NetworkSenderEvent::PublishMessage { data, topic } => {
+                if let Err(e) = self.swarm.behaviour_mut().gossipsub.publish(topic, data) {
                     log::debug!("gossipsub publishing failed: {e}")
                 }
             }
@@ -577,6 +581,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);
 
@@ -599,6 +604,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::*;
@@ -612,6 +621,7 @@ mod tests {
 
         let tmp_dir1 = tempfile::tempdir().unwrap();
         let config = NetworkEventLoopConfig::new_memory(tmp_dir1.path().to_path_buf(), "/memory/1");
+
         let signer1 = ethexe_signer::Signer::new(tmp_dir1.path().join("key")).unwrap();
         let db = Database::from_one(&MemDb::default(), [0; 20]);
         let service1 = NetworkService::new(config.clone(), &signer1, db).unwrap();
@@ -640,7 +650,7 @@ mod tests {
         // Send a commitment from service1
         let commitment_data = b"test commitment".to_vec();
 
-        sender.publish_message(commitment_data.clone());
+        sender.publish_commitment(commitment_data.clone());
 
         let mut receiver = service2.receiver;
 

ethexe/rpc/Cargo.toml

@@ -10,12 +10,13 @@ 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-db.workspace = true
 ethexe-processor.workspace = true
+ethexe-tx-pool.workspace = true
 tower = { workspace = true, features = ["util"] }
 tower-http = { workspace = true, features = ["cors"] }
 jsonrpsee = { workspace = true, features = ["server", "macros"] }

ethexe/rpc/src/apis/mod.rs

@@ -18,6 +18,8 @@
 
 mod block;
 mod program;
+mod tx_pool;
 
 pub use block::{BlockApi, BlockServer};
 pub use program::{ProgramApi, ProgramServer};
+pub use tx_pool::{TransactionPoolApi, TransactionPoolServer};

ethexe/rpc/src/apis/tx_pool.rs

@@ -0,0 +1,104 @@
+// This file is part of Gear.
+//
+// Copyright (C) 2024 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/>.
+
+//! Transaction pool rpc interface.
+
+use crate::errors;
+use ethexe_tx_pool::{InputTask, RawTransacton, SignedTransaction, Transaction, TxPoolSender};
+use gprimitives::{H160, H256};
+use jsonrpsee::{
+    core::{async_trait, RpcResult},
+    proc_macros::rpc,
+};
+use tokio::sync::oneshot;
+
+#[rpc(server)]
+pub trait TransactionPool {
+    #[method(name = "transactionPool_sendMessage")]
+    async fn send_message(
+        &self,
+        program_id: H160,
+        payload: Vec<u8>,
+        value: u128,
+        reference_block: H256,
+        signature: Vec<u8>,
+    ) -> RpcResult<H256>;
+}
+
+#[derive(Clone)]
+pub struct TransactionPoolApi {
+    tx_pool_sender: TxPoolSender,
+}
+
+impl TransactionPoolApi {
+    pub fn new(tx_pool_sender: TxPoolSender) -> Self {
+        Self { tx_pool_sender }
+    }
+}
+
+#[async_trait]
+impl TransactionPoolServer for TransactionPoolApi {
+    async fn send_message(
+        &self,
+        program_id: H160,
+        payload: Vec<u8>,
+        value: u128,
+        reference_block: H256,
+        signature: Vec<u8>,
+    ) -> RpcResult<H256> {
+        let signed_ethexe_tx = SignedTransaction {
+            transaction: Transaction {
+                raw: RawTransacton::SendMessage {
+                    program_id,
+                    payload,
+                    value,
+                },
+                reference_block,
+            },
+            signature,
+        };
+        log::debug!("Called send_message with vars: {signed_ethexe_tx:#?}");
+
+        let (response_sender, response_receiver) = oneshot::channel();
+        let input_task = InputTask::AddTransaction {
+            transaction: signed_ethexe_tx,
+            response_sender: Some(response_sender),
+        };
+
+        self.tx_pool_sender.send(input_task).map_err(|e| {
+            // No panic case as a responsibility of the RPC API is fulfilled.
+            // The dropped tx pool input task receiver might signalize that
+            // the transaction pool has been stooped.
+            log::error!(
+                "Failed to send tx pool add transaction input task: {e}. \
+                The receiving end in the tx pool might have been dropped."
+            );
+            errors::internal()
+        })?;
+
+        let res = response_receiver.await.map_err(|e| {
+            // No panic case as a responsibility of the RPC API is fulfilled.
+            // The dropped sender signalizes that the transaction pool
+            // has crashed or is malformed, so problems should be handled there.
+            log::error!("Tx pool has dropped response sender: {e}");
+            errors::internal()
+        })?;
+
+        res.map_err(errors::tx_pool)
+    }
+}

ethexe/rpc/src/errors.rs

@@ -18,6 +18,8 @@
 
 use jsonrpsee::types::ErrorObject;
 
+// TODO #4364: https://github.com/gear-tech/gear/issues/4364
+
 pub fn db(err: &'static str) -> ErrorObject<'static> {
     ErrorObject::owned(8000, "Database error", Some(err))
 }
@@ -29,3 +31,7 @@ pub fn runtime(err: anyhow::Error) -> ErrorObject<'static> {
 pub fn internal() -> ErrorObject<'static> {
     ErrorObject::owned(8000, "Internal error", None::<&str>)
 }
+
+pub fn tx_pool(err: anyhow::Error) -> ErrorObject<'static> {
+    ErrorObject::owned(8000, "Transaction pool error", Some(format!("{err}")))
+}