reth_prune/segments/user/
history.rs

1use crate::PruneLimiter;
2use alloy_primitives::BlockNumber;
3use itertools::Itertools;
4use reth_db_api::{
5    cursor::{DbCursorRO, DbCursorRW},
6    models::ShardedKey,
7    table::Table,
8    transaction::DbTxMut,
9    BlockNumberList, DatabaseError, RawKey, RawTable, RawValue,
10};
11use reth_provider::DBProvider;
12use reth_prune_types::{SegmentOutput, SegmentOutputCheckpoint};
13use rustc_hash::FxHashMap;
14
15enum PruneShardOutcome {
16    Deleted,
17    Updated,
18    Unchanged,
19}
20
21#[derive(Debug, Default)]
22pub(crate) struct PrunedIndices {
23    pub(crate) deleted: usize,
24    pub(crate) updated: usize,
25    pub(crate) unchanged: usize,
26}
27
28/// Result of pruning history changesets, used to build the final output.
29pub(crate) struct HistoryPruneResult<K> {
30    /// Map of the highest deleted changeset keys to their block numbers.
31    pub(crate) highest_deleted: FxHashMap<K, BlockNumber>,
32    /// The highest block number whose changesets are fully pruned, becoming the checkpoint.
33    ///
34    /// Checkpoints have block granularity, so a caller that can stop in the middle of a block
35    /// must report the block before it, and prune the interrupted block again on the next run.
36    pub(crate) last_pruned_block: Option<BlockNumber>,
37    /// Number of changesets pruned.
38    pub(crate) pruned_count: usize,
39    /// Whether pruning is complete.
40    pub(crate) done: bool,
41}
42
43/// Finalizes history pruning by sorting sharded keys, pruning history indices, and building output.
44///
45/// This is shared between static file and database pruning for both account and storage history.
46pub(crate) fn finalize_history_prune<Provider, T, K, SK>(
47    provider: &Provider,
48    result: HistoryPruneResult<K>,
49    range_end: BlockNumber,
50    limiter: &PruneLimiter,
51    to_sharded_key: impl Fn(K, BlockNumber) -> T::Key,
52    key_matches: impl Fn(&T::Key, &T::Key) -> bool,
53) -> Result<SegmentOutput, DatabaseError>
54where
55    Provider: DBProvider<Tx: DbTxMut>,
56    T: Table<Value = BlockNumberList>,
57    T::Key: AsRef<ShardedKey<SK>>,
58    K: Ord,
59{
60    let HistoryPruneResult { highest_deleted, last_pruned_block, pruned_count, done } = result;
61
62    // Nothing was pruned only when the range held no changesets at all, so the whole range is
63    // done.
64    let last_changeset_pruned_block = last_pruned_block.unwrap_or(range_end);
65
66    // Sort highest deleted block numbers and turn them into sharded keys.
67    // We use `sorted_unstable` because no equal keys exist in the map.
68    let highest_sharded_keys =
69        highest_deleted.into_iter().sorted_unstable().map(|(key, block_number)| {
70            to_sharded_key(key, block_number.min(last_changeset_pruned_block))
71        });
72
73    let outcomes =
74        prune_history_indices::<Provider, T, _>(provider, highest_sharded_keys, key_matches)?;
75
76    let progress = limiter.progress(done);
77
78    Ok(SegmentOutput {
79        progress,
80        pruned: pruned_count + outcomes.deleted,
81        checkpoint: Some(SegmentOutputCheckpoint {
82            block_number: Some(last_changeset_pruned_block),
83            tx_number: None,
84        }),
85    })
86}
87
88/// Prune history indices according to the provided list of highest sharded keys.
89///
90/// Returns total number of deleted, updated and unchanged entities.
91pub(crate) fn prune_history_indices<Provider, T, SK>(
92    provider: &Provider,
93    highest_sharded_keys: impl IntoIterator<Item = T::Key>,
94    key_matches: impl Fn(&T::Key, &T::Key) -> bool,
95) -> Result<PrunedIndices, DatabaseError>
96where
97    Provider: DBProvider<Tx: DbTxMut>,
98    T: Table<Value = BlockNumberList>,
99    T::Key: AsRef<ShardedKey<SK>>,
100{
101    let mut outcomes = PrunedIndices::default();
102    let mut cursor = provider.tx_ref().cursor_write::<RawTable<T>>()?;
103
104    for sharded_key in highest_sharded_keys {
105        // Seek to the shard that has the key >= the given sharded key
106        // TODO: optimize
107        let mut shard = cursor.seek(RawKey::new(sharded_key.clone()))?;
108
109        // Get the highest block number that needs to be deleted for this sharded key
110        let to_block = sharded_key.as_ref().highest_block_number;
111
112        'shard: loop {
113            let Some((key, block_nums)) =
114                shard.map(|(k, v)| Result::<_, DatabaseError>::Ok((k.key()?, v))).transpose()?
115            else {
116                break
117            };
118
119            if key_matches(&key, &sharded_key) {
120                match prune_shard(&mut cursor, key, block_nums, to_block, &key_matches)? {
121                    PruneShardOutcome::Deleted => outcomes.deleted += 1,
122                    PruneShardOutcome::Updated => outcomes.updated += 1,
123                    PruneShardOutcome::Unchanged => outcomes.unchanged += 1,
124                }
125            } else {
126                // If such shard doesn't exist, skip to the next sharded key
127                break 'shard
128            }
129
130            shard = cursor.next()?;
131        }
132    }
133
134    Ok(outcomes)
135}
136
137/// Prunes one shard of a history table.
138///
139/// 1. If the shard has `highest_block_number` less than or equal to the target block number for
140///    pruning, delete the shard completely.
141/// 2. If the shard has `highest_block_number` greater than the target block number for pruning,
142///    filter block numbers inside the shard which are less than the target block number for
143///    pruning.
144fn prune_shard<C, T, SK>(
145    cursor: &mut C,
146    key: T::Key,
147    raw_blocks: RawValue<T::Value>,
148    to_block: BlockNumber,
149    key_matches: impl Fn(&T::Key, &T::Key) -> bool,
150) -> Result<PruneShardOutcome, DatabaseError>
151where
152    C: DbCursorRO<RawTable<T>> + DbCursorRW<RawTable<T>>,
153    T: Table<Value = BlockNumberList>,
154    T::Key: AsRef<ShardedKey<SK>>,
155{
156    // If shard consists only of block numbers less than the target one, delete shard
157    // completely.
158    if key.as_ref().highest_block_number <= to_block {
159        cursor.delete_current()?;
160        Ok(PruneShardOutcome::Deleted)
161    }
162    // Shard contains block numbers that are higher than the target one, so we need to
163    // filter it. It is guaranteed that further shards for this sharded key will not
164    // contain the target block number, as it's in this shard.
165    else {
166        let blocks = raw_blocks.value()?;
167        let higher_blocks =
168            blocks.iter().skip_while(|block| *block <= to_block).collect::<Vec<_>>();
169
170        // If there were blocks less than or equal to the target one
171        // (so the shard has changed), update the shard.
172        if blocks.len() as usize == higher_blocks.len() {
173            return Ok(PruneShardOutcome::Unchanged);
174        }
175
176        // If there will be no more blocks in the shard after pruning blocks below target
177        // block, we need to remove it, as empty shards are not allowed.
178        if higher_blocks.is_empty() {
179            if key.as_ref().highest_block_number == u64::MAX {
180                let prev_row = cursor
181                    .prev()?
182                    .map(|(k, v)| Result::<_, DatabaseError>::Ok((k.key()?, v)))
183                    .transpose()?;
184                match prev_row {
185                    // If current shard is the last shard for the sharded key that
186                    // has previous shards, replace it with the previous shard.
187                    Some((prev_key, prev_value)) if key_matches(&prev_key, &key) => {
188                        cursor.delete_current()?;
189                        // Upsert will replace the last shard for this sharded key with
190                        // the previous value.
191                        cursor.upsert(RawKey::new(key), &prev_value)?;
192                        Ok(PruneShardOutcome::Updated)
193                    }
194                    // If there's no previous shard for this sharded key,
195                    // just delete last shard completely.
196                    _ => {
197                        // If we successfully moved the cursor to a previous row,
198                        // jump to the original last shard.
199                        if prev_row.is_some() {
200                            cursor.next()?;
201                        }
202                        // Delete shard.
203                        cursor.delete_current()?;
204                        Ok(PruneShardOutcome::Deleted)
205                    }
206                }
207            }
208            // If current shard is not the last shard for this sharded key,
209            // just delete it.
210            else {
211                cursor.delete_current()?;
212                Ok(PruneShardOutcome::Deleted)
213            }
214        } else {
215            cursor.upsert(
216                RawKey::new(key),
217                &RawValue::new(BlockNumberList::new_pre_sorted(higher_blocks)),
218            )?;
219            Ok(PruneShardOutcome::Updated)
220        }
221    }
222}
reth_prune/segments/user/history.rs

reth_prune/segments/user/
history.rs
