add snapshot rotation to save our diskspace

Cargo.toml

@@ -4,12 +4,27 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+summit-types = {git = "https://github.com/SeismicSystems/summit.git", rev = "bd13c7e80176fd9efaad6e92e80b67a7cd54423e"}
+
+# summit-types at rev bd13c7e was written against commonware-* 2026.2.0. Cargo otherwise
+# resolves these to 2026.3.0 (trait API incompatible), so pin them explicitly.
+commonware-consensus = "=2026.2.0"
+commonware-cryptography = "=2026.2.0"
+commonware-codec = "=2026.2.0"
+commonware-math = "=2026.2.0"
+commonware-utils = "=2026.2.0"
+commonware-resolver = "=2026.2.0"
+commonware-p2p = "=2026.2.0"
+commonware-runtime = "=2026.2.0"
+commonware-parallel = "=2026.2.0"
 # Async runtime
 tokio = { version = "1.42", features = ["full"] }
 
 # HTTP client for RPC
 reqwest = { version = "0.12", features = ["json"] }
 
+ethereum_ssz = "0.9.0"
+
 # JSON-RPC client
 jsonrpsee = { version = "0.26.0", features = ["http-client", "client-core", "server", "macros"] }
 

config.toml

@@ -25,6 +25,9 @@ compact = true
 
 # Path to mdbx_copy binary (can be absolute path or just "mdbx_copy" if in PATH)
 mdbx_copy_path = "mdbx_copy"
+
+# Maximum amount of snapshots we will hold in output_dir. Optional
+#max_snapshots = 50
 
 # Path to reth binary (can be absolute path or just "reth" if in PATH)
 reth_path = "reth"

src/checkpoint/manager.rs

@@ -35,7 +35,11 @@ impl CheckpointManager {
         }
     }
 
