Fix trusted-waypoint restore history handling (#341)

## Summary
- skip ledger-info signature validation for transaction/state restore at
or before the highest trusted waypoint
- persist ledger pruner progress at the requested ledger history start
version during `bootstrap-db` restore
- add regression coverage for the pre-waypoint verification bypass

## Why
`bootstrap-db` with `--trust-waypoint` was failing in two ways on
Movement backups:
1. historical transaction chunk proofs could still hit pre-waypoint
validator/signature verification even though the restore should trust
data up to the supplied waypoint
2. even after a successful restore, a node booting from a DB restored
with `--ledger-history-start-version 0` could expose
`oldest_ledger_version` at the snapshot boundary because missing pruner
metadata was inferred from the first `VersionData` row

## Validation
- `cargo test -p aptos-backup-cli
epoch_history_skips_verification_before_highest_trusted_waypoint --
--nocapture`
- `cargo test -p aptos-backup-cli trusted_waypoints -- --nocapture`
- seeded oneoff Movement testnet backup from genesis through version
`36050159`
- restored with:
- `target/release/aptos-debugger aptos-db restore bootstrap-db
--local-fs-dir /private/tmp/movement-testnet-backup-seed.wop5l4
--target-db-dir
/private/tmp/movement-testnet-restore-seed-fixed.WAMrR2/db
--ledger-history-start-version 0 --target-version 36050159
--trust-waypoint
26050159:45860ebee4fe4b8f945bda1fd9d5d0c03778afd2cfa178a7e1806bd0caa823e4
--replay-all`
- started a fullnode on the restored DB and verified:
  - `GET /v1` reports `oldest_ledger_version = 0`
  - `GET /v1/transactions/by_version/0` succeeds

Author: areshand

storage/aptosdb/src/backup/restore_handler.rs

@@ -63,6 +63,10 @@ impl RestoreHandler {
         restore_utils::save_ledger_infos(self.aptosdb.ledger_db.metadata_db(), ledger_infos, None)
     }
 
+    pub fn save_ledger_pruner_progress(&self, version: Version) -> Result<()> {
+        self.ledger_db.write_pruner_progress(version)
+    }
+
     pub fn confirm_or_save_frozen_subtrees(
         &self,
         num_leaves: LeafCount,

storage/backup/backup-cli/src/backup_types/epoch_ending/restore.rs

@@ -285,6 +285,7 @@ pub struct EpochHistory {
 impl EpochHistory {
     pub fn verify_ledger_info(&self, li_with_sigs: &LedgerInfoWithSignatures) -> Result<()> {
         let epoch = li_with_sigs.ledger_info().epoch();
+        let version = li_with_sigs.ledger_info().version();
         ensure!(!self.epoch_endings.is_empty(), "Empty epoch history.",);
         if epoch > self.epoch_endings.len() as u64 {
             // TODO(aldenhu): fix this from upper level
@@ -303,7 +304,7 @@ impl EpochHistory {
             );
         } else if let Some(wp_trusted) = self
             .trusted_waypoints
-            .get(&li_with_sigs.ledger_info().version())
+            .get(&version)
         {
             let wp_li = Waypoint::new_any(li_with_sigs.ledger_info());
             ensure!(
@@ -312,6 +313,15 @@ impl EpochHistory {
                 wp_li,
                 wp_trusted,
             );
+        } else if self
+            .trusted_waypoints
+            .keys()
+            .max()
+            .is_some_and(|max_wp| version <= *max_wp)
+        {
+            // For historical restore, anything up to the highest trusted waypoint is treated as
+            // trusted data. This keeps transaction/state restore aligned with epoch-ending
+            // preheat, which already bypasses signature verification for this range.
         } else {
             self.epoch_endings[epoch as usize - 1]
                 .next_epoch_state()

storage/backup/backup-cli/src/backup_types/epoch_ending/tests.rs

@@ -18,17 +18,20 @@ use crate::{
 };
 use aptos_backup_service::start_backup_service;
 use aptos_config::utils::get_available_port;
+use aptos_crypto::hash::HashValue;
 use aptos_db::AptosDB;
 use aptos_storage_interface::DbReader;
 use aptos_temppath::TempPath;
 use aptos_types::{
     aggregate_signature::AggregateSignature,
-    ledger_info::LedgerInfoWithSignatures,
+    block_info::BlockInfo,
+    ledger_info::{LedgerInfo, LedgerInfoWithSignatures},
     proptest_types::{AccountInfoUniverse, LedgerInfoWithSignaturesGen},
     waypoint::Waypoint,
 };
 use proptest::{collection::vec, prelude::*};
 use std::{
+    collections::HashMap,
     convert::TryInto,
     io::Write,
     net::{IpAddr, Ipv4Addr, SocketAddr},
@@ -269,3 +272,47 @@ proptest! {
         Runtime::new().unwrap().block_on(test_trusted_waypoints_impl(lis, trusted_waypoints, should_fail_without))
     }
 }
+
+#[test]
+fn epoch_history_skips_verification_before_highest_trusted_waypoint() {
+    let pre_waypoint_li = LedgerInfoWithSignatures::new(
+        LedgerInfo::new(
+            BlockInfo::new(
+                1,
+                0,
+                HashValue::zero(),
+                HashValue::zero(),
+                10,
+                0,
+                None,
+            ),
+            HashValue::zero(),
+        ),
+        AggregateSignature::empty(),
+    );
+    let trusted_waypoint_li = LedgerInfoWithSignatures::new(
+        LedgerInfo::new(
+            BlockInfo::new(
+                2,
+                0,
+                HashValue::zero(),
+                HashValue::zero(),
+                20,
+                0,
+                None,
+            ),
+            HashValue::zero(),
+        ),
+        AggregateSignature::empty(),
+    );
+
+    let epoch_history = crate::backup_types::epoch_ending::restore::EpochHistory {
+        epoch_endings: vec![LedgerInfo::dummy(), LedgerInfo::dummy(), LedgerInfo::dummy()],
+        trusted_waypoints: Arc::new(HashMap::from([(
+            20,
+            Waypoint::new_any(trusted_waypoint_li.ledger_info()),
+        )])),
+    };
+
+    epoch_history.verify_ledger_info(&pre_waypoint_li).unwrap();
+}

storage/backup/backup-cli/src/coordinators/restore.rs

@@ -135,6 +135,12 @@ impl RestoreCoordinator {
         COORDINATOR_TARGET_VERSION.set(target_version as i64);
         let lhs = self.ledger_history_start_version();
 
+        // Make the readable ledger history boundary explicit in the restored DB.
+        // Otherwise, when these metadata keys are missing, node startup infers
+        // the ledger pruner progress from the first VersionData row, which is
+        // typically the state snapshot version rather than the requested lhs.
+        self.global_opt.run_mode.save_ledger_pruner_progress(lhs)?;
+
         let latest_tree_version = self
             .global_opt
             .run_mode

storage/backup/backup-cli/src/utils/mod.rs

@@ -262,6 +262,15 @@ impl RestoreRunMode {
         }
     }
 
+    pub fn save_ledger_pruner_progress(&self, version: Version) -> Result<()> {
+        match self {
+            RestoreRunMode::Restore { restore_handler } => {
+                restore_handler.save_ledger_pruner_progress(version)
+            },
+            RestoreRunMode::Verify => Ok(()),
+        }
+    }
+
     pub fn get_state_snapshot_before(&self, version: Version) -> Option<(Version, HashValue)> {
         match self {
             RestoreRunMode::Restore { restore_handler } => restore_handler