logos-co · bacv · Jun 19, 2025 · Jun 17, 2025 · Jun 17, 2025 · Jun 17, 2025
diff --git a/dependencies_graph.dot b/dependencies_graph.dot
@@ -30,11 +30,11 @@ digraph {
     28 [ label = "nomos-da-verifier" shape = box]
     29 [ label = "nomos-storage" shape = box]
     30 [ label = "nomos-mempool" shape = box]
-    31 [ label = "nomos-tracing-service" shape = box]
-    32 [ label = "nomos-time" shape = box]
-    33 [ label = "nomos-da-dispersal" shape = box]
-    34 [ label = "nomos-da-indexer" shape = box]
-    35 [ label = "nomos-mantle-core" shape = box]
+    31 [ label = "nomos-mantle-core" shape = box]
+    32 [ label = "nomos-tracing-service" shape = box]
+    33 [ label = "nomos-time" shape = box]
+    34 [ label = "nomos-da-dispersal" shape = box]
+    35 [ label = "nomos-da-indexer" shape = box]
     36 [ label = "nomos-node" shape = box]
     37 [ label = "nomos-system-sig" shape = box]
     38 [ label = "executor-http-client" shape = box]
@@ -50,11 +50,11 @@ digraph {
     6 -> 36 [ ]
     7 -> 8 [ ]
     7 -> 3 [ ]
-    9 -> 33 [ ]
     9 -> 34 [ ]
+    9 -> 35 [ ]
     10 -> 11 [ ]
     10 -> 30 [ ]
-    10 -> 32 [ ]
+    10 -> 33 [ ]
     11 -> 14 [ ]
     11 -> 16 [ ]
     11 -> 17 [ ]
@@ -83,16 +83,16 @@ digraph {
     28 -> 29 [ ]
     29 -> 3 [ ]
     30 -> 18 [ ]
+    30 -> 31 [ ]
     30 -> 30 [ color = blue]
     30 -> 16 [ ]
-    30 -> 31 [ color = blue]
-    31 -> 27 [ ]
-    32 -> 0 [ ]
-    33 -> 30 [ ]
-    34 -> 10 [ ]
-    35 -> 26 [ ]
+    30 -> 32 [ color = blue]
+    31 -> 26 [ ]
+    32 -> 27 [ ]
+    33 -> 0 [ ]
+    34 -> 30 [ ]
+    35 -> 10 [ ]
     36 -> 9 [ ]
-    36 -> 35 [ ]
     36 -> 37 [ ]
     38 -> 19 [ ]
     40 -> 38 [ ]

diff --git a/nodes/nomos-executor/executor/Cargo.toml b/nodes/nomos-executor/executor/Cargo.toml
@@ -43,6 +43,7 @@ nomos-network            = { workspace = true, features = ["libp2p"] }
 nomos-node               = { workspace = true }
 nomos-storage            = { workspace = true, features = ["rocksdb-backend"] }
 nomos-time               = { workspace = true }
+nomos-utils              = { workspace = true }
 overwatch                = { workspace = true }
 rand                     = "0.8"
 rand_chacha              = "0.3"

diff --git a/nodes/nomos-executor/executor/src/lib.rs b/nodes/nomos-executor/executor/src/lib.rs
@@ -39,6 +39,7 @@ use nomos_node::{
     NomosDaMembership, RocksBackend, SystemSig, Wire, MB16,
 };
 use nomos_time::backends::NtpTimeBackend;
+use nomos_utils::noop_service::NoService;
 use overwatch::derive_services;
 use rand_chacha::ChaCha20Rng;
 
@@ -266,4 +267,5 @@ pub struct NomosExecutor {
     http: ApiService,
     storage: StorageService,
     system_sig: SystemSigService,
+    no_service: NoService,
 }
diff --git a/nodes/nomos-executor/executor/src/main.rs b/nodes/nomos-executor/executor/src/main.rs
@@ -82,6 +82,7 @@ async fn main() -> Result<()> {
                     topic: String::from(CL_TOPIC),
                     id: <SignedMantleTx as Transaction>::hash,
                 },
+                    processor: (),
                 recovery_path: config.mempool.cl_pool_recovery_path,
             },
             da_mempool: DaMempoolSettings {
@@ -109,6 +110,7 @@ async fn main() -> Result<()> {
                     initial_locators_mapping: HashMap::default(),
                 },
             },
+            no_service: (),
         },
         None,
     )

