reth_transaction_pool/validate/
task.rs

1//! A validation service for transactions.
2
3use crate::{
4    blobstore::BlobStore,
5    metrics::TxPoolValidatorMetrics,
6    validate::{EthTransactionValidatorBuilder, TransactionValidatorError},
7    EthTransactionValidator, PoolTransaction, TransactionOrigin, TransactionValidationOutcome,
8    TransactionValidator,
9};
10use futures_util::{lock::Mutex, StreamExt};
11use reth_chainspec::{ChainSpecProvider, EthereumHardforks};
12use reth_evm::ConfigureEvm;
13use reth_primitives_traits::{HeaderTy, SealedBlock};
14use reth_storage_api::BlockReaderIdExt;
15use reth_tasks::Runtime;
16use std::{future::Future, pin::Pin, sync::Arc};
17use tokio::{
18    sync,
19    sync::{mpsc, oneshot},
20};
21use tokio_stream::wrappers::ReceiverStream;
22
23/// Represents a future outputting unit type and is sendable.
24type ValidationFuture = Pin<Box<dyn Future<Output = ()> + Send>>;
25
26/// Represents a stream of validation futures.
27type ValidationStream = ReceiverStream<ValidationFuture>;
28
29/// A service that performs validation jobs.
30///
31/// This listens for incoming validation jobs and executes them.
32///
33/// This should be spawned as a task: [`ValidationTask::run`]
34#[derive(Clone)]
35pub struct ValidationTask {
36    validation_jobs: Arc<Mutex<ValidationStream>>,
37}
38
39impl ValidationTask {
40    /// Creates a new cloneable task pair.
41    ///
42    /// The sender sends new (transaction) validation tasks to an available validation task.
43    pub fn new() -> (ValidationJobSender, Self) {
44        Self::with_capacity(1)
45    }
46
47    /// Creates a new cloneable task pair with the given channel capacity.
48    pub fn with_capacity(capacity: usize) -> (ValidationJobSender, Self) {
49        let (tx, rx) = mpsc::channel(capacity);
50        let metrics = TxPoolValidatorMetrics::default();
51        (ValidationJobSender { tx, metrics }, Self::with_receiver(rx))
52    }
53
54    /// Creates a new task with the given receiver.
55    pub fn with_receiver(jobs: mpsc::Receiver<Pin<Box<dyn Future<Output = ()> + Send>>>) -> Self {
56        Self { validation_jobs: Arc::new(Mutex::new(ReceiverStream::new(jobs))) }
57    }
58
59    /// Executes all new validation jobs that come in.
60    ///
61    /// This will run as long as the channel is alive and is expected to be spawned as a task.
62    pub async fn run(self) {
63        while let Some(task) = self.validation_jobs.lock().await.next().await {
64            task.await;
65        }
66    }
67}
68
69impl std::fmt::Debug for ValidationTask {
70    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
71        f.debug_struct("ValidationTask").field("validation_jobs", &"...").finish()
72    }
73}
74
75/// A sender new type for sending validation jobs to [`ValidationTask`].
76#[derive(Debug)]
77pub struct ValidationJobSender {
78    tx: mpsc::Sender<Pin<Box<dyn Future<Output = ()> + Send>>>,
79    metrics: TxPoolValidatorMetrics,
80}
81
82impl ValidationJobSender {
83    /// Sends the given job to the validation task.
84    pub async fn send(
85        &self,
86        job: Pin<Box<dyn Future<Output = ()> + Send>>,
87    ) -> Result<(), TransactionValidatorError> {
88        self.metrics.inflight_validation_jobs.increment(1);
89        let res = self
90            .tx
91            .send(job)
92            .await
93            .map_err(|_| TransactionValidatorError::ValidationServiceUnreachable);
94        self.metrics.inflight_validation_jobs.decrement(1);
95        res
96    }
97}
98
99/// A [`TransactionValidator`] implementation that validates ethereum transaction.
100/// This validator is non-blocking, all validation work is done in a separate task.
101#[derive(Debug)]
102pub struct TransactionValidationTaskExecutor<V> {
103    /// The validator that will validate transactions on a separate task.
104    pub validator: Arc<V>,
105    /// The sender half to validation tasks that perform the actual validation.
106    pub to_validation_task: Arc<sync::Mutex<ValidationJobSender>>,
107}
108
109impl<V> Clone for TransactionValidationTaskExecutor<V> {
110    fn clone(&self) -> Self {
111        Self {
112            validator: self.validator.clone(),
113            to_validation_task: self.to_validation_task.clone(),
114        }
115    }
116}
117
118// === impl TransactionValidationTaskExecutor ===
119
120impl TransactionValidationTaskExecutor<()> {
121    /// Convenience method to create a [`EthTransactionValidatorBuilder`]
122    pub fn eth_builder<Client, Evm>(
123        client: Client,
124        evm_config: Evm,
125    ) -> EthTransactionValidatorBuilder<Client, Evm>
126    where
127        Client: ChainSpecProvider<ChainSpec: EthereumHardforks>
128            + BlockReaderIdExt<Header = HeaderTy<Evm::Primitives>>,
129        Evm: ConfigureEvm,
130    {
131        EthTransactionValidatorBuilder::new(client, evm_config)
132    }
133}
134
135impl<V> TransactionValidationTaskExecutor<V> {
136    /// Maps the given validator to a new type.
137    pub fn map<F, T>(self, mut f: F) -> TransactionValidationTaskExecutor<T>
138    where
139        F: FnMut(V) -> T,
140    {
141        TransactionValidationTaskExecutor {
142            validator: Arc::new(f(Arc::into_inner(self.validator).unwrap())),
143            to_validation_task: self.to_validation_task,
144        }
145    }
146
147    /// Returns the validator.
148    pub fn validator(&self) -> &V {
149        &self.validator
150    }
151}
152
153impl<Client, Tx, Evm> TransactionValidationTaskExecutor<EthTransactionValidator<Client, Tx, Evm>> {
154    /// Creates a new instance for the given client
155    ///
156    /// This will spawn a single validation tasks that performs the actual validation.
157    /// See [`TransactionValidationTaskExecutor::eth_with_additional_tasks`]
158    pub fn eth<S: BlobStore>(client: Client, evm_config: Evm, blob_store: S, tasks: Runtime) -> Self
159    where
160        Client: ChainSpecProvider<ChainSpec: EthereumHardforks>
161            + BlockReaderIdExt<Header = HeaderTy<Evm::Primitives>>,
162        Evm: ConfigureEvm,
163    {
164        Self::eth_with_additional_tasks(client, evm_config, blob_store, tasks, 0)
165    }
166
167    /// Creates a new instance for the given client
168    ///
169    /// By default this will enable support for:
170    ///   - shanghai
171    ///   - eip1559
172    ///   - eip2930
173    ///
174    /// This will always spawn a validation task that performs the actual validation. It will spawn
175    /// `num_additional_tasks` additional tasks.
176    pub fn eth_with_additional_tasks<S: BlobStore>(
177        client: Client,
178        evm_config: Evm,
179        blob_store: S,
180        tasks: Runtime,
181        num_additional_tasks: usize,
182    ) -> Self
183    where
184        Client: ChainSpecProvider<ChainSpec: EthereumHardforks>
185            + BlockReaderIdExt<Header = HeaderTy<Evm::Primitives>>,
186        Evm: ConfigureEvm,
187    {
188        EthTransactionValidatorBuilder::new(client, evm_config)
189            .with_additional_tasks(num_additional_tasks)
190            .build_with_tasks(tasks, blob_store)
191    }
192}
193
194impl<V> TransactionValidationTaskExecutor<V> {
195    /// Creates a new executor instance with the given validator for transaction validation.
196    ///
197    /// Initializes the executor with the provided validator and sets up communication for
198    /// validation tasks.
199    pub fn new(validator: V) -> (Self, ValidationTask) {
200        let (tx, task) = ValidationTask::new();
201        (
202            Self {
203                validator: Arc::new(validator),
204                to_validation_task: Arc::new(sync::Mutex::new(tx)),
205            },
206            task,
207        )
208    }
209
210    /// Creates a new executor and spawns the validation tasks on the given runtime.
211    ///
212    /// This spawns `additional_tasks` extra blocking tasks plus one critical blocking task
213    /// for the validation service.
214    pub fn spawn(validator: V, tasks: &Runtime, additional_tasks: usize) -> Self {
215        let (tx, task) = ValidationTask::new();
216
217        for _ in 0..additional_tasks {
218            let task = task.clone();
219            tasks.spawn_blocking_task(async move {
220                task.run().await;
221            });
222        }
223
224        tasks.spawn_critical_blocking_task("transaction-validation-service", async move {
225            task.run().await;
226        });
227
228        Self { validator: Arc::new(validator), to_validation_task: Arc::new(sync::Mutex::new(tx)) }
229    }
230}
231
232impl<V> TransactionValidator for TransactionValidationTaskExecutor<V>
233where
234    V: TransactionValidator + 'static,
235{
236    type Transaction = <V as TransactionValidator>::Transaction;
237    type Block = V::Block;
238
239    async fn validate_transaction(
240        &self,
241        origin: TransactionOrigin,
242        transaction: Self::Transaction,
243    ) -> TransactionValidationOutcome<Self::Transaction> {
244        let hash = *transaction.hash();
245        let (tx, rx) = oneshot::channel();
246        {
247            let res = {
248                let validator = self.validator.clone();
249                let fut = Box::pin(async move {
250                    let res = validator.validate_transaction(origin, transaction).await;
251                    let _ = tx.send(res);
252                });
253                self.to_validation_task.lock().await.send(fut).await
254            };
255            if res.is_err() {
256                return TransactionValidationOutcome::Error(
257                    hash,
258                    Box::new(TransactionValidatorError::ValidationServiceUnreachable),
259                );
260            }
261        }
262
263        match rx.await {
264            Ok(res) => res,
265            Err(_) => TransactionValidationOutcome::Error(
266                hash,
267                Box::new(TransactionValidatorError::ValidationServiceUnreachable),
268            ),
269        }
270    }
271
272    async fn validate_transactions(
273        &self,
274        transactions: impl IntoIterator<Item = (TransactionOrigin, Self::Transaction), IntoIter: Send>
275            + Send,
276    ) -> Vec<TransactionValidationOutcome<Self::Transaction>> {
277        let transactions: Vec<_> = transactions.into_iter().collect();
278        let hashes: Vec<_> = transactions.iter().map(|(_, tx)| *tx.hash()).collect();
279        let (tx, rx) = oneshot::channel();
280        {
281            let res = {
282                let validator = self.validator.clone();
283                let fut = Box::pin(async move {
284                    let res = validator.validate_transactions(transactions).await;
285                    let _ = tx.send(res);
286                });
287                self.to_validation_task.lock().await.send(fut).await
288            };
289            if res.is_err() {
290                return validation_service_error_outcomes(hashes)
291            }
292        }
293        match rx.await {
294            Ok(res) => res,
295            Err(_) => validation_service_error_outcomes(hashes),
296        }
297    }
298
299    async fn validate_transactions_with_origin(
300        &self,
301        origin: TransactionOrigin,
302        transactions: impl IntoIterator<Item = Self::Transaction, IntoIter: Send> + Send,
303    ) -> Vec<TransactionValidationOutcome<Self::Transaction>> {
304        let transactions: Vec<_> = transactions.into_iter().collect();
305        let hashes: Vec<_> = transactions.iter().map(|tx| *tx.hash()).collect();
306        let (tx, rx) = oneshot::channel();
307        let validator = self.validator.clone();
308        let fut = Box::pin(async move {
309            let res = validator.validate_transactions_with_origin(origin, transactions).await;
310            let _ = tx.send(res);
311        });
312
313        if self.to_validation_task.lock().await.send(fut).await.is_err() {
314            return validation_service_error_outcomes(hashes)
315        }
316
317        match rx.await {
318            Ok(res) => res,
319            Err(_) => validation_service_error_outcomes(hashes),
320        }
321    }
322
323    fn on_new_head_block(&self, new_tip_block: &SealedBlock<Self::Block>) {
324        self.validator.on_new_head_block(new_tip_block)
325    }
326}
327
328#[inline]
329fn validation_service_error_outcomes<T: PoolTransaction>(
330    hashes: Vec<alloy_primitives::TxHash>,
331) -> Vec<TransactionValidationOutcome<T>> {
332    hashes
333        .into_iter()
334        .map(|hash| {
335            TransactionValidationOutcome::Error(
336                hash,
337                Box::new(TransactionValidatorError::ValidationServiceUnreachable),
338            )
339        })
340        .collect()
341}
342
343#[cfg(test)]
344mod tests {
345    use super::*;
346    use crate::{
347        test_utils::MockTransaction,
348        validate::{TransactionValidationOutcome, ValidTransaction},
349        TransactionOrigin,
350    };
351    use alloy_primitives::{Address, U256};
352
353    #[derive(Debug)]
354    struct NoopValidator;
355
356    impl TransactionValidator for NoopValidator {
357        type Transaction = MockTransaction;
358        type Block = reth_ethereum_primitives::Block;
359
360        async fn validate_transaction(
361            &self,
362            _origin: TransactionOrigin,
363            transaction: Self::Transaction,
364        ) -> TransactionValidationOutcome<Self::Transaction> {
365            TransactionValidationOutcome::Valid {
366                balance: U256::ZERO,
367                state_nonce: 0,
368                bytecode_hash: None,
369                transaction: ValidTransaction::Valid(transaction),
370                propagate: false,
371                authorities: Some(Vec::<Address>::new()),
372            }
373        }
374    }
375
376    #[tokio::test]
377    async fn executor_new_spawns_and_validates_single() {
378        let validator = NoopValidator;
379        let (executor, task) = TransactionValidationTaskExecutor::new(validator);
380        tokio::spawn(task.run());
381        let tx = MockTransaction::legacy();
382        let out = executor.validate_transaction(TransactionOrigin::External, tx).await;
383        assert!(matches!(out, TransactionValidationOutcome::Valid { .. }));
384    }
385
386    #[tokio::test]
387    async fn executor_new_spawns_and_validates_batch() {
388        let validator = NoopValidator;
389        let (executor, task) = TransactionValidationTaskExecutor::new(validator);
390        tokio::spawn(task.run());
391        let txs = vec![
392            (TransactionOrigin::External, MockTransaction::legacy()),
393            (TransactionOrigin::Local, MockTransaction::legacy()),
394        ];
395        let out = executor.validate_transactions(txs).await;
396        assert_eq!(out.len(), 2);
397        assert!(out.iter().all(|o| matches!(o, TransactionValidationOutcome::Valid { .. })));
398    }
399
400    #[derive(Debug)]
401    struct SameOriginBatchValidator;
402
403    impl TransactionValidator for SameOriginBatchValidator {
404        type Transaction = MockTransaction;
405        type Block = reth_ethereum_primitives::Block;
406
407        async fn validate_transaction(
408            &self,
409            _origin: TransactionOrigin,
410            _transaction: Self::Transaction,
411        ) -> TransactionValidationOutcome<Self::Transaction> {
412            panic!("same-origin batches must use the batch validator")
413        }
414
415        async fn validate_transactions_with_origin(
416            &self,
417            origin: TransactionOrigin,
418            transactions: impl IntoIterator<Item = Self::Transaction, IntoIter: Send> + Send,
419        ) -> Vec<TransactionValidationOutcome<Self::Transaction>> {
420            transactions
421                .into_iter()
422                .map(|transaction| TransactionValidationOutcome::Valid {
423                    balance: U256::ZERO,
424                    state_nonce: 0,
425                    bytecode_hash: None,
426                    transaction: ValidTransaction::Valid(transaction),
427                    propagate: matches!(origin, TransactionOrigin::Local),
428                    authorities: None,
429                })
430                .collect()
431        }
432    }
433
434    #[tokio::test]
435    async fn executor_forwards_same_origin_batches() {
436        let (executor, task) = TransactionValidationTaskExecutor::new(SameOriginBatchValidator);
437        tokio::spawn(task.run());
438
439        let transactions = vec![MockTransaction::legacy(), MockTransaction::eip1559()];
440        let expected_hashes = transactions.iter().map(|tx| *tx.hash()).collect::<Vec<_>>();
441        let outcomes = executor
442            .validate_transactions_with_origin(TransactionOrigin::Local, transactions)
443            .await;
444
445        assert_eq!(outcomes.len(), expected_hashes.len());
446        assert!(outcomes.into_iter().zip(expected_hashes).all(|(outcome, expected_hash)| {
447            matches!(
448                outcome,
449                TransactionValidationOutcome::Valid { transaction, propagate: true, .. }
450                    if transaction.hash() == &expected_hash
451            )
452        }));
453    }
454}
reth_transaction_pool/validate/task.rs

reth_transaction_pool/validate/
task.rs
