feat: EN Pruning (#1418)

Signed-off-by: tomg10 <lemures64@gmail.com>

Cargo.toml

@@ -138,6 +138,7 @@ strum = "0.24"
 tempdir = "0.3.7"
 tempfile = "3.0.2"
 test-casing = "0.1.2"
+test-log = "0.2.15"
 thiserror = "1"
 thread_local = "1.1"
 tikv-jemallocator = "0.5"

checks-config/era.dic

@@ -425,6 +425,8 @@ SetChainId
 setChainId
 SetChainIdUpgrade
 state_transition_manager_contract
+prunable
+bytea
 
 // Names
 Vyper

core/bin/external_node/src/config/mod.rs

@@ -290,6 +290,15 @@ pub(crate) struct OptionalENConfig {
 
     #[serde(default = "OptionalENConfig::default_l1_batch_commit_data_generator_mode")]
     pub l1_batch_commit_data_generator_mode: L1BatchCommitDataGeneratorMode,
+
+    #[serde(default = "OptionalENConfig::default_snapshots_recovery_enabled")]
+    pub snapshots_recovery_enabled: bool,
+
+    #[serde(default = "OptionalENConfig::default_pruning_chunk_size")]
+    pub pruning_chunk_size: u32,
+
+    /// If set, l1 batches will be pruned after they are that long
+    pub pruning_data_retention_hours: Option<u64>,
 }
 
 #[derive(Debug, Clone, PartialEq, Deserialize)]
@@ -425,6 +434,14 @@ impl OptionalENConfig {
         L1BatchCommitDataGeneratorMode::Rollup
     }
 
+    const fn default_snapshots_recovery_enabled() -> bool {
+        false
+    }
+
+    const fn default_pruning_chunk_size() -> u32 {
+        10
+    }
+
     pub fn polling_interval(&self) -> Duration {
         Duration::from_millis(self.polling_interval)
     }

core/bin/external_node/src/main.rs

@@ -24,6 +24,13 @@ use zksync_core::{
     commitment_generator::CommitmentGenerator,
     consensus,
     consistency_checker::ConsistencyChecker,
+    db_pruner::{
+        prune_conditions::{
+            L1BatchExistsCondition, L1BatchOlderThanPruneCondition,
+            NextL1BatchHasMetadataCondition, NextL1BatchWasExecutedCondition,
+        },
+        DbPruner, DbPrunerConfig,
+    },
     eth_sender::l1_batch_commit_data_generator::{
         L1BatchCommitDataGenerator, RollupModeL1BatchCommitDataGenerator,
         ValidiumModeL1BatchCommitDataGenerator,
@@ -255,6 +262,39 @@ async fn run_core(
         }
     }));
 
+    if let Some(data_retention_hours) = config.optional.pruning_data_retention_hours {
+        let l1_batch_age_to_prune = Duration::from_secs(3600 * data_retention_hours);
+        tracing::info!(
+            "Configured pruning of batches after they become {l1_batch_age_to_prune:?} old"
+        );
+        let db_pruner = DbPruner::new(
+            DbPrunerConfig {
+                // don't change this value without adjusting API server pruning info cache max age
+                soft_and_hard_pruning_time_delta: Duration::from_secs(60),
+                pruned_batch_chunk_size: config.optional.pruning_chunk_size,
+                next_iterations_delay: Duration::from_secs(30),
+            },
+            vec![
+                Arc::new(L1BatchExistsCondition {
+                    conn: connection_pool.clone(),
+                }),
+                Arc::new(NextL1BatchHasMetadataCondition {
+                    conn: connection_pool.clone(),
+                }),
+                Arc::new(NextL1BatchWasExecutedCondition {
+                    conn: connection_pool.clone(),
+                }),
+                Arc::new(L1BatchOlderThanPruneCondition {
+                    minimal_age: l1_batch_age_to_prune,
+                    conn: connection_pool.clone(),
+                }),
+            ],
+        )?;
+        task_handles.push(tokio::spawn(
+            db_pruner.run(connection_pool.clone(), stop_receiver.clone()),
+        ));
+    }
+
     let reorg_detector = ReorgDetector::new(main_node_client.clone(), connection_pool.clone());
     app_health.insert_component(reorg_detector.health_check().clone());
     task_handles.push(tokio::spawn({
@@ -663,13 +703,7 @@ struct Cli {
     /// do not use unless you know what you're doing.
     #[arg(long)]
     enable_consensus: bool,
-    /// Enables application-level snapshot recovery. Required to start a node that was recovered from a snapshot,
-    /// or to initialize a node from a snapshot. Has no effect if a node that was initialized from a Postgres dump
-    /// or was synced from genesis.
-    ///
-    /// This is an experimental and incomplete feature; do not use unless you know what you're doing.
-    #[arg(long)]
-    enable_snapshots_recovery: bool,
+
     /// Comma-separated list of components to launch.
     #[arg(long, default_value = "all")]
     components: ComponentsToRun,
@@ -840,7 +874,7 @@ async fn main() -> anyhow::Result<()> {
         main_node_client.clone(),
         &app_health,
         config.remote.l2_chain_id,
-        opt.enable_snapshots_recovery,
+        config.optional.snapshots_recovery_enabled,
     )
     .await?;
     let sigint_receiver = setup_sigint_handler();