diff --git a/nodes/nomos-node/node/Cargo.toml b/nodes/nomos-node/node/Cargo.toml
@@ -49,6 +49,7 @@ nomos-system-sig         = { workspace = true }
 nomos-time               = { workspace = true, features = ["ntp", "serde"] }
 nomos-tracing            = { workspace = true }
 nomos-tracing-service    = { workspace = true }
+nomos-utils              = { workspace = true }
 overwatch                = { workspace = true }
 rand_chacha              = "0.3"
 serde                    = "1"

diff --git a/nodes/nomos-node/node/src/lib.rs b/nodes/nomos-node/node/src/lib.rs
@@ -47,6 +47,7 @@ pub use nomos_system_sig::SystemSig;
 use nomos_time::backends::NtpTimeBackend;
 #[cfg(feature = "tracing")]
 pub use nomos_tracing_service::Tracing;
+use nomos_utils::noop_service::NoService;
 use overwatch::derive_services;
 use rand_chacha::ChaCha20Rng;
 use serde::{de::DeserializeOwned, Serialize};
@@ -241,4 +242,5 @@ pub struct Nomos {
     http: ApiService,
     storage: StorageService,
     system_sig: SystemSigService,
+    no_service: NoService,
 }
diff --git a/nodes/nomos-node/node/src/main.rs b/nodes/nomos-node/node/src/main.rs
@@ -82,6 +82,7 @@ async fn main() -> Result<()> {
                     topic: String::from(nomos_node::CL_TOPIC),
                     id: <SignedMantleTx as Transaction>::hash,
                 },
+                processor: (),
                 recovery_path: config.mempool.cl_pool_recovery_path,
             },
             da_mempool: nomos_mempool::DaMempoolSettings {
@@ -108,6 +109,7 @@ async fn main() -> Result<()> {
                     initial_locators_mapping: HashMap::default(),
                 },
             },
+            no_service: (),
         },
         None,
     )

@@ -1,3 +1,3 @@
 mod gas;
-mod ops;
+pub mod ops;
 pub mod tx;
