1use crate::{
2 db_ext::DbTxPruneExt,
3 segments::{PruneInput, Segment, SegmentOutput},
4 PrunerError,
5};
6use alloy_primitives::TxNumber;
7use reth_db_api::{table::Value, tables, transaction::DbTxMut};
8use reth_primitives_traits::{NodePrimitives, SignedTransaction};
9use reth_provider::{
10 BlockReader, DBProvider, PruneCheckpointReader, RocksDBProviderFactory,
11 StaticFileProviderFactory, TransactionsProviderExt,
12};
13use reth_prune_types::{
14 PruneCheckpoint, PruneMode, PruneProgress, PrunePurpose, PruneSegment, SegmentOutputCheckpoint,
15};
16use reth_static_file_types::StaticFileSegment;
17use reth_storage_api::StorageSettingsCache;
18use tracing::{debug, instrument, trace};
19
20#[derive(Debug)]
21pub struct TransactionLookup {
22 mode: PruneMode,
23}
24
25impl TransactionLookup {
26 pub const fn new(mode: PruneMode) -> Self {
27 Self { mode }
28 }
29}
30
31impl<Provider> Segment<Provider> for TransactionLookup
32where
33 Provider: DBProvider<Tx: DbTxMut>
34 + BlockReader<Transaction: SignedTransaction>
35 + PruneCheckpointReader
36 + StorageSettingsCache
37 + RocksDBProviderFactory
38 + StaticFileProviderFactory<
39 Primitives: NodePrimitives<SignedTx: Value, Receipt: Value, BlockHeader: Value>,
40 >,
41{
42 fn segment(&self) -> PruneSegment {
43 PruneSegment::TransactionLookup
44 }
45
46 fn mode(&self) -> Option<PruneMode> {
47 Some(self.mode)
48 }
49
50 fn purpose(&self) -> PrunePurpose {
51 PrunePurpose::User
52 }
53
54 #[instrument(
55 name = "TransactionLookup::prune",
56 target = "pruner",
57 skip(self, provider),
58 ret(level = "trace")
59 )]
60 fn prune(
61 &self,
62 provider: &Provider,
63 mut input: PruneInput,
64 ) -> Result<SegmentOutput, PrunerError> {
65 if let Some(lowest_range) =
70 provider.static_file_provider().get_lowest_range(StaticFileSegment::Transactions) &&
71 input
72 .previous_checkpoint
73 .is_none_or(|checkpoint| checkpoint.block_number < Some(lowest_range.start())) &&
74 let Some(previous_block) = lowest_range.start().checked_sub(1) &&
75 let Some(body_indices) = provider.block_body_indices(previous_block)? &&
76 let Some(tx_number) = body_indices.next_tx_num().checked_sub(1)
77 {
78 input.previous_checkpoint = Some(PruneCheckpoint {
80 block_number: Some(previous_block),
81 tx_number: Some(tx_number),
82 prune_mode: self.mode,
83 });
84 debug!(
85 target: "pruner",
86 static_file_checkpoint = ?input.previous_checkpoint,
87 "Using static file transaction checkpoint as TransactionLookup starting point"
88 );
89 }
90
91 let (start, end) = match input.get_next_tx_num_range(provider)? {
92 Some(range) => range,
93 None => {
94 trace!(target: "pruner", "No transaction lookup entries to prune");
95 return Ok(SegmentOutput::done())
96 }
97 }
98 .into_inner();
99
100 if provider.cached_storage_settings().storage_v2 {
102 return self.prune_rocksdb(provider, input, start, end);
103 }
104
105 if self.mode.is_full() {
107 let pruned = provider.tx_ref().clear_table::<tables::TransactionHashNumbers>()?;
108 trace!(target: "pruner", %pruned, "Cleared transaction lookup table");
109
110 let last_pruned_block = provider
111 .block_by_transaction_id(end)?
112 .ok_or(PrunerError::InconsistentData("Block for transaction is not found"))?;
113
114 return Ok(SegmentOutput {
115 progress: PruneProgress::Finished,
116 pruned,
117 checkpoint: Some(SegmentOutputCheckpoint {
118 block_number: Some(last_pruned_block),
119 tx_number: Some(end),
120 }),
121 });
122 }
123
124 let tx_range = start..=
125 Some(end)
126 .min(
127 input
128 .limiter
129 .deleted_entries_limit_left()
130 .map(|left| start.saturating_add(left as u64) - 1),
133 )
134 .unwrap();
135 let tx_range_end = *tx_range.end();
136
137 let mut hashes = provider
138 .static_file_provider()
139 .transaction_hashes_by_range(
140 *tx_range.start()..
141 tx_range_end
142 .checked_add(1)
143 .ok_or(PrunerError::InconsistentData("Transaction range end overflow"))?,
144 )?
145 .into_iter()
146 .map(|(hash, _)| hash)
147 .collect::<Vec<_>>();
148
149 hashes.sort_unstable();
153
154 let tx_count = tx_range.count();
156 if hashes.len() != tx_count {
157 return Err(PrunerError::InconsistentData(
158 "Unexpected number of transaction hashes retrieved by transaction number range",
159 ))
160 }
161
162 let mut limiter = input.limiter;
163
164 let mut last_pruned_transaction = None;
165 let (pruned, done) =
166 provider.tx_ref().prune_table_with_iterator::<tables::TransactionHashNumbers>(
167 hashes,
168 &mut limiter,
169 |row| {
170 last_pruned_transaction =
171 Some(last_pruned_transaction.unwrap_or(row.1).max(row.1))
172 },
173 )?;
174
175 let done = done && tx_range_end == end;
176 trace!(target: "pruner", %pruned, %done, "Pruned transaction lookup");
177
178 let last_pruned_transaction = last_pruned_transaction.unwrap_or(tx_range_end);
179
180 let last_pruned_block = provider
181 .block_by_transaction_id(last_pruned_transaction)?
182 .ok_or(PrunerError::InconsistentData("Block for transaction is not found"))?
183 .checked_sub(if done { 0 } else { 1 });
187
188 let progress = limiter.progress(done);
189
190 Ok(SegmentOutput {
191 progress,
192 pruned,
193 checkpoint: Some(SegmentOutputCheckpoint {
194 block_number: last_pruned_block,
195 tx_number: Some(last_pruned_transaction),
196 }),
197 })
198 }
199}
200
201impl TransactionLookup {
202 fn prune_rocksdb<Provider>(
207 &self,
208 provider: &Provider,
209 input: PruneInput,
210 start: TxNumber,
211 end: TxNumber,
212 ) -> Result<SegmentOutput, PrunerError>
213 where
214 Provider: DBProvider
215 + BlockReader<Transaction: SignedTransaction>
216 + RocksDBProviderFactory
217 + StaticFileProviderFactory<
218 Primitives: NodePrimitives<SignedTx: Value, Receipt: Value, BlockHeader: Value>,
219 >,
220 {
221 if self.mode.is_full() {
223 let rocksdb = provider.rocksdb_provider();
224 rocksdb.clear::<tables::TransactionHashNumbers>()?;
225 trace!(target: "pruner", "Cleared transaction lookup table (RocksDB)");
226
227 let last_pruned_block = provider
228 .block_by_transaction_id(end)?
229 .ok_or(PrunerError::InconsistentData("Block for transaction is not found"))?;
230
231 return Ok(SegmentOutput {
232 progress: PruneProgress::Finished,
233 pruned: 0, checkpoint: Some(SegmentOutputCheckpoint {
235 block_number: Some(last_pruned_block),
236 tx_number: Some(end),
237 }),
238 });
239 }
240
241 let tx_range_end = input
242 .limiter
243 .deleted_entries_limit_left()
244 .map(|left| start.saturating_add(left as u64).saturating_sub(1))
245 .map_or(end, |limited| limited.min(end));
246 let tx_range = start..=tx_range_end;
247
248 let hashes = provider.static_file_provider().transaction_hashes_by_range(
249 *tx_range.start()..
250 tx_range_end
251 .checked_add(1)
252 .ok_or(PrunerError::InconsistentData("Transaction range end overflow"))?,
253 )?;
254
255 let tx_count = tx_range.count();
257 if hashes.len() != tx_count {
258 return Err(PrunerError::InconsistentData(
259 "Unexpected number of transaction hashes retrieved by transaction number range",
260 ))
261 }
262
263 let mut limiter = input.limiter;
264
265 let mut deleted = 0usize;
267 provider.with_rocksdb_batch(|mut batch| {
268 for (hash, _) in &hashes {
269 if limiter.is_limit_reached() {
270 break;
271 }
272 batch.delete::<tables::TransactionHashNumbers>(*hash)?;
273 limiter.increment_deleted_entries_count();
274 deleted += 1;
275 }
276 Ok(((), Some(batch.into_inner())))
277 })?;
278
279 let done = deleted == hashes.len() && tx_range_end == end;
280 trace!(target: "pruner", %deleted, %done, "Pruned transaction lookup (RocksDB)");
281
282 let last_pruned_transaction =
283 if deleted > 0 { start + deleted as u64 - 1 } else { tx_range_end };
284
285 let last_pruned_block = provider
286 .block_by_transaction_id(last_pruned_transaction)?
287 .ok_or(PrunerError::InconsistentData("Block for transaction is not found"))?
288 .checked_sub(if done { 0 } else { 1 });
289
290 let progress = limiter.progress(done);
291
292 Ok(SegmentOutput {
293 progress,
294 pruned: deleted,
295 checkpoint: Some(SegmentOutputCheckpoint {
296 block_number: last_pruned_block,
297 tx_number: Some(last_pruned_transaction),
298 }),
299 })
300 }
301}
302
303#[cfg(test)]
304mod tests {
305 use crate::segments::{PruneInput, PruneLimiter, Segment, SegmentOutput, TransactionLookup};
306 use alloy_primitives::{BlockNumber, TxNumber, B256};
307 use assert_matches::assert_matches;
308 use itertools::{
309 FoldWhile::{Continue, Done},
310 Itertools,
311 };
312 use reth_db_api::{models::StorageSettings, tables};
313 use reth_provider::{
314 DBProvider, DatabaseProviderFactory, PruneCheckpointReader, RocksDBProviderFactory,
315 StaticFileProviderFactory, StaticFileWriter,
316 };
317 use reth_prune_types::{
318 PruneCheckpoint, PruneInterruptReason, PruneMode, PruneProgress, PruneSegment,
319 };
320 use reth_stages::test_utils::{StorageKind, TestStageDB};
321 use reth_static_file_types::{StaticFileSegment, DEFAULT_BLOCKS_PER_STATIC_FILE};
322 use reth_storage_api::StorageSettingsCache;
323 use reth_testing_utils::generators::{self, random_block_range, BlockRangeParams};
324 use std::ops::Sub;
325
326 #[test]
327 fn prune_transaction_zero_after_empty_genesis() {
328 let db = TestStageDB::default();
329 let mut rng = generators::rng();
330 let blocks = [
331 random_block_range(
332 &mut rng,
333 0..=0,
334 BlockRangeParams { tx_count: 0..1, ..Default::default() },
335 ),
336 random_block_range(
337 &mut rng,
338 1..=2,
339 BlockRangeParams { tx_count: 1..2, ..Default::default() },
340 ),
341 ]
342 .concat();
343 db.insert_blocks(blocks.iter(), StorageKind::Static).unwrap();
344 let hashes = blocks
345 .iter()
346 .flat_map(|block| &block.body().transactions)
347 .map(|tx| *tx.tx_hash())
348 .collect::<Vec<_>>();
349 db.insert_tx_hash_numbers(hashes.iter().enumerate().map(|(id, hash)| (*hash, id as u64)))
350 .unwrap();
351
352 let provider = db.factory.database_provider_rw().unwrap();
353 let result = TransactionLookup::new(PruneMode::Before(2))
354 .prune(
355 &provider,
356 PruneInput {
357 previous_checkpoint: None,
358 to_block: 1,
359 limiter: PruneLimiter::default().set_deleted_entries_limit(10),
360 },
361 )
362 .unwrap();
363 assert_eq!(result.pruned, 1);
364 assert_eq!(result.checkpoint.unwrap().tx_number, Some(0));
365 provider.commit().unwrap();
366 assert_eq!(db.table::<tables::TransactionHashNumbers>().unwrap(), vec![(hashes[1], 1)]);
367 }
368
369 #[test]
370 fn prune_transaction_zero_when_earlier_blocks_are_empty() {
371 let db = TestStageDB::default();
372 let mut rng = generators::rng();
373 let first_retained_block = DEFAULT_BLOCKS_PER_STATIC_FILE;
374 let blocks = [
375 random_block_range(
376 &mut rng,
377 first_retained_block - 1..=first_retained_block - 1,
378 BlockRangeParams { tx_count: 0..1, ..Default::default() },
379 ),
380 random_block_range(
381 &mut rng,
382 first_retained_block..=first_retained_block + 1,
383 BlockRangeParams { tx_count: 1..2, ..Default::default() },
384 ),
385 ]
386 .concat();
387 db.insert_blocks(blocks.iter(), StorageKind::Database(None)).unwrap();
388
389 let static_files = db.factory.static_file_provider();
391 {
392 let mut writer = static_files
393 .get_writer(first_retained_block, StaticFileSegment::Transactions)
394 .unwrap();
395 for (tx_number, block) in blocks[1..].iter().enumerate() {
396 writer.append_transaction(tx_number as u64, &block.body().transactions[0]).unwrap();
397 writer.increment_block(block.number).unwrap();
398 }
399 writer.commit().unwrap();
400 }
401 assert_eq!(
402 static_files.get_lowest_range(StaticFileSegment::Transactions).unwrap().start(),
403 first_retained_block,
404 );
405
406 let hashes = blocks[1..]
407 .iter()
408 .map(|block| *block.body().transactions[0].tx_hash())
409 .collect::<Vec<_>>();
410 db.insert_tx_hash_numbers(hashes.iter().enumerate().map(|(id, hash)| (*hash, id as u64)))
411 .unwrap();
412
413 let provider = db.factory.database_provider_rw().unwrap();
414 let result = TransactionLookup::new(PruneMode::Before(first_retained_block + 1))
415 .prune(
416 &provider,
417 PruneInput {
418 previous_checkpoint: None,
419 to_block: first_retained_block,
420 limiter: PruneLimiter::default().set_deleted_entries_limit(10),
421 },
422 )
423 .unwrap();
424 assert!(result.progress.is_finished());
425 assert_eq!(result.pruned, 1);
426 let checkpoint = result.checkpoint.unwrap();
427 assert_eq!(checkpoint.block_number, Some(first_retained_block));
428 assert_eq!(checkpoint.tx_number, Some(0));
429 provider.commit().unwrap();
430 assert_eq!(db.table::<tables::TransactionHashNumbers>().unwrap(), vec![(hashes[1], 1)]);
431 }
432
433 #[test]
434 fn prune() {
435 let db = TestStageDB::default();
436 let mut rng = generators::rng();
437
438 let blocks = random_block_range(
439 &mut rng,
440 1..=10,
441 BlockRangeParams { parent: Some(B256::ZERO), tx_count: 2..3, ..Default::default() },
442 );
443 db.insert_blocks(blocks.iter(), StorageKind::Static).expect("insert blocks");
444
445 let mut tx_hash_numbers = Vec::new();
446 for block in &blocks {
447 tx_hash_numbers.reserve_exact(block.transaction_count());
448 for transaction in &block.body().transactions {
449 tx_hash_numbers.push((*transaction.tx_hash(), tx_hash_numbers.len() as u64));
450 }
451 }
452 let tx_hash_numbers_len = tx_hash_numbers.len();
453 db.insert_tx_hash_numbers(tx_hash_numbers).expect("insert tx hash numbers");
454
455 assert_eq!(
456 db.count_entries::<tables::Transactions>().unwrap(),
457 blocks.iter().map(|block| block.transaction_count()).sum::<usize>()
458 );
459 assert_eq!(
460 db.count_entries::<tables::Transactions>().unwrap(),
461 db.table::<tables::TransactionHashNumbers>().unwrap().len()
462 );
463
464 let test_prune = |to_block: BlockNumber, expected_result: (PruneProgress, usize)| {
465 let prune_mode = PruneMode::Before(to_block);
466 let segment = TransactionLookup::new(prune_mode);
467 let mut limiter = PruneLimiter::default().set_deleted_entries_limit(10);
468 let input = PruneInput {
469 previous_checkpoint: db
470 .factory
471 .provider()
472 .unwrap()
473 .get_prune_checkpoint(PruneSegment::TransactionLookup)
474 .unwrap(),
475 to_block,
476 limiter: limiter.clone(),
477 };
478
479 let next_tx_number_to_prune = db
480 .factory
481 .provider()
482 .unwrap()
483 .get_prune_checkpoint(PruneSegment::TransactionLookup)
484 .unwrap()
485 .and_then(|checkpoint| checkpoint.tx_number)
486 .map(|tx_number| tx_number + 1)
487 .unwrap_or_default();
488
489 let last_pruned_tx_number = blocks
490 .iter()
491 .take(to_block as usize)
492 .map(|block| block.transaction_count())
493 .sum::<usize>()
494 .min(
495 next_tx_number_to_prune as usize +
496 input.limiter.deleted_entries_limit().unwrap(),
497 )
498 .sub(1);
499
500 let last_pruned_block_number = blocks
501 .iter()
502 .fold_while((0, 0), |(_, mut tx_count), block| {
503 tx_count += block.transaction_count();
504
505 if tx_count > last_pruned_tx_number {
506 Done((block.number, tx_count))
507 } else {
508 Continue((block.number, tx_count))
509 }
510 })
511 .into_inner()
512 .0;
513
514 let provider = db.factory.database_provider_rw().unwrap();
515 let result = segment.prune(&provider, input).unwrap();
516 limiter.increment_deleted_entries_count_by(result.pruned);
517
518 assert_matches!(
519 result,
520 SegmentOutput {progress, pruned, checkpoint: Some(_)}
521 if (progress, pruned) == expected_result
522 );
523
524 segment
525 .save_checkpoint(
526 &provider,
527 result.checkpoint.unwrap().as_prune_checkpoint(prune_mode),
528 )
529 .unwrap();
530 provider.commit().expect("commit");
531
532 let last_pruned_block_number = last_pruned_block_number
533 .checked_sub(if result.progress.is_finished() { 0 } else { 1 });
534
535 assert_eq!(
536 db.table::<tables::TransactionHashNumbers>().unwrap().len(),
537 tx_hash_numbers_len - (last_pruned_tx_number + 1)
538 );
539 assert_eq!(
540 db.factory
541 .provider()
542 .unwrap()
543 .get_prune_checkpoint(PruneSegment::TransactionLookup)
544 .unwrap(),
545 Some(PruneCheckpoint {
546 block_number: last_pruned_block_number,
547 tx_number: Some(last_pruned_tx_number as TxNumber),
548 prune_mode
549 })
550 );
551 };
552
553 test_prune(
554 6,
555 (PruneProgress::HasMoreData(PruneInterruptReason::DeletedEntriesLimitReached), 10),
556 );
557 test_prune(6, (PruneProgress::Finished, 2));
558 test_prune(10, (PruneProgress::Finished, 8));
559 }
560
561 #[test]
562 fn prune_rocksdb() {
563 let db = TestStageDB::default();
564 let mut rng = generators::rng();
565
566 let blocks = random_block_range(
567 &mut rng,
568 1..=10,
569 BlockRangeParams { parent: Some(B256::ZERO), tx_count: 2..3, ..Default::default() },
570 );
571 db.insert_blocks(blocks.iter(), StorageKind::Static).expect("insert blocks");
572
573 let mut tx_hash_numbers = Vec::new();
575 for block in &blocks {
576 tx_hash_numbers.reserve_exact(block.transaction_count());
577 for transaction in &block.body().transactions {
578 tx_hash_numbers.push((*transaction.tx_hash(), tx_hash_numbers.len() as u64));
579 }
580 }
581 let tx_hash_numbers_len = tx_hash_numbers.len();
582
583 {
585 let rocksdb = db.factory.rocksdb_provider();
586 let mut batch = rocksdb.batch();
587 for (hash, tx_num) in &tx_hash_numbers {
588 batch.put::<tables::TransactionHashNumbers>(*hash, tx_num).unwrap();
589 }
590 batch.commit().expect("commit rocksdb batch");
591 }
592
593 {
595 let rocksdb = db.factory.rocksdb_provider();
596 for (hash, expected_tx_num) in &tx_hash_numbers {
597 let actual = rocksdb.get::<tables::TransactionHashNumbers>(*hash).unwrap();
598 assert_eq!(actual, Some(*expected_tx_num));
599 }
600 }
601
602 let to_block: BlockNumber = 6;
603 let prune_mode = PruneMode::Before(to_block);
604 let input =
605 PruneInput { previous_checkpoint: None, to_block, limiter: PruneLimiter::default() };
606 let segment = TransactionLookup::new(prune_mode);
607
608 db.factory.set_storage_settings_cache(StorageSettings::v2());
610
611 let provider = db.factory.database_provider_rw().unwrap();
612 let result = segment.prune(&provider, input).unwrap();
613 provider.commit().expect("commit");
614
615 assert_matches!(
616 result,
617 SegmentOutput { progress: PruneProgress::Finished, pruned, checkpoint: Some(_) }
618 if pruned > 0
619 );
620
621 let txs_up_to_block_6: usize = blocks.iter().take(6).map(|b| b.transaction_count()).sum();
623
624 {
626 let rocksdb = db.factory.rocksdb_provider();
627 for (i, (hash, _)) in tx_hash_numbers.iter().enumerate() {
628 let entry = rocksdb.get::<tables::TransactionHashNumbers>(*hash).unwrap();
629 if i < txs_up_to_block_6 {
630 assert!(entry.is_none(), "Entry {} (hash {:?}) should be pruned", i, hash);
631 } else {
632 assert!(entry.is_some(), "Entry {} (hash {:?}) should still exist", i, hash);
633 }
634 }
635 }
636
637 {
639 let rocksdb = db.factory.rocksdb_provider();
640 let remaining: Vec<_> =
641 rocksdb.iter::<tables::TransactionHashNumbers>().unwrap().collect();
642 assert_eq!(
643 remaining.len(),
644 tx_hash_numbers_len - txs_up_to_block_6,
645 "Remaining RocksDB entries should match expected"
646 );
647 }
648 }
649
650 #[test]
658 fn prune_rocksdb_zero_deleted_checkpoint() {
659 let db = TestStageDB::default();
660 let mut rng = generators::rng();
661
662 let blocks = random_block_range(
663 &mut rng,
664 1..=10,
665 BlockRangeParams { parent: Some(B256::ZERO), tx_count: 2..3, ..Default::default() },
666 );
667 db.insert_blocks(blocks.iter(), StorageKind::Static).expect("insert blocks");
668
669 let mut tx_hash_numbers = Vec::new();
671 for block in &blocks {
672 tx_hash_numbers.reserve_exact(block.transaction_count());
673 for transaction in &block.body().transactions {
674 tx_hash_numbers.push((*transaction.tx_hash(), tx_hash_numbers.len() as u64));
675 }
676 }
677
678 {
680 let rocksdb = db.factory.rocksdb_provider();
681 let mut batch = rocksdb.batch();
682 for (hash, tx_num) in &tx_hash_numbers {
683 batch.put::<tables::TransactionHashNumbers>(*hash, tx_num).unwrap();
684 }
685 batch.commit().expect("commit rocksdb batch");
686 }
687
688 db.factory.set_storage_settings_cache(StorageSettings::v2());
690
691 let to_block: BlockNumber = 6;
692 let prune_mode = PruneMode::Before(to_block);
693
694 let previous_checkpoint =
696 Some(PruneCheckpoint { block_number: Some(2), tx_number: Some(5), prune_mode });
697
698 let mut limiter = PruneLimiter::default().set_deleted_entries_limit(1);
701 limiter.increment_deleted_entries_count(); let input = PruneInput { previous_checkpoint, to_block, limiter };
704 let segment = TransactionLookup::new(prune_mode);
705
706 let provider = db.factory.database_provider_rw().unwrap();
707 let result = segment.prune(&provider, input).unwrap();
708 provider.commit().expect("commit");
709
710 assert_eq!(result.pruned, 0, "Nothing should be pruned with exhausted limit");
712
713 if let Some(checkpoint) = &result.checkpoint {
717 assert_eq!(
718 checkpoint.tx_number,
719 Some(5),
720 "Checkpoint should stay at 5 (previous), not advance to 6 (start)"
721 );
722 }
723
724 {
726 let rocksdb = db.factory.rocksdb_provider();
727 let remaining: Vec<_> =
728 rocksdb.iter::<tables::TransactionHashNumbers>().unwrap().collect();
729 assert_eq!(
730 remaining.len(),
731 tx_hash_numbers.len(),
732 "All RocksDB entries should still exist"
733 );
734 }
735 }
736}