make lint

Cargo.lock

@@ -10885,6 +10885,7 @@ dependencies = [
  "tempfile",
  "thiserror 2.0.17",
  "tokio",
+ "tokio-stream",
  "tracing",
 ]
 

crates/optimism/node/src/node.rs

@@ -995,14 +995,15 @@ where
                     .additional_validation_tasks
                     .unwrap_or_else(|| ctx.config().txpool.additional_validation_tasks),
             )
-            .build_with_tasks(ctx.task_executor().clone(), blob_store.clone())
+            .into_tasks_builder(blob_store.clone())
             .map(|validator| {
                 OpTransactionValidator::new(validator)
                     // In --dev mode we can't require gas fees because we're unable to decode
                     // the L1 block info
                     .require_l1_data_gas_fee(!ctx.config().dev.dev)
                     .with_supervisor(supervisor_client.clone())
-            });
+            })
+            .build_and_spawn(ctx.task_executor().clone());
 
         let final_pool_config = pool_config_overrides.apply(ctx.pool_config());
 

crates/transaction-pool/Cargo.toml

@@ -73,6 +73,7 @@ tempfile.workspace = true
 serde_json.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread"] }
 futures.workspace = true
+tokio-stream.workspace = true
 
 [features]
 serde = [

crates/transaction-pool/src/validate/task.rs

@@ -69,18 +69,16 @@ impl<V: TransactionValidator> ValidationTask<V> {
 
         loop {
             // Lock the receiver and batch receive transactions
-            let count = {
-                let mut jobs = self.validation_jobs.lock().await;
-                jobs.recv_many(&mut buffer, Self::BATCH_SIZE).await
-            };
-
-            self.metrics.inflight_validation_jobs.decrement(count as f64);
+            let count =
+                self.validation_jobs.lock().await.recv_many(&mut buffer, Self::BATCH_SIZE).await;
 
             if count == 0 {
                 // Channel closed, exit
                 break;
             }
 
+            self.metrics.inflight_validation_jobs.decrement(count as f64);
+
             // Split into transactions and response senders
             #[expect(clippy::iter_with_drain)]
             let (txs, senders): (Vec<_>, Vec<_>) =
@@ -213,16 +211,12 @@ where
         }
 
         // Wait for the result
-        let result = match rx.await {
-            Ok(res) => res,
-            Err(_) => TransactionValidationOutcome::Error(
+        rx.await.unwrap_or_else(|_| {
+            TransactionValidationOutcome::Error(
                 hash,
                 Box::new(TransactionValidatorError::ValidationServiceUnreachable),
-            ),
-        };
-
-        self.metrics.inflight_validation_jobs.decrement(1);
-        result
+            )
+        })
     }
 
     async fn validate_transactions(
@@ -247,8 +241,6 @@ where
 
             self.metrics.inflight_validation_jobs.increment(1);
             if self.to_validation_task.send((origin, transaction, tx)).is_err() {
-                // Channel closed - return errors for remaining
-                self.metrics.inflight_validation_jobs.decrement(len as f64);
                 return hashes
                     .into_iter()
                     .map(|h| {
@@ -274,7 +266,6 @@ where
             results.push(result);
         }
 
-        self.metrics.inflight_validation_jobs.decrement(len as f64);
         results
     }