feat(l1): pruning of historical blocks/receipts/txs

cmd/ethrex/Cargo.toml

@@ -99,7 +99,12 @@ c-kzg = [
   "ethrex-p2p/c-kzg",
   "ethrex-crypto/c-kzg",
 ]
-metrics = ["ethrex-blockchain/metrics", "ethrex-l2?/metrics", "ethrex-p2p/metrics"]
+metrics = [
+  "ethrex-blockchain/metrics",
+  "ethrex-l2?/metrics",
+  "ethrex-p2p/metrics",
+  "ethrex-storage/metrics",
+]
 rocksdb = ["ethrex-storage/rocksdb", "ethrex-p2p/rocksdb", "ethrex-l2?/rocksdb"]
 jemalloc = ["dep:tikv-jemallocator"]
 jemalloc_profiling = [

cmd/ethrex/cli.rs

@@ -24,6 +24,7 @@ use ethrex_p2p::{
     types::Node,
 };
 use ethrex_rlp::encode::RLPEncode;
+use ethrex_storage::pruner::{ASSUMED_SECONDS_PER_SLOT, HistoryRetention};
 use ethrex_storage::{DB_COMMIT_THRESHOLD, error::StoreError, has_valid_db};
 use tokio_util::sync::CancellationToken;
 use tracing::{Level, error, info, warn};
@@ -530,6 +531,31 @@ pub struct Options {
         env = "ETHREX_PRECOMPUTE_WITNESSES"
     )]
     pub precompute_witnesses: bool,
+    #[arg(
+        long = "history.retention",
+        value_name = "RETENTION",
+        value_parser = parse_history_retention,
+        help = "How much block history to keep. `cl-window` (default) keeps the CL block-retention window of 33024 epochs, the longest range a consensus client is required to serve and therefore the least an execution client should hold. `all` never prunes. `<N>epochs` keeps N epochs exactly. `<N>d`/`<N>h`/`<N>m` keep a wall-clock window, converted to a block distance once at startup assuming 12s slots — convenient, but `<N>epochs` is exact and survives a slot-time change. Bodies, receipts and transaction locations below the window are deleted permanently; canonical headers are always kept.",
+        help_heading = "P2P options",
+        env = "ETHREX_HISTORY_RETENTION"
+    )]
+    pub history_retention: Option<HistoryRetention>,
+    #[arg(
+        long = "history.retention.dry-run",
+        default_value_t = false,
+        help = "Resolve the retention policy, report what pruning would delete, and exit without deleting anything. Use this before enabling pruning on a datadir you cannot replace: pruning is irreversible without a resync.",
+        help_heading = "P2P options",
+        env = "ETHREX_HISTORY_RETENTION_DRY_RUN"
+    )]
+    pub history_retention_dry_run: bool,
+    #[arg(
+        long = "history.retention.below-cl-window",
+        default_value_t = false,
+        help = "Permit a --history.retention below the CL block-retention window. Such a node cannot serve the range its peers are entitled to ask for; intended for devnets and short-lived chains.",
+        help_heading = "P2P options",
+        env = "ETHREX_HISTORY_RETENTION_BELOW_CL_WINDOW"
+    )]
+    pub history_retention_below_cl_window: bool,
     #[arg(
         long = "max-reorg-depth",
         value_name = "MAX_REORG_DEPTH",
@@ -635,6 +661,9 @@ impl Default for Options {
             no_migrate: false,
             skip_genesis_validation: false,
             no_precompile_cache: false,
+            history_retention: None,
+            history_retention_below_cl_window: false,
+            history_retention_dry_run: false,
             no_bal_parallel_exec: false,
             no_bal_prefetch: false,
             no_bal_parallel_trie: false,
@@ -1405,12 +1434,125 @@ pub async fn export_blocks(
     );
 }
 
