1use crate::{
4 blobstore::{BlobStoreCanonTracker, BlobStoreUpdates},
5 error::PoolError,
6 metrics::MaintainPoolMetrics,
7 traits::{CanonicalStateUpdate, EthPoolTransaction, TransactionPool, TransactionPoolExt},
8 BlockInfo, PoolTransaction, PoolUpdateKind, TransactionOrigin,
9};
10use alloy_consensus::{BlockHeader, Typed2718};
11use alloy_eips::{BlockNumberOrTag, Decodable2718, Encodable2718};
12use alloy_primitives::{Address, BlockHash, BlockNumber};
13use alloy_rlp::{Bytes, Encodable};
14use futures_util::{
15 future::{BoxFuture, Fuse, FusedFuture},
16 FutureExt, Stream, StreamExt,
17};
18use reth_chain_state::CanonStateNotification;
19use reth_chainspec::{ChainSpecProvider, EthChainSpec};
20use reth_execution_types::ChangedAccount;
21use reth_fs_util::FsPathError;
22use reth_primitives_traits::{
23 transaction::signed::SignedTransaction, NodePrimitives, SealedHeader,
24};
25use reth_storage_api::{errors::provider::ProviderError, BlockReaderIdExt, StateProviderFactory};
26use reth_tasks::TaskSpawner;
27use serde::{Deserialize, Serialize};
28use std::{
29 borrow::Borrow,
30 collections::HashSet,
31 hash::{Hash, Hasher},
32 path::{Path, PathBuf},
33 sync::Arc,
34};
35use tokio::{
36 sync::oneshot,
37 time::{self, Duration},
38};
39use tracing::{debug, error, info, trace, warn};
40
41pub const MAX_QUEUED_TRANSACTION_LIFETIME: Duration = Duration::from_secs(3 * 60 * 60);
43
44#[derive(Debug, Clone, Copy, PartialEq, Eq)]
46pub struct MaintainPoolConfig {
47 pub max_update_depth: u64,
52 pub max_reload_accounts: usize,
56
57 pub max_tx_lifetime: Duration,
60
61 pub no_local_exemptions: bool,
67}
68
69impl Default for MaintainPoolConfig {
70 fn default() -> Self {
71 Self {
72 max_update_depth: 64,
73 max_reload_accounts: 100,
74 max_tx_lifetime: MAX_QUEUED_TRANSACTION_LIFETIME,
75 no_local_exemptions: false,
76 }
77 }
78}
79
80#[derive(Debug, Clone, Default)]
82pub struct LocalTransactionBackupConfig {
83 pub transactions_path: Option<PathBuf>,
85}
86
87impl LocalTransactionBackupConfig {
88 pub const fn with_local_txs_backup(transactions_path: PathBuf) -> Self {
90 Self { transactions_path: Some(transactions_path) }
91 }
92}
93
94pub fn maintain_transaction_pool_future<N, Client, P, St, Tasks>(
96 client: Client,
97 pool: P,
98 events: St,
99 task_spawner: Tasks,
100 config: MaintainPoolConfig,
101) -> BoxFuture<'static, ()>
102where
103 N: NodePrimitives,
104 Client: StateProviderFactory
105 + BlockReaderIdExt<Header = N::BlockHeader>
106 + ChainSpecProvider<ChainSpec: EthChainSpec<Header = N::BlockHeader>>
107 + Clone
108 + 'static,
109 P: TransactionPoolExt<Transaction: PoolTransaction<Consensus = N::SignedTx>> + 'static,
110 St: Stream<Item = CanonStateNotification<N>> + Send + Unpin + 'static,
111 Tasks: TaskSpawner + 'static,
112{
113 async move {
114 maintain_transaction_pool(client, pool, events, task_spawner, config).await;
115 }
116 .boxed()
117}
118
119pub async fn maintain_transaction_pool<N, Client, P, St, Tasks>(
123 client: Client,
124 pool: P,
125 mut events: St,
126 task_spawner: Tasks,
127 config: MaintainPoolConfig,
128) where
129 N: NodePrimitives,
130 Client: StateProviderFactory
131 + BlockReaderIdExt<Header = N::BlockHeader>
132 + ChainSpecProvider<ChainSpec: EthChainSpec<Header = N::BlockHeader>>
133 + Clone
134 + 'static,
135 P: TransactionPoolExt<Transaction: PoolTransaction<Consensus = N::SignedTx>> + 'static,
136 St: Stream<Item = CanonStateNotification<N>> + Send + Unpin + 'static,
137 Tasks: TaskSpawner + 'static,
138{
139 let metrics = MaintainPoolMetrics::default();
140 let MaintainPoolConfig { max_update_depth, max_reload_accounts, .. } = config;
141 if let Ok(Some(latest)) = client.header_by_number_or_tag(BlockNumberOrTag::Latest) {
143 let latest = SealedHeader::seal_slow(latest);
144 let chain_spec = client.chain_spec();
145 let info = BlockInfo {
146 block_gas_limit: latest.gas_limit(),
147 last_seen_block_hash: latest.hash(),
148 last_seen_block_number: latest.number(),
149 pending_basefee: chain_spec
150 .next_block_base_fee(latest.header(), latest.timestamp())
151 .unwrap_or_default(),
152 pending_blob_fee: latest
153 .maybe_next_block_blob_fee(chain_spec.blob_params_at_timestamp(latest.timestamp())),
154 };
155 pool.set_block_info(info);
156 }
157
158 let mut blob_store_tracker = BlobStoreCanonTracker::default();
160
161 let mut last_finalized_block =
163 FinalizedBlockTracker::new(client.finalized_block_number().ok().flatten());
164
165 let mut dirty_addresses = HashSet::default();
167
168 let mut maintained_state = MaintainedPoolState::InSync;
170
171 let mut reload_accounts_fut = Fuse::terminated();
173
174 let mut stale_eviction_interval = time::interval(config.max_tx_lifetime);
176
177 let mut first_event = true;
179
180 loop {
183 trace!(target: "txpool", state=?maintained_state, "awaiting new block or reorg");
184
185 metrics.set_dirty_accounts_len(dirty_addresses.len());
186 let pool_info = pool.block_info();
187
188 if maintained_state.is_drifted() {
192 metrics.inc_drift();
193 dirty_addresses = pool.unique_senders();
195 maintained_state = MaintainedPoolState::InSync;
197 }
198
199 if !dirty_addresses.is_empty() && reload_accounts_fut.is_terminated() {
201 let (tx, rx) = oneshot::channel();
202 let c = client.clone();
203 let at = pool_info.last_seen_block_hash;
204 let fut = if dirty_addresses.len() > max_reload_accounts {
205 let accs_to_reload =
207 dirty_addresses.iter().copied().take(max_reload_accounts).collect::<Vec<_>>();
208 for acc in &accs_to_reload {
209 dirty_addresses.remove(acc);
211 }
212 async move {
213 let res = load_accounts(c, at, accs_to_reload.into_iter());
214 let _ = tx.send(res);
215 }
216 .boxed()
217 } else {
218 let accs_to_reload = std::mem::take(&mut dirty_addresses);
220 async move {
221 let res = load_accounts(c, at, accs_to_reload.into_iter());
222 let _ = tx.send(res);
223 }
224 .boxed()
225 };
226 reload_accounts_fut = rx.fuse();
227 task_spawner.spawn_blocking(fut);
228 }
229
230 if let Some(finalized) =
232 last_finalized_block.update(client.finalized_block_number().ok().flatten())
233 {
234 if let BlobStoreUpdates::Finalized(blobs) =
235 blob_store_tracker.on_finalized_block(finalized)
236 {
237 metrics.inc_deleted_tracked_blobs(blobs.len());
238 pool.delete_blobs(blobs);
240 let pool = pool.clone();
242 task_spawner.spawn_blocking(Box::pin(async move {
243 debug!(target: "txpool", finalized_block = %finalized, "cleaning up blob store");
244 pool.cleanup_blobs();
245 }));
246 }
247 }
248
249 let mut event = None;
251 let mut reloaded = None;
252
253 tokio::select! {
256 res = &mut reload_accounts_fut => {
257 reloaded = Some(res);
258 }
259 ev = events.next() => {
260 if ev.is_none() {
261 break;
263 }
264 event = ev;
265 if first_event {
268 maintained_state = MaintainedPoolState::Drifted;
269 first_event = false
270 }
271 }
272 _ = stale_eviction_interval.tick() => {
273 let stale_txs: Vec<_> = pool
274 .queued_transactions()
275 .into_iter()
276 .filter(|tx| {
277 (tx.origin.is_external() || config.no_local_exemptions) && tx.timestamp.elapsed() > config.max_tx_lifetime
279 })
280 .map(|tx| *tx.hash())
281 .collect();
282 debug!(target: "txpool", count=%stale_txs.len(), "removing stale transactions");
283 pool.remove_transactions(stale_txs);
284 }
285 }
286 match reloaded {
288 Some(Ok(Ok(LoadedAccounts { accounts, failed_to_load }))) => {
289 dirty_addresses.extend(failed_to_load);
292 pool.update_accounts(accounts);
294 }
295 Some(Ok(Err(res))) => {
296 let (accs, err) = *res;
298 debug!(target: "txpool", %err, "failed to load accounts");
299 dirty_addresses.extend(accs);
300 }
301 Some(Err(_)) => {
302 maintained_state = MaintainedPoolState::Drifted;
304 }
305 None => {}
306 }
307
308 let Some(event) = event else { continue };
310 match event {
311 CanonStateNotification::Reorg { old, new } => {
312 let (old_blocks, old_state) = old.inner();
313 let (new_blocks, new_state) = new.inner();
314 let new_tip = new_blocks.tip();
315 let new_first = new_blocks.first();
316 let old_first = old_blocks.first();
317
318 if !(old_first.parent_hash() == pool_info.last_seen_block_hash ||
320 new_first.parent_hash() == pool_info.last_seen_block_hash)
321 {
322 maintained_state = MaintainedPoolState::Drifted;
324 }
325
326 let chain_spec = client.chain_spec();
327
328 let pending_block_base_fee = chain_spec
330 .next_block_base_fee(new_tip.header(), new_tip.timestamp())
331 .unwrap_or_default();
332 let pending_block_blob_fee = new_tip.header().maybe_next_block_blob_fee(
333 chain_spec.blob_params_at_timestamp(new_tip.timestamp()),
334 );
335
336 let new_changed_accounts: HashSet<_> =
338 new_state.changed_accounts().map(ChangedAccountEntry).collect();
339
340 let missing_changed_acc = old_state
342 .accounts_iter()
343 .map(|(a, _)| a)
344 .filter(|addr| !new_changed_accounts.contains(addr));
345
346 let mut changed_accounts =
348 match load_accounts(client.clone(), new_tip.hash(), missing_changed_acc) {
349 Ok(LoadedAccounts { accounts, failed_to_load }) => {
350 dirty_addresses.extend(failed_to_load);
352
353 accounts
354 }
355 Err(err) => {
356 let (addresses, err) = *err;
357 debug!(
358 target: "txpool",
359 %err,
360 "failed to load missing changed accounts at new tip: {:?}",
361 new_tip.hash()
362 );
363 dirty_addresses.extend(addresses);
364 vec![]
365 }
366 };
367
368 changed_accounts.extend(new_changed_accounts.into_iter().map(|entry| entry.0));
371
372 let new_mined_transactions: HashSet<_> = new_blocks.transaction_hashes().collect();
374
375 let pruned_old_transactions = old_blocks
378 .transactions_ecrecovered()
379 .filter(|tx| !new_mined_transactions.contains(tx.tx_hash()))
380 .filter_map(|tx| {
381 if tx.is_eip4844() {
382 pool.get_blob(*tx.tx_hash())
388 .ok()
389 .flatten()
390 .map(Arc::unwrap_or_clone)
391 .and_then(|sidecar| {
392 <P as TransactionPool>::Transaction::try_from_eip4844(
393 tx, sidecar,
394 )
395 })
396 } else {
397 <P as TransactionPool>::Transaction::try_from_consensus(tx).ok()
398 }
399 })
400 .collect::<Vec<_>>();
401
402 let update = CanonicalStateUpdate {
404 new_tip: new_tip.sealed_block(),
405 pending_block_base_fee,
406 pending_block_blob_fee,
407 changed_accounts,
408 mined_transactions: new_blocks.transaction_hashes().collect(),
410 update_kind: PoolUpdateKind::Reorg,
411 };
412 pool.on_canonical_state_change(update);
413
414 metrics.inc_reinserted_transactions(pruned_old_transactions.len());
421 let _ = pool.add_external_transactions(pruned_old_transactions).await;
422
423 blob_store_tracker.add_new_chain_blocks(&new_blocks);
425 }
426 CanonStateNotification::Commit { new } => {
427 let (blocks, state) = new.inner();
428 let tip = blocks.tip();
429 let chain_spec = client.chain_spec();
430
431 let pending_block_base_fee = chain_spec
433 .next_block_base_fee(tip.header(), tip.timestamp())
434 .unwrap_or_default();
435 let pending_block_blob_fee = tip.header().maybe_next_block_blob_fee(
436 chain_spec.blob_params_at_timestamp(tip.timestamp()),
437 );
438
439 let first_block = blocks.first();
440 trace!(
441 target: "txpool",
442 first = first_block.number(),
443 tip = tip.number(),
444 pool_block = pool_info.last_seen_block_number,
445 "update pool on new commit"
446 );
447
448 let depth = tip.number().abs_diff(pool_info.last_seen_block_number);
451 if depth > max_update_depth {
452 maintained_state = MaintainedPoolState::Drifted;
453 debug!(target: "txpool", ?depth, "skipping deep canonical update");
454 let info = BlockInfo {
455 block_gas_limit: tip.header().gas_limit(),
456 last_seen_block_hash: tip.hash(),
457 last_seen_block_number: tip.number(),
458 pending_basefee: pending_block_base_fee,
459 pending_blob_fee: pending_block_blob_fee,
460 };
461 pool.set_block_info(info);
462
463 blob_store_tracker.add_new_chain_blocks(&blocks);
465
466 continue
467 }
468
469 let mut changed_accounts = Vec::with_capacity(state.state().len());
470 for acc in state.changed_accounts() {
471 dirty_addresses.remove(&acc.address);
473 changed_accounts.push(acc);
474 }
475
476 let mined_transactions = blocks.transaction_hashes().collect();
477
478 if first_block.parent_hash() != pool_info.last_seen_block_hash {
480 maintained_state = MaintainedPoolState::Drifted;
484 }
485
486 let update = CanonicalStateUpdate {
488 new_tip: tip.sealed_block(),
489 pending_block_base_fee,
490 pending_block_blob_fee,
491 changed_accounts,
492 mined_transactions,
493 update_kind: PoolUpdateKind::Commit,
494 };
495 pool.on_canonical_state_change(update);
496
497 blob_store_tracker.add_new_chain_blocks(&blocks);
499 }
500 }
501 }
502}
503
504struct FinalizedBlockTracker {
505 last_finalized_block: Option<BlockNumber>,
506}
507
508impl FinalizedBlockTracker {
509 const fn new(last_finalized_block: Option<BlockNumber>) -> Self {
510 Self { last_finalized_block }
511 }
512
513 fn update(&mut self, finalized_block: Option<BlockNumber>) -> Option<BlockNumber> {
515 let finalized = finalized_block?;
516 self.last_finalized_block
517 .replace(finalized)
518 .is_none_or(|last| last < finalized)
519 .then_some(finalized)
520 }
521}
522
523#[derive(Debug, PartialEq, Eq)]
526enum MaintainedPoolState {
527 InSync,
529 Drifted,
531}
532
533impl MaintainedPoolState {
534 #[inline]
536 const fn is_drifted(&self) -> bool {
537 matches!(self, Self::Drifted)
538 }
539}
540
541#[derive(Eq)]
543struct ChangedAccountEntry(ChangedAccount);
544
545impl PartialEq for ChangedAccountEntry {
546 fn eq(&self, other: &Self) -> bool {
547 self.0.address == other.0.address
548 }
549}
550
551impl Hash for ChangedAccountEntry {
552 fn hash<H: Hasher>(&self, state: &mut H) {
553 self.0.address.hash(state);
554 }
555}
556
557impl Borrow<Address> for ChangedAccountEntry {
558 fn borrow(&self) -> &Address {
559 &self.0.address
560 }
561}
562
563#[derive(Default)]
564struct LoadedAccounts {
565 accounts: Vec<ChangedAccount>,
567 failed_to_load: Vec<Address>,
569}
570
571fn load_accounts<Client, I>(
577 client: Client,
578 at: BlockHash,
579 addresses: I,
580) -> Result<LoadedAccounts, Box<(HashSet<Address>, ProviderError)>>
581where
582 I: IntoIterator<Item = Address>,
583 Client: StateProviderFactory,
584{
585 let addresses = addresses.into_iter();
586 let mut res = LoadedAccounts::default();
587 let state = match client.history_by_block_hash(at) {
588 Ok(state) => state,
589 Err(err) => return Err(Box::new((addresses.collect(), err))),
590 };
591 for addr in addresses {
592 if let Ok(maybe_acc) = state.basic_account(&addr) {
593 let acc = maybe_acc
594 .map(|acc| ChangedAccount { address: addr, nonce: acc.nonce, balance: acc.balance })
595 .unwrap_or_else(|| ChangedAccount::empty(addr));
596 res.accounts.push(acc)
597 } else {
598 res.failed_to_load.push(addr);
600 }
601 }
602 Ok(res)
603}
604
605async fn load_and_reinsert_transactions<P>(
609 pool: P,
610 file_path: &Path,
611) -> Result<(), TransactionsBackupError>
612where
613 P: TransactionPool<Transaction: PoolTransaction<Consensus: SignedTransaction>>,
614{
615 if !file_path.exists() {
616 return Ok(())
617 }
618
619 debug!(target: "txpool", txs_file =?file_path, "Check local persistent storage for saved transactions");
620 let data = reth_fs_util::read(file_path)?;
621
622 if data.is_empty() {
623 return Ok(())
624 }
625
626 let pool_transactions: Vec<(TransactionOrigin, <P as TransactionPool>::Transaction)> =
627 if let Ok(tx_backups) = serde_json::from_slice::<Vec<TxBackup>>(&data) {
628 tx_backups
629 .into_iter()
630 .filter_map(|backup| {
631 let tx_signed = <P::Transaction as PoolTransaction>::Consensus::decode_2718(
632 &mut backup.rlp.as_ref(),
633 )
634 .ok()?;
635 let recovered = tx_signed.try_into_recovered().ok()?;
636 let pool_tx =
637 <P::Transaction as PoolTransaction>::try_from_consensus(recovered).ok()?;
638
639 Some((backup.origin, pool_tx))
640 })
641 .collect()
642 } else {
643 let txs_signed: Vec<<P::Transaction as PoolTransaction>::Consensus> =
644 alloy_rlp::Decodable::decode(&mut data.as_slice())?;
645
646 txs_signed
647 .into_iter()
648 .filter_map(|tx| tx.try_into_recovered().ok())
649 .filter_map(|tx| {
650 <P::Transaction as PoolTransaction>::try_from_consensus(tx)
651 .ok()
652 .map(|pool_tx| (TransactionOrigin::Local, pool_tx))
653 })
654 .collect()
655 };
656
657 let inserted = futures_util::future::join_all(
658 pool_transactions.into_iter().map(|(origin, tx)| pool.add_transaction(origin, tx)),
659 )
660 .await;
661
662 info!(target: "txpool", txs_file =?file_path, num_txs=%inserted.len(), "Successfully reinserted local transactions from file");
663 reth_fs_util::remove_file(file_path)?;
664 Ok(())
665}
666
667fn save_local_txs_backup<P>(pool: P, file_path: &Path)
668where
669 P: TransactionPool<Transaction: PoolTransaction<Consensus: Encodable>>,
670{
671 let local_transactions = pool.get_local_transactions();
672 if local_transactions.is_empty() {
673 trace!(target: "txpool", "no local transactions to save");
674 return
675 }
676
677 let local_transactions = local_transactions
678 .into_iter()
679 .map(|tx| {
680 let consensus_tx = tx.transaction.clone_into_consensus().into_inner();
681 let rlp_data = consensus_tx.encoded_2718();
682
683 TxBackup { rlp: rlp_data.into(), origin: tx.origin }
684 })
685 .collect::<Vec<_>>();
686
687 let json_data = match serde_json::to_string(&local_transactions) {
688 Ok(data) => data,
689 Err(err) => {
690 warn!(target: "txpool", %err, txs_file=?file_path, "failed to serialize local transactions to json");
691 return
692 }
693 };
694
695 info!(target: "txpool", txs_file =?file_path, num_txs=%local_transactions.len(), "Saving current local transactions");
696 let parent_dir = file_path.parent().map(std::fs::create_dir_all).transpose();
697
698 match parent_dir.map(|_| reth_fs_util::write(file_path, json_data)) {
699 Ok(_) => {
700 info!(target: "txpool", txs_file=?file_path, "Wrote local transactions to file");
701 }
702 Err(err) => {
703 warn!(target: "txpool", %err, txs_file=?file_path, "Failed to write local transactions to file");
704 }
705 }
706}
707
708#[derive(Debug, Deserialize, Serialize)]
711pub struct TxBackup {
712 pub rlp: Bytes,
714 pub origin: TransactionOrigin,
716}
717
718#[derive(thiserror::Error, Debug)]
720pub enum TransactionsBackupError {
721 #[error("failed to apply transactions backup. Encountered RLP decode error: {0}")]
723 Decode(#[from] alloy_rlp::Error),
724 #[error("failed to apply transactions backup. Encountered JSON decode error: {0}")]
726 Json(#[from] serde_json::Error),
727 #[error("failed to apply transactions backup. Encountered file error: {0}")]
729 FsPath(#[from] FsPathError),
730 #[error("failed to insert transactions to the transactions pool. Encountered pool error: {0}")]
732 Pool(#[from] PoolError),
733}
734
735pub async fn backup_local_transactions_task<P>(
738 shutdown: reth_tasks::shutdown::GracefulShutdown,
739 pool: P,
740 config: LocalTransactionBackupConfig,
741) where
742 P: TransactionPool<Transaction: PoolTransaction<Consensus: SignedTransaction>> + Clone,
743{
744 let Some(transactions_path) = config.transactions_path else {
745 return
747 };
748
749 if let Err(err) = load_and_reinsert_transactions(pool.clone(), &transactions_path).await {
750 error!(target: "txpool", "{}", err)
751 }
752
753 let graceful_guard = shutdown.await;
754
755 save_local_txs_backup(pool, &transactions_path);
757
758 drop(graceful_guard)
759}
760
761#[cfg(test)]
762mod tests {
763 use super::*;
764 use crate::{
765 blobstore::InMemoryBlobStore, validate::EthTransactionValidatorBuilder,
766 CoinbaseTipOrdering, EthPooledTransaction, Pool, TransactionOrigin,
767 };
768 use alloy_eips::eip2718::Decodable2718;
769 use alloy_primitives::{hex, U256};
770 use reth_ethereum_primitives::PooledTransactionVariant;
771 use reth_fs_util as fs;
772 use reth_provider::test_utils::{ExtendedAccount, MockEthProvider};
773 use reth_tasks::TaskManager;
774
775 #[test]
776 fn changed_acc_entry() {
777 let changed_acc = ChangedAccountEntry(ChangedAccount::empty(Address::random()));
778 let mut copy = changed_acc.0;
779 copy.nonce = 10;
780 assert!(changed_acc.eq(&ChangedAccountEntry(copy)));
781 }
782
783 const EXTENSION: &str = "json";
784 const FILENAME: &str = "test_transactions_backup";
785
786 #[tokio::test(flavor = "multi_thread")]
787 async fn test_save_local_txs_backup() {
788 let temp_dir = tempfile::tempdir().unwrap();
789 let transactions_path = temp_dir.path().join(FILENAME).with_extension(EXTENSION);
790 let tx_bytes = hex!(
791 "02f87201830655c2808505ef61f08482565f94388c818ca8b9251b393131c08a736a67ccb192978801049e39c4b5b1f580c001a01764ace353514e8abdfb92446de356b260e3c1225b73fc4c8876a6258d12a129a04f02294aa61ca7676061cd99f29275491218b4754b46a0248e5e42bc5091f507"
792 );
793 let tx = PooledTransactionVariant::decode_2718(&mut &tx_bytes[..]).unwrap();
794 let provider = MockEthProvider::default();
795 let transaction = EthPooledTransaction::from_pooled(tx.try_into_recovered().unwrap());
796 let tx_to_cmp = transaction.clone();
797 let sender = hex!("1f9090aaE28b8a3dCeaDf281B0F12828e676c326").into();
798 provider.add_account(sender, ExtendedAccount::new(42, U256::MAX));
799 let blob_store = InMemoryBlobStore::default();
800 let validator = EthTransactionValidatorBuilder::new(provider).build(blob_store.clone());
801
802 let txpool = Pool::new(
803 validator,
804 CoinbaseTipOrdering::default(),
805 blob_store.clone(),
806 Default::default(),
807 );
808
809 txpool.add_transaction(TransactionOrigin::Local, transaction.clone()).await.unwrap();
810
811 let handle = tokio::runtime::Handle::current();
812 let manager = TaskManager::new(handle);
813 let config = LocalTransactionBackupConfig::with_local_txs_backup(transactions_path.clone());
814 manager.executor().spawn_critical_with_graceful_shutdown_signal("test task", |shutdown| {
815 backup_local_transactions_task(shutdown, txpool.clone(), config)
816 });
817
818 let mut txns = txpool.get_local_transactions();
819 let tx_on_finish = txns.pop().expect("there should be 1 transaction");
820
821 assert_eq!(*tx_to_cmp.hash(), *tx_on_finish.hash());
822
823 manager.graceful_shutdown();
825
826 let data = fs::read(transactions_path).unwrap();
827
828 let txs: Vec<TxBackup> = serde_json::from_slice::<Vec<TxBackup>>(&data).unwrap();
829 assert_eq!(txs.len(), 1);
830
831 temp_dir.close().unwrap();
832 }
833
834 #[test]
835 fn test_update_with_higher_finalized_block() {
836 let mut tracker = FinalizedBlockTracker::new(Some(10));
837 assert_eq!(tracker.update(Some(15)), Some(15));
838 assert_eq!(tracker.last_finalized_block, Some(15));
839 }
840
841 #[test]
842 fn test_update_with_lower_finalized_block() {
843 let mut tracker = FinalizedBlockTracker::new(Some(20));
844 assert_eq!(tracker.update(Some(15)), None);
845 assert_eq!(tracker.last_finalized_block, Some(15));
846 }
847
848 #[test]
849 fn test_update_with_equal_finalized_block() {
850 let mut tracker = FinalizedBlockTracker::new(Some(20));
851 assert_eq!(tracker.update(Some(20)), None);
852 assert_eq!(tracker.last_finalized_block, Some(20));
853 }
854
855 #[test]
856 fn test_update_with_no_last_finalized_block() {
857 let mut tracker = FinalizedBlockTracker::new(None);
858 assert_eq!(tracker.update(Some(10)), Some(10));
859 assert_eq!(tracker.last_finalized_block, Some(10));
860 }
861
862 #[test]
863 fn test_update_with_no_new_finalized_block() {
864 let mut tracker = FinalizedBlockTracker::new(Some(10));
865 assert_eq!(tracker.update(None), None);
866 assert_eq!(tracker.last_finalized_block, Some(10));
867 }
868
869 #[test]
870 fn test_update_with_no_finalized_blocks() {
871 let mut tracker = FinalizedBlockTracker::new(None);
872 assert_eq!(tracker.update(None), None);
873 assert_eq!(tracker.last_finalized_block, None);
874 }
875}