-    /// Create a checkpoint for the given epoch and block
+    /// Create a checkpoint for the given epoch.
+    ///
+    /// `block_number` is the height of the last block included in the finalized epoch
+    /// (i.e. `summit_types::Block::from_ssz_bytes(checkpoint.last_block)?.header.height`).
+    /// The reth db copy is unwound to `block_number - 1`, matching summit's finalized tip.
     pub async fn create_checkpoint(&self, epoch: u64, block_number: u64) -> Result<()> {
         let start = std::time::Instant::now();
 
@@ -72,24 +76,20 @@ impl CheckpointManager {
         tracing::info!("Step 3/7: Deleting lock file");
         self.executor.delete_lock_file(&checkpoint_path).await?;
 
-        // Step 4: Unwind database to epoch_block - 2
-        let unwind_target = block_number.saturating_sub(2);
+        // Step 4: Unwind database to one block before the summit-finalized tip of this epoch.
+        let unwind_target = block_number.saturating_sub(1);
         tracing::info!("Step 4/7: Unwinding database to block {}", unwind_target);
         self.executor.unwind_database(&checkpoint_path, unwind_target).await?;
 
         // Step 5: Fetch and write Summit checkpoint data
         tracing::info!("Step 5/7: Fetching Summit checkpoint data");
         if let Some(summit_client) = &self.rpc_client.summit {
-            // Calculate Summit epoch: (block_number / epoch_blocks) - 1
-            // Epochs start at 0, so block 200 is epoch 0, block 400 is epoch 1, etc.
-            let summit_epoch = (block_number / self.config.epoch_blocks).saturating_sub(1);
-
-            tracing::debug!(
-                "Calculated Summit epoch: {} (block {} / epoch_blocks {} - 1)",
-                summit_epoch,
-                block_number,
-                self.config.epoch_blocks
-            );
+            // The `epoch` parameter is now the summit-authoritative epoch number (passed in
+            // from the monitor / ensure_latest_checkpoint, which read it from
+            // summit.getLatestEpoch). Fetch checkpoint data for that exact epoch.
+            let summit_epoch = epoch;
+
+            tracing::debug!("Fetching Summit checkpoint data for epoch {}", summit_epoch);
 
             match summit_client.get_checkpoint(summit_epoch).await {
                 Ok(checkpoint_data) => {
@@ -157,6 +157,11 @@ impl CheckpointManager {
         // Update state tracker
         self.state_tracker.update_last_checkpoint(epoch, block_number).await?;
 
+        // Cleanup old snapshots if retention limit is set
+        if let Err(e) = self.cleanup_old_snapshots().await {
+            tracing::warn!("Failed to cleanup old snapshots: {}", e);
+        }
+
         let duration = start.elapsed();
         tracing::info!(
             "Checkpoint completed successfully: {} (took {:?})",
@@ -167,8 +172,123 @@ impl CheckpointManager {
         Ok(())
     }
 
+    /// Remove oldest snapshot directories when count exceeds the configured max_snapshots limit.
+    async fn cleanup_old_snapshots(&self) -> Result<()> {
+        let max = match self.config.max_snapshots {
+            Some(max) => max as usize,
+            None => return Ok(()),
+        };
+
+        let mut entries = tokio::fs::read_dir(&self.config.output_dir).await?;
+        let mut snapshots: Vec<(u64, PathBuf)> = Vec::new();
+
+        while let Some(entry) = entries.next_entry().await? {
+            let name = entry.file_name();
+            let name_str = name.to_string_lossy();
+            if let Some(epoch_str) = name_str.strip_prefix("epoch_") {
+                if let Ok(epoch) = epoch_str.parse::<u64>() {
+                    snapshots.push((epoch, entry.path()));
+                }
+            }
+        }
+
+        if snapshots.len() <= max {
+            return Ok(());
+        }
+
+        snapshots.sort_by_key(|(epoch, _)| *epoch);
+        let to_remove = snapshots.len() - max;
+
+        for (epoch, path) in snapshots.into_iter().take(to_remove) {
+            tracing::info!("Removing old snapshot: epoch_{}", epoch);
+            tokio::fs::remove_dir_all(&path).await?;
+        }
+
+        Ok(())
+    }
+
     /// Verify that the checkpoint tools are available
     pub async fn verify_checkpoint_tools(&self) -> Result<()> {
         self.executor.verify_available().await
     }
+
+    /// Check whether a compressed checkpoint archive already exists on disk for the given epoch.
+    pub fn checkpoint_exists(&self, epoch: u64) -> bool {
+        self.config
+            .output_dir
+            .join(format!("epoch_{}", epoch))
+            .join(format!("epoch_{}.tar.gz", epoch))
+            .exists()
+    }
+
+    /// Ensure a checkpoint exists for the latest finalized epoch. Intended to be called once
+    /// at startup so we don't have to wait for the next epoch boundary to take a checkpoint.
+    ///
+    /// When summit is enabled (the authoritative path), we call `getLatestCheckpoint` and
+    /// SSZ-decode the returned `last_block` to learn the exact tip of the finalized epoch.
+    /// The unwind inside `create_checkpoint` then lands at `height - 1`.
+    ///
+    /// If summit is disabled, we fall back to block-math (which is incorrect after an
+    /// epoch-length change but preserves legacy behavior for fixed-length chains).
+    pub async fn ensure_latest_checkpoint(&self) -> Result<()> {
+        let (latest_epoch, last_block_height) = match &self.rpc_client.summit {
+            Some(summit) => summit.get_latest_epoch_last_block().await?,
+            None => {
+                let current_block = self.rpc_client.reth.get_block_number().await?;
+                let current_epoch = current_block / self.config.epoch_blocks;
+                if current_epoch == 0 {
+                    tracing::info!(
+                        "Startup check: current block {} is still in epoch 0, nothing to checkpoint",
+                        current_block
+                    );
+                    return Ok(());
+                }
+                // Previous epoch is the most recently finalized; its last block is the
+                // block immediately before the current epoch's first block.
+                (current_epoch - 1, current_epoch * self.config.epoch_blocks - 1)
+            }
+        };
+
+        let last_checkpointed_epoch = self.state_tracker.last_epoch().await;
+        if latest_epoch <= last_checkpointed_epoch {
+            tracing::info!(
+                "Startup check: latest epoch {} already checkpointed (last_epoch={}), skipping",
+                latest_epoch,
+                last_checkpointed_epoch
+            );
+            return Ok(());
+        }
+
+        if self.checkpoint_exists(latest_epoch) {
+            tracing::info!(
+                "Startup check: checkpoint archive for epoch {} already exists on disk, syncing state tracker",
+                latest_epoch
+            );
+            self.state_tracker.update_last_checkpoint(latest_epoch, last_block_height).await?;
+            return Ok(());
+        }
+
+        // Make sure reth has caught up at least to the epoch's final block — otherwise
+        // we can't unwind to it.
+        let current_reth_block = self.rpc_client.reth.get_block_number().await?;
+        if current_reth_block < last_block_height {
+            tracing::info!(
+                "Startup check: reth at block {} is behind summit epoch {} tip (block {}); deferring to monitor",
+                current_reth_block,
+                latest_epoch,
+                last_block_height
+            );
+            return Ok(());
+        }
+
+        tracing::info!(
+            "Startup check: creating checkpoint for epoch {} at finalized block {} (reth at {})",
+            latest_epoch,
+            last_block_height,
+            current_reth_block
+        );
+        self.create_checkpoint(latest_epoch, last_block_height).await?;
+
+        Ok(())
+    }
 }

src/config.rs

@@ -29,6 +29,7 @@ pub struct CheckpointConfig {
     pub compact: bool,
     pub mdbx_copy_path: PathBuf,
     pub reth_path: PathBuf,
+    pub max_snapshots: Option<u64>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -83,6 +84,10 @@ pub struct Cli {
     #[arg(long)]
     pub output_dir: Option<PathBuf>,
 
+    /// Maximum number of snapshots to retain (unlimited if not set)
+    #[arg(long)]
+    pub max_snapshots: Option<u64>,
+
     /// Override mdbx_copy binary path
     #[arg(long)]
     pub mdbx_copy_path: Option<PathBuf>,
@@ -116,6 +121,7 @@ impl Config {
             .set_default("checkpoint.compact", true)?
             .set_default("checkpoint.mdbx_copy_path", "mdbx_copy")?
             .set_default("checkpoint.reth_path", "reth")?
+            .set_default("checkpoint.max_snapshots", None::<u64>)?
             .set_default("summit.enabled", false)?
             .set_default("summit.rpc_url", "http://localhost:5052")?
             .set_default("monitor.poll_interval_secs", 12)?
@@ -153,6 +159,9 @@ impl Config {
         if let Some(output_dir) = &cli.output_dir {
             config.checkpoint.output_dir = output_dir.clone();
         }
+        if let Some(max_snapshots) = cli.max_snapshots {
+            config.checkpoint.max_snapshots = Some(max_snapshots);
+        }
         if let Some(mdbx_copy_path) = &cli.mdbx_copy_path {
             config.checkpoint.mdbx_copy_path = mdbx_copy_path.clone();
         }

src/main.rs

@@ -40,6 +40,15 @@ async fn main() -> Result<()> {
     tracing::info!("Verifying checkpoint tools (mdbx_copy, reth)...");
     checkpoint_manager.verify_checkpoint_tools().await?;
 
+    // Try to create a checkpoint for the latest completed epoch immediately, so we don't
+    // have to wait for the next epoch boundary to get caught up on startup.
+    if let Err(e) = checkpoint_manager.ensure_latest_checkpoint().await {
+        tracing::warn!(
+            "Startup checkpoint attempt failed: {}. Continuing; block monitor will retry.",
+            e
+        );
+    }
+
     // Setup graceful shutdown signal handling
     let shutdown_token = CancellationToken::new();
     let shutdown_signal = shutdown_token.clone();

src/monitor.rs

@@ -71,59 +71,60 @@ impl BlockMonitor {
 
     /// Check current block and create checkpoint if conditions are met
     async fn check_and_checkpoint(&self) -> Result<()> {
-        // Get current block number from reth
-        let current_block = self.rpc_client.reth.get_block_number().await?;
-
-        // Calculate current epoch
-        let current_epoch = current_block / self.epoch_blocks;
+        // Summit is authoritative for both the epoch number and the finalized tip block.
+        // Fall back to block-math only when summit is disabled (broken after an
+        // epoch-length change, but preserves legacy behavior for fixed-length chains).
+        let (latest_epoch, last_block_height) = match &self.rpc_client.summit {
+            Some(summit) => summit.get_latest_epoch_last_block().await?,
+            None => {
+                let current_block = self.rpc_client.reth.get_block_number().await?;
+                let current_epoch = current_block / self.epoch_blocks;
+                if current_epoch == 0 {
+                    return Ok(());
+                }
+                let last_block_of_finalized = current_epoch * self.epoch_blocks - 1;
+                // Keep the configured delay in the fallback path only — summit-based
+                // gating already implies finality.
+                if current_block < last_block_of_finalized + 1 + self.checkpoint_delay_blocks {
+                    return Ok(());
+                }
+                (current_epoch - 1, last_block_of_finalized)
+            }
+        };
 
-        // Get last checkpointed epoch from state
         let last_checkpointed_epoch = self.checkpoint_manager.state_tracker.last_epoch().await;
 
-        // Check if we have crossed an epoch boundary and waited long enough
-        if current_epoch > last_checkpointed_epoch {
-            // Calculate the block number at the epoch boundary we just crossed
-            let epoch_block = current_epoch * self.epoch_blocks;
-
-            // Check if we've waited long enough after the epoch boundary
-            if current_block >= epoch_block + self.checkpoint_delay_blocks {
-                tracing::info!(
-                    "Epoch boundary crossed and delay satisfied: epoch {} at block {} (current block: {}, waited {} blocks)",
-                    current_epoch,
-                    epoch_block,
-                    current_block,
-                    current_block - epoch_block
-                );
-
-                // Create checkpoint for the epoch block, not the current block
-                self.checkpoint_manager.create_checkpoint(current_epoch, epoch_block).await?;
-            } else {
-                let blocks_waited = current_block - epoch_block;
-                let blocks_to_wait = self.checkpoint_delay_blocks - blocks_waited;
-
-                tracing::debug!(
-                    "Epoch boundary crossed at block {} (epoch {}), waiting for {} more blocks before checkpoint (current: {}/{})",
-                    epoch_block,
-                    current_epoch,
-                    blocks_to_wait,
-                    blocks_waited,
-                    self.checkpoint_delay_blocks
-                );
-            }
-        } else {
-            let blocks_in_epoch = current_block % self.epoch_blocks;
-            let blocks_until_next = self.epoch_blocks - blocks_in_epoch;
+        if latest_epoch <= last_checkpointed_epoch {
+            tracing::debug!(
+                "No new epoch: latest={}, last_checkpointed={}",
+                latest_epoch,
+                last_checkpointed_epoch
+            );
+            return Ok(());
+        }
 
+        // Reth must be caught up to the epoch's final block before we can unwind to it.
+        let current_reth_block = self.rpc_client.reth.get_block_number().await?;
+        if current_reth_block < last_block_height {
             tracing::debug!(
-                "Current: epoch={}, block={}, progress={}/{}, blocks_until_epoch_boundary={}",
-                current_epoch,
-                current_block,
-                blocks_in_epoch,
-                self.epoch_blocks,
-                blocks_until_next
+                "Reth at block {} is behind summit epoch {} tip ({}); waiting",
+                current_reth_block,
+                latest_epoch,
+                last_block_height
             );
+            return Ok(());
         }
 
+        tracing::info!(
+            "New epoch detected (epoch={}, prev_checkpointed={}); creating checkpoint at finalized block {} (reth at {})",
+            latest_epoch,
+            last_checkpointed_epoch,
+            last_block_height,
+            current_reth_block
+        );
+
+        self.checkpoint_manager.create_checkpoint(latest_epoch, last_block_height).await?;
+
         Ok(())
     }
 }