+/// `cl-window` | `all` | `<N>epochs`.
+///
+/// Deliberately not a bare integer: `--history.chain=22000000` already means an
+/// absolute block number, so an adjacent flag where a bare number meant a
+/// *distance* would read identically and mean something else. The `epochs` suffix
+/// removes the ambiguity. Wall-clock durations are gone too — the window is
+/// defined in epochs by the consensus specs, and deriving it from a clock made a
+/// skewed host prune real history.
+fn parse_history_retention(s: &str) -> Result<HistoryRetention, String> {
+    let value = s.trim().to_ascii_lowercase();
+    match value.as_str() {
+        "cl-window" => return Ok(HistoryRetention::CL_WINDOW),
+        "all" => return Ok(HistoryRetention::All),
+        _ => {}
+    }
+    // A wall-clock window is converted to a block distance here, once, rather than
+    // being compared against the host clock on every pruning pass — that is what made
+    // the previous duration-based implementation able to delete a year of history from
+    // a machine whose clock was a year fast. The conversion assumes
+    // `ASSUMED_SECONDS_PER_SLOT`, so `<N>epochs` remains the exact spelling.
+    for (suffix, secs) in [("d", 86_400u64), ("h", 3_600), ("m", 60)] {
+        if let Some(count) = value.strip_suffix(suffix)
+            && let Ok(count) = count.trim().parse::<u64>()
+        {
+            if count == 0 {
+                return Err("a retention of 0 would keep nothing; use a positive value".into());
+            }
+            let blocks = count
+                .saturating_mul(secs)
+                .saturating_div(ASSUMED_SECONDS_PER_SLOT);
+            return Ok(HistoryRetention::Blocks(blocks));
+        }
+    }
+    if let Some(epochs) = value.strip_suffix("epochs") {
+        let epochs: u64 = epochs
+            .trim()
+            .parse()
+            .map_err(|_| format!("invalid epoch count in `{s}`"))?;
+        if epochs == 0 {
+            return Err("a retention of 0 epochs would keep nothing; use a positive count".into());
+        }
+        return Ok(HistoryRetention::Epochs(epochs));
+    }
+    Err(format!(
+        "invalid history retention `{s}`: expected `cl-window`, `all`, `<N>epochs`, \
+         or a wall-clock window such as `30d`, `12h` or `90m`"
+    ))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use clap::Parser;
     use ethrex_rpc::RpcNamespace;
 
+    #[test]
+    fn parses_history_retention_named_modes() {
+        let cli = CLI::try_parse_from(["ethrex", "--history.retention", "cl-window"]).unwrap();
+        assert_eq!(
+            cli.opts.history_retention,
+            Some(HistoryRetention::CL_WINDOW)
+        );
+        let cli = CLI::try_parse_from(["ethrex", "--history.retention", "all"]).unwrap();
+        assert_eq!(cli.opts.history_retention, Some(HistoryRetention::All));
+    }
+
+    #[test]
+    fn parses_history_retention_epochs() {
+        let cli = CLI::try_parse_from(["ethrex", "--history.retention", "82125epochs"]).unwrap();
+        assert_eq!(
+            cli.opts.history_retention,
+            Some(HistoryRetention::Epochs(82125))
+        );
+    }
+
+    /// A bare number is rejected: `--history.chain` already takes an absolute block
+    /// number, so a neighbouring flag where a bare number meant a *distance* would
+    /// read identically and mean something else.
+    #[test]
+    fn rejects_history_retention_without_a_unit() {
+        assert!(CLI::try_parse_from(["ethrex", "--history.retention", "1056768"]).is_err());
+        assert!(CLI::try_parse_from(["ethrex", "--history.retention", "0epochs"]).is_err());
+        assert!(CLI::try_parse_from(["ethrex", "--history.retention", "30"]).is_err());
+        assert!(CLI::try_parse_from(["ethrex", "--history.retention", "30years"]).is_err());
+    }
+
+    /// A wall-clock window is a convenience that resolves to a block distance at
+    /// startup, so it never reads the host clock while running.
+    #[test]
+    fn parses_history_retention_wall_clock() {
+        let cli = CLI::try_parse_from(["ethrex", "--history.retention", "30d"]).unwrap();
+        assert_eq!(
+            cli.opts.history_retention,
+            // 30 days of 12s slots.
+            Some(HistoryRetention::Blocks(30 * 86_400 / 12))
+        );
+        let cli = CLI::try_parse_from(["ethrex", "--history.retention", "12h"]).unwrap();
+        assert_eq!(
+            cli.opts.history_retention,
+            Some(HistoryRetention::Blocks(12 * 3_600 / 12))
+        );
+        assert!(CLI::try_parse_from(["ethrex", "--history.retention", "0d"]).is_err());
+    }
+
+    /// Absent means "use the default", which the initializer resolves to the CL
+    /// window; it must stay distinguishable from an explicit value, because only an
+    /// explicit one may delete history a node already holds.
+    #[test]
+    fn history_retention_defaults_to_unset() {
+        let cli = CLI::try_parse_from(["ethrex"]).unwrap();
+        assert_eq!(cli.opts.history_retention, None);
+    }
+
+    #[test]
+    fn history_retention_default_is_none() {
+        let cli = CLI::try_parse_from(["ethrex"]).unwrap();
+        assert_eq!(cli.opts.history_retention, None);
+    }
+
     /// `--http.addr` must default to `127.0.0.1` so a fresh install on a public
     /// host is not exposed to the open internet.
     #[test]

cmd/ethrex/ethrex.rs

@@ -52,10 +52,22 @@ async fn server_shutdown(
     peer_table: PeerTable,
     local_node_record: NodeRecord,
     store: &Store,
+    pruner_handle: Option<tokio::task::JoinHandle<()>>,
 ) {
     info!("Server shut down started...");
     // Stop feeding new blocks before draining, so the persist queue can't grow.
     cancel_token.cancel();
+    // Await the history pruner before flushing. Cancellation stops it from starting
+    // another pass but does not abort one in flight, and it writes through its own
+    // transactions rather than the persist worker, so a pass that outlived the flush
+    // below would land a batch after the final fsync and force WAL recovery. Bounded
+    // by the pruner's own pass deadline, so this cannot hang shutdown for long.
+    if let Some(handle) = pruner_handle {
+        info!("Waiting for the history pruner to finish its current pass...");
+        if let Err(err) = handle.await {
+            error!("History pruner task did not shut down cleanly: {err}");
+        }
+    }
     // Drain the persist worker, force-flush the block-data buffer, and fsync the
     // DB. Without this an abrupt exit (e.g. `docker restart -t 0`) loses the
     // buffered block-data tail and leaves the DB needing WAL recovery on next
@@ -191,23 +203,26 @@ async fn main() -> eyre::Result<()> {
     info!("ethrex version: {}", get_client_version());
     tokio::spawn(periodically_check_version_update());
 
-    let (datadir, cancel_token, peer_table, local_node_record, store) =
+    let (datadir, cancel_token, peer_table, local_node_record, store, pruner_handle) =
         init_l1(opts, Some(log_filter_handler)).await?;
 
     let mut signal_terminate = signal(SignalKind::terminate())?;
 
     log_global_allocator();
 
+    // `take()` because each `select!` arm would otherwise move the same handle.
+    let mut pruner_handle = pruner_handle;
+
     tokio::select! {
         _ = tokio::signal::ctrl_c() => {
-            server_shutdown(&datadir, &cancel_token, peer_table, local_node_record, &store).await;
+            server_shutdown(&datadir, &cancel_token, peer_table, local_node_record, &store, pruner_handle.take()).await;
         }
         _ = signal_terminate.recv() => {
-            server_shutdown(&datadir, &cancel_token, peer_table, local_node_record, &store).await;
+            server_shutdown(&datadir, &cancel_token, peer_table, local_node_record, &store, pruner_handle.take()).await;
         }
         // A fatal subsystem (e.g. the RPC server) cancels the token to abort the node.
         _ = cancel_token.cancelled() => {
-            server_shutdown(&datadir, &cancel_token, peer_table, local_node_record, &store).await;
+            server_shutdown(&datadir, &cancel_token, peer_table, local_node_record, &store, pruner_handle.take()).await;
         }
     }