diff --git a/nomos-services/mempool/Cargo.toml b/nomos-services/mempool/Cargo.toml
@@ -22,7 +22,9 @@ linked-hash-map       = { version = "0.5.6", optional = true, features = ["serde
 nomos-core            = { workspace = true }
 nomos-da-network-core = { workspace = true }
 nomos-da-sampling     = { workspace = true }
+nomos-mantle-core     = { workspace = true }
 nomos-network         = { workspace = true }
+nomos-utils           = { workspace = true }
 overwatch             = { workspace = true }
 rand                  = { version = "0.8" }
 serde                 = { version = "1.0", features = ["derive"] }

diff --git a/nomos-services/mempool/src/backend/mockpool/mod.rs b/nomos-services/mempool/src/backend/mockpool/mod.rs
@@ -228,7 +228,7 @@ where
     BlockId: Ord + Copy,
 {
     type Error = Infallible;
-    type Settings = TxMempoolSettings<(), ()>;
+    type Settings = TxMempoolSettings<(), (), ()>;
 
     fn from_settings(_settings: &Self::Settings) -> Result<Self, Self::Error> {
         Ok(<Self as MemPool>::new(()))

diff --git a/nomos-services/mempool/src/lib.rs b/nomos-services/mempool/src/lib.rs
@@ -1,8 +1,8 @@
 pub mod backend;
 pub mod da;
 pub mod network;
+pub mod processor;
 pub mod tx;
-pub mod verify;
 
 use std::fmt::{Debug, Error, Formatter};
 

diff --git a/nomos-services/mempool/src/processor/mod.rs b/nomos-services/mempool/src/processor/mod.rs
@@ -0,0 +1,23 @@
+pub mod noop;
+pub mod tx;
+
+use std::error::Error;
+
+use overwatch::services::{relay::OutboundRelay, ServiceData};
+
+#[async_trait::async_trait]
+pub trait PayloadProcessor {
+    type Payload;
+    type Settings;
+    type Error: Error;
+
+    type DaSamplingService: ServiceData;
+
+    fn new(
+        settings: Self::Settings,
+        outbound_relay: OutboundRelay<<Self::DaSamplingService as ServiceData>::Message>,
+    ) -> Self;
+
+    /// Executes required procedures before adding payload to the pool.
+    async fn process(&self, payload: &Self::Payload) -> Result<(), Vec<Self::Error>>;
+}
diff --git a/nomos-services/mempool/src/processor/noop.rs b/nomos-services/mempool/src/processor/noop.rs
@@ -0,0 +1,31 @@
+use std::{convert::Infallible, marker::PhantomData};
+
+use overwatch::services::{relay::OutboundRelay, ServiceData};
+
+use super::PayloadProcessor;
+
+pub type NoOpPayloadProcessor<Service, Payload> = PhantomData<(Service, Payload)>;
+
+#[async_trait::async_trait]
+impl<Service, Payload> PayloadProcessor for NoOpPayloadProcessor<Service, Payload>
+where
+    Payload: Send + Sync,
+    Service: ServiceData + Send + Sync,
+{
+    type Payload = Payload;
+    type Settings = ();
+    type Error = Infallible;
+
+    type DaSamplingService = Service;
+
+    fn new(
+        (): Self::Settings,
+        _: OutboundRelay<<Self::DaSamplingService as ServiceData>::Message>,
+    ) -> Self {
+        Self
+    }
+
+    async fn process(&self, _: &Self::Payload) -> Result<(), Vec<Self::Error>> {
+        Ok(())
+    }
+}
diff --git a/nomos-services/mempool/src/processor/tx.rs b/nomos-services/mempool/src/processor/tx.rs
@@ -0,0 +1,81 @@
+use std::error::Error;
+
+use futures::{stream::FuturesUnordered, StreamExt};
+use nomos_core::da::BlobId;
+use nomos_da_sampling::DaSamplingServiceMsg;
+use nomos_mantle_core::{ops::Op, tx::SignedMantleTx};
+use overwatch::services::{relay::OutboundRelay, ServiceData};
+
+use super::PayloadProcessor;
+
+#[derive(thiserror::Error, Debug)]
+pub enum SignedTxProcessorError {
+    #[error("Error from sampling relay {0}")]
+    Sampling(Box<dyn Error + Send>),
+}
+
+impl SignedTxProcessorError {
+    fn sampling_error(err: impl Error + Send + 'static) -> Self {
+        Self::Sampling(Box::new(err))
+    }
+}
+
+pub struct SignedTxProcessor<SamplingService>
+where
+    SamplingService: ServiceData,
+{
+    sampling_relay: OutboundRelay<<SamplingService as ServiceData>::Message>,
+}
+
+#[async_trait::async_trait]
+impl<SamplingService> PayloadProcessor for SignedTxProcessor<SamplingService>
+where
+    SamplingService: ServiceData<Message = DaSamplingServiceMsg<BlobId>> + Send + Sync,
+{
+    type Payload = SignedMantleTx;
+    type Settings = ();
+    type Error = SignedTxProcessorError;
+
+    type DaSamplingService = SamplingService;
+
+    fn new(
+        (): Self::Settings,
+        sampling_relay: OutboundRelay<<Self::DaSamplingService as ServiceData>::Message>,
+    ) -> Self {
+        Self { sampling_relay }
+    }
+
+    async fn process(&self, payload: &Self::Payload) -> Result<(), Vec<Self::Error>> {
+        let all_futures: FuturesUnordered<_> = payload
+            .mantle_tx
+            .ops
+            .iter()
+            .filter_map(|op| {
+                if let Op::Blob(blob_op) = op {
+                    Some(async {
+                        self.sampling_relay
+                            .send(DaSamplingServiceMsg::TriggerSampling {
+                                blob_id: blob_op.blob,
+                            })
+                            .await
+                            .map_err(|(e, _)| SignedTxProcessorError::sampling_error(e))
+                    })
+                } else {
+                    None
+                }
+            })
+            .collect();
+
+        let errors: Vec<SignedTxProcessorError> = StreamExt::collect::<Vec<_>>(all_futures)
+            .await
+            .into_iter()
+            .filter_map(Result::err)
+            .collect();
+
+        if errors.is_empty() {
+            Ok(())
+        } else {
+            Err(errors)
+        }
+    }
+}