reth_node_builder/components/
pool.rs1use alloy_primitives::Address;
4use reth_chain_state::CanonStateSubscriptions;
5use reth_node_api::TxTy;
6use reth_transaction_pool::{
7 blobstore::DiskFileBlobStore, CoinbaseTipOrdering, PoolConfig, PoolTransaction, SubPoolLimit,
8 TransactionPool, TransactionValidationTaskExecutor, TransactionValidator,
9};
10use std::{collections::HashSet, future::Future};
11
12use crate::{BuilderContext, FullNodeTypes};
13
14pub trait PoolBuilder<Node: FullNodeTypes>: Send {
16 type Pool: TransactionPool<Transaction: PoolTransaction<Consensus = TxTy<Node::Types>>>
18 + Unpin
19 + 'static;
20
21 fn build_pool(
23 self,
24 ctx: &BuilderContext<Node>,
25 ) -> impl Future<Output = eyre::Result<Self::Pool>> + Send;
26}
27
28impl<Node, F, Fut, Pool> PoolBuilder<Node> for F
29where
30 Node: FullNodeTypes,
31 Pool: TransactionPool<Transaction: PoolTransaction<Consensus = TxTy<Node::Types>>>
32 + Unpin
33 + 'static,
34 F: FnOnce(&BuilderContext<Node>) -> Fut + Send,
35 Fut: Future<Output = eyre::Result<Pool>> + Send,
36{
37 type Pool = Pool;
38
39 fn build_pool(
40 self,
41 ctx: &BuilderContext<Node>,
42 ) -> impl Future<Output = eyre::Result<Self::Pool>> {
43 self(ctx)
44 }
45}
46
47#[derive(Debug, Clone, Default)]
49pub struct PoolBuilderConfigOverrides {
50 pub pending_limit: Option<SubPoolLimit>,
52 pub basefee_limit: Option<SubPoolLimit>,
54 pub queued_limit: Option<SubPoolLimit>,
56 pub blob_limit: Option<SubPoolLimit>,
58 pub max_account_slots: Option<usize>,
60 pub minimal_protocol_basefee: Option<u64>,
62 pub local_addresses: HashSet<Address>,
64 pub additional_validation_tasks: Option<usize>,
66}
67
68impl PoolBuilderConfigOverrides {
69 pub fn apply(self, mut config: PoolConfig) -> PoolConfig {
71 let Self {
72 pending_limit,
73 basefee_limit,
74 queued_limit,
75 blob_limit,
76 max_account_slots,
77 minimal_protocol_basefee,
78 local_addresses,
79 additional_validation_tasks: _,
80 } = self;
81
82 if let Some(pending_limit) = pending_limit {
83 config.pending_limit = pending_limit;
84 }
85 if let Some(basefee_limit) = basefee_limit {
86 config.basefee_limit = basefee_limit;
87 }
88 if let Some(queued_limit) = queued_limit {
89 config.queued_limit = queued_limit;
90 }
91 if let Some(blob_limit) = blob_limit {
92 config.blob_limit = blob_limit;
93 }
94 if let Some(max_account_slots) = max_account_slots {
95 config.max_account_slots = max_account_slots;
96 }
97 if let Some(minimal_protocol_basefee) = minimal_protocol_basefee {
98 config.minimal_protocol_basefee = minimal_protocol_basefee;
99 }
100 config.local_transactions_config.local_addresses.extend(local_addresses);
101
102 config
103 }
104}
105
106pub struct TxPoolBuilder<'a, Node: FullNodeTypes, V = ()> {
111 ctx: &'a BuilderContext<Node>,
112 validator: V,
113}
114
115impl<'a, Node: FullNodeTypes> TxPoolBuilder<'a, Node> {
116 pub const fn new(ctx: &'a BuilderContext<Node>) -> Self {
118 Self { ctx, validator: () }
119 }
120}
121
122impl<'a, Node: FullNodeTypes, V> TxPoolBuilder<'a, Node, V> {
123 pub fn with_validator<NewV>(self, validator: NewV) -> TxPoolBuilder<'a, Node, NewV> {
125 TxPoolBuilder { ctx: self.ctx, validator }
126 }
127}
128
129impl<'a, Node: FullNodeTypes, V> TxPoolBuilder<'a, Node, TransactionValidationTaskExecutor<V>>
130where
131 V: TransactionValidator + 'static,
132 V::Transaction:
133 PoolTransaction<Consensus = TxTy<Node::Types>> + reth_transaction_pool::EthPoolTransaction,
134{
135 pub fn build_and_spawn_maintenance_task(
138 self,
139 blob_store: DiskFileBlobStore,
140 pool_config: PoolConfig,
141 ) -> eyre::Result<
142 reth_transaction_pool::Pool<
143 TransactionValidationTaskExecutor<V>,
144 CoinbaseTipOrdering<V::Transaction>,
145 DiskFileBlobStore,
146 >,
147 > {
148 let TxPoolBuilder { ctx, validator, .. } = self;
150
151 let transaction_pool = reth_transaction_pool::Pool::new(
152 validator,
153 CoinbaseTipOrdering::default(),
154 blob_store,
155 pool_config.clone(),
156 );
157
158 spawn_maintenance_tasks(ctx, transaction_pool.clone(), &pool_config)?;
160
161 Ok(transaction_pool)
162 }
163}
164
165pub fn create_blob_store<Node: FullNodeTypes>(
167 ctx: &BuilderContext<Node>,
168) -> eyre::Result<DiskFileBlobStore> {
169 let data_dir = ctx.config().datadir();
170 Ok(reth_transaction_pool::blobstore::DiskFileBlobStore::open(
171 data_dir.blobstore(),
172 Default::default(),
173 )?)
174}
175
176pub fn create_blob_store_with_cache<Node: FullNodeTypes>(
178 ctx: &BuilderContext<Node>,
179 cache_size: Option<u32>,
180) -> eyre::Result<DiskFileBlobStore> {
181 let data_dir = ctx.config().datadir();
182 let config = if let Some(cache_size) = cache_size {
183 reth_transaction_pool::blobstore::DiskFileBlobStoreConfig::default()
184 .with_max_cached_entries(cache_size)
185 } else {
186 Default::default()
187 };
188
189 Ok(reth_transaction_pool::blobstore::DiskFileBlobStore::open(data_dir.blobstore(), config)?)
190}
191
192fn spawn_local_backup_task<Node, Pool>(ctx: &BuilderContext<Node>, pool: Pool) -> eyre::Result<()>
194where
195 Node: FullNodeTypes,
196 Pool: TransactionPool + Clone + 'static,
197{
198 if !ctx.config().txpool.disable_transactions_backup {
199 let data_dir = ctx.config().datadir();
200 let transactions_path = ctx
201 .config()
202 .txpool
203 .transactions_backup_path
204 .clone()
205 .unwrap_or_else(|| data_dir.txpool_transactions());
206
207 let transactions_backup_config =
208 reth_transaction_pool::maintain::LocalTransactionBackupConfig::with_local_txs_backup(
209 transactions_path,
210 );
211
212 ctx.task_executor().spawn_critical_with_graceful_shutdown_signal(
213 "local transactions backup task",
214 |shutdown| {
215 reth_transaction_pool::maintain::backup_local_transactions_task(
216 shutdown,
217 pool,
218 transactions_backup_config,
219 )
220 },
221 );
222 }
223 Ok(())
224}
225
226fn spawn_pool_maintenance_task<Node, Pool>(
228 ctx: &BuilderContext<Node>,
229 pool: Pool,
230 pool_config: &PoolConfig,
231) -> eyre::Result<()>
232where
233 Node: FullNodeTypes,
234 Pool: reth_transaction_pool::TransactionPoolExt + Clone + 'static,
235 Pool::Transaction: PoolTransaction<Consensus = TxTy<Node::Types>>,
236{
237 let chain_events = ctx.provider().canonical_state_stream();
238 let client = ctx.provider().clone();
239
240 ctx.task_executor().spawn_critical(
241 "txpool maintenance task",
242 reth_transaction_pool::maintain::maintain_transaction_pool_future(
243 client,
244 pool,
245 chain_events,
246 ctx.task_executor().clone(),
247 reth_transaction_pool::maintain::MaintainPoolConfig {
248 max_tx_lifetime: pool_config.max_queued_lifetime,
249 no_local_exemptions: pool_config.local_transactions_config.no_exemptions,
250 ..Default::default()
251 },
252 ),
253 );
254
255 Ok(())
256}
257
258fn spawn_maintenance_tasks<Node, Pool>(
260 ctx: &BuilderContext<Node>,
261 pool: Pool,
262 pool_config: &PoolConfig,
263) -> eyre::Result<()>
264where
265 Node: FullNodeTypes,
266 Pool: reth_transaction_pool::TransactionPoolExt + Clone + 'static,
267 Pool::Transaction: PoolTransaction<Consensus = TxTy<Node::Types>>,
268{
269 spawn_local_backup_task(ctx, pool.clone())?;
270 spawn_pool_maintenance_task(ctx, pool, pool_config)?;
271 Ok(())
272}
273
274impl<Node: FullNodeTypes, V: std::fmt::Debug> std::fmt::Debug for TxPoolBuilder<'_, Node, V> {
275 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
276 f.debug_struct("TxPoolBuilder").field("validator", &self.validator).finish()
277 }
278}
279
280#[cfg(test)]
281mod tests {
282 use super::*;
283 use reth_transaction_pool::PoolConfig;
284
285 #[test]
286 fn test_pool_builder_config_overrides_apply() {
287 let base_config = PoolConfig::default();
288 let overrides = PoolBuilderConfigOverrides {
289 pending_limit: Some(SubPoolLimit::default()),
290 max_account_slots: Some(100),
291 minimal_protocol_basefee: Some(1000),
292 ..Default::default()
293 };
294
295 let updated_config = overrides.apply(base_config);
296 assert_eq!(updated_config.max_account_slots, 100);
297 assert_eq!(updated_config.minimal_protocol_basefee, 1000);
298 }
299
300 #[test]
301 fn test_pool_builder_config_overrides_default() {
302 let overrides = PoolBuilderConfigOverrides::default();
303 assert!(overrides.pending_limit.is_none());
304 assert!(overrides.max_account_slots.is_none());
305 assert!(overrides.local_addresses.is_empty());
306 }
307}