Skip to main content

solana_runtime/
snapshot_utils.rs

1#[cfg(feature = "dev-context-only-utils")]
2use solana_accounts_db::utils::create_accounts_run_and_snapshot_dirs;
3use {
4    crate::{
5        bank::BankFieldsToDeserialize,
6        serde_snapshot::{
7            self, AccountsDbFields, ExtraFieldsToSerialize, SerdeObsoleteAccountsMap,
8            SnapshotAccountsDbFields, SnapshotBankFields, SnapshotStreams, StorageListItem,
9            StoragesList,
10        },
11        snapshot_package::BankSnapshotPackage,
12        snapshot_utils::snapshot_storage_rebuilder::{
13            SnapshotStorageRebuilder, get_slot_and_append_vec_id,
14        },
15    },
16    agave_fs::{
17        FileInfo, FileSize,
18        buffered_reader::large_file_buf_reader,
19        buffered_writer::{SizeLimitedWriter, large_file_buf_writer},
20        io_setup::IoSetupState,
21    },
22    agave_snapshots::{
23        ArchiveFormat, Result, SnapshotArchiveKind, SnapshotVersion, archive_snapshot,
24        error::{
25            AddBankSnapshotError, SnapshotError, SnapshotFastbootError, SnapshotNewFromDirError,
26        },
27        paths::{self as snapshot_paths, incremental_snapshot_archives_iter},
28        snapshot_archive_info::{
29            FullSnapshotArchiveInfo, IncrementalSnapshotArchiveInfo, SnapshotArchiveInfo,
30            SnapshotArchiveInfoGetter,
31        },
32        snapshot_config::SnapshotConfig,
33        snapshot_hash::SnapshotHash,
34        streaming_unarchive_snapshot,
35    },
36    crossbeam_channel::Receiver,
37    log::*,
38    regex::Regex,
39    semver::Version,
40    solana_accounts_db::{
41        account_storage::AccountStorageMap,
42        account_storage_entry::AccountStorageEntry,
43        accounts_db::{AccountsFileId, AtomicAccountsFileId},
44        utils::{
45            ACCOUNTS_RUN_DIR, ACCOUNTS_SNAPSHOT_DIR, move_and_async_delete_path,
46            move_and_async_delete_path_contents,
47        },
48    },
49    solana_clock::Slot,
50    solana_measure::{measure::Measure, measure_time, measure_us},
51    std::{
52        cmp::Ordering,
53        collections::{HashMap, HashSet},
54        fs,
55        io::{self, BufReader, Error as IoError, Read, Seek, Write},
56        mem,
57        num::NonZeroUsize,
58        path::{Path, PathBuf},
59        str::FromStr,
60        sync::{Arc, LazyLock},
61        thread,
62    },
63    tempfile::TempDir,
64    wincode::io::std_read::ReadAdapter,
65};
66
67pub mod snapshot_storage_rebuilder;
68
69/// Limit the size of the obsolete accounts file
70/// If it exceeds this limit, remove the file which will force restore from archives
71/// Limit is set assuming 24 bytes per entry, 5% of 10 billion accounts
72/// = 500 million entries * 24 bytes = 12 GB
73pub const MAX_OBSOLETE_ACCOUNTS_FILE_SIZE: u64 = 1024 * 1024 * 1024 * 12; // 12 GB
74/// Limit the size of the storages list file.
75/// Each `(slot, id)` entry encodes to 12 bytes; 100 MiB covers ~8.7 million entries, well past
76/// any realistic storage count.
77pub const MAX_STORAGES_LIST_FILE_SIZE: u64 = 100 * 1024 * 1024; // 100 MiB
78pub const MAX_SNAPSHOT_DATA_FILE_SIZE: u64 = 32 * 1024 * 1024 * 1024; // 32 GiB
79const MAX_SNAPSHOT_VERSION_FILE_SIZE: u64 = 8; // byte
80/// Buffer size for reading auxiliary per-snapshot files (obsolete accounts, storages list).
81/// Sized to allow several concurrent reads at the default io-uring reader read size (1MiB).
82const AUX_SNAPSHOT_FILE_READ_BUF_SIZE: usize = 4 * 1024 * 1024;
83
84// Snapshot Fastboot Version History
85// Legacy - No fastboot version file, storages flushed file presence determines if snapshot is loadable
86// 1.0.0 - Initial version file. Backwards and forwards compatible with Legacy.
87// 2.0.0 - Obsolete Accounts File added, storages flushed file not written anymore
88//         Snapshots created with version 2.0.0 will not fastboot to older versions
89//         Snapshots created with versions <2.0.0 will fastboot to version 2.0.0
90// 3.0.0 - Storages List file added, replaces the per-storage hardlink dirs.
91//         3.0.0 validators can still fastboot from 2.0.0 snapshots: the legacy hardlinks are
92//         migrated back into the account run dirs at load time (see `migrate_legacy_hardlinks`),
93//         and the next teardown writes the new-format storages list.
94//         Note: 2.0.0 validators cannot fastboot from 3.0.0 snapshots because the per-storage
95//         hardlink dirs they rely on are no longer written; they must fall back to archive.
96const SNAPSHOT_FASTBOOT_VERSION: Version = Version::new(3, 0, 0);
97
98/// Information about a bank snapshot. Namely the slot of the bank, the path to the snapshot, and
99/// the kind of the snapshot.
100#[derive(PartialEq, Eq, Debug)]
101pub struct BankSnapshotInfo {
102    /// Slot of the bank
103    pub slot: Slot,
104    /// Path to the bank snapshot directory
105    pub snapshot_dir: PathBuf,
106    /// Snapshot version
107    pub snapshot_version: SnapshotVersion,
108    /// Fastboot version
109    pub fastboot_version: Option<Version>,
110}
111
112impl PartialOrd for BankSnapshotInfo {
113    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
114        Some(self.cmp(other))
115    }
116}
117
118// Order BankSnapshotInfo by slot (ascending), which practically is sorting chronologically
119impl Ord for BankSnapshotInfo {
120    fn cmp(&self, other: &Self) -> Ordering {
121        self.slot.cmp(&other.slot)
122    }
123}
124
125impl BankSnapshotInfo {
126    pub fn new_from_dir(
127        bank_snapshots_dir: impl AsRef<Path>,
128        slot: Slot,
129    ) -> std::result::Result<BankSnapshotInfo, SnapshotNewFromDirError> {
130        // check this directory to see if there is a BankSnapshotPre and/or
131        // BankSnapshotPost file
132        let bank_snapshot_dir = snapshot_paths::get_bank_snapshot_dir(&bank_snapshots_dir, slot);
133
134        if !bank_snapshot_dir.is_dir() {
135            return Err(SnapshotNewFromDirError::InvalidBankSnapshotDir(
136                bank_snapshot_dir,
137            ));
138        }
139
140        // Among the files checks, the completion flag file check should be done first to avoid the later
141        // I/O errors.
142
143        // There is a time window from the slot directory being created, and the content being completely
144        // filled.  Check the version file as it is the last file written to avoid using a highest
145        // found slot directory with missing content
146        let version_path = bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
147        let version_file_info = FileInfo::new_from_path(&version_path)
148            .map_err(|err| SnapshotNewFromDirError::IncompleteDir(err, version_path))?;
149        let version_str = snapshot_version_from_file(version_file_info).map_err(|err| {
150            SnapshotNewFromDirError::IncompleteDir(err, bank_snapshot_dir.clone())
151        })?;
152
153        let snapshot_version = SnapshotVersion::from_str(version_str.as_str())
154            .or(Err(SnapshotNewFromDirError::InvalidVersion(version_str)))?;
155
156        let status_cache_file =
157            bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME);
158        if !status_cache_file.is_file() {
159            return Err(SnapshotNewFromDirError::MissingStatusCacheFile(
160                status_cache_file,
161            ));
162        }
163
164        let bank_snapshot_path =
165            bank_snapshot_dir.join(snapshot_paths::get_snapshot_file_name(slot));
166        if !bank_snapshot_path.is_file() {
167            return Err(SnapshotNewFromDirError::MissingSnapshotFile(
168                bank_snapshot_dir,
169            ));
170        };
171
172        let snapshot_fastboot_version_path =
173            bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_FASTBOOT_VERSION_FILENAME);
174
175        // If the version file is absent, fastboot_version will be None. This allows versions 3.1+
176        // to load snapshots created by versions <3.1. In version 3.2, the version file will become
177        // mandatory, and its absence can be treated as an error.
178        let fastboot_version = fs::read_to_string(&snapshot_fastboot_version_path)
179            .ok()
180            .map(|version_string| {
181                Version::from_str(version_string.trim())
182                    .map_err(|_| SnapshotNewFromDirError::InvalidFastbootVersion(version_string))
183            })
184            .transpose()?;
185
186        Ok(BankSnapshotInfo {
187            slot,
188            snapshot_dir: bank_snapshot_dir,
189            snapshot_version,
190            fastboot_version,
191        })
192    }
193
194    pub fn snapshot_path(&self) -> PathBuf {
195        self.snapshot_dir
196            .join(snapshot_paths::get_snapshot_file_name(self.slot))
197    }
198}
199
200/// When constructing a bank a snapshot, traditionally the snapshot was from a snapshot archive.  Now,
201/// the snapshot can be from a snapshot directory, or from a snapshot archive.  This is the flag to
202/// indicate which.
203#[derive(Clone, Copy, Debug, Eq, PartialEq)]
204pub enum SnapshotFrom {
205    /// Build from the snapshot archive
206    Archive,
207    /// Build directly from the bank snapshot directory
208    Dir,
209}
210
211/// Helper type when rebuilding from snapshots.  Designed to handle when rebuilding from just a
212/// full snapshot, or from both a full snapshot and an incremental snapshot.
213#[derive(Debug)]
214pub struct SnapshotRootPaths {
215    pub full_snapshot_root_file_path: PathBuf,
216    pub incremental_snapshot_root_file_path: Option<PathBuf>,
217}
218
219/// Helper type to bundle up the results from `unarchive_snapshot()`
220#[derive(Debug)]
221pub struct UnarchivedSnapshot {
222    unpack_dir: TempDir,
223    pub storage: AccountStorageMap,
224    pub bank_fields: BankFieldsToDeserialize,
225    pub(crate) accounts_db_fields: AccountsDbFields,
226    pub unpacked_snapshots_dir_and_version: UnpackedSnapshotsDirAndVersion,
227    pub measure_untar: Measure,
228}
229
230/// Helper type to bundle up the results from `verify_and_unarchive_snapshots()`.
231#[derive(Debug)]
232pub struct UnarchivedSnapshots {
233    pub full_storage: AccountStorageMap,
234    pub incremental_storage: Option<AccountStorageMap>,
235    pub bank_fields: SnapshotBankFields,
236    pub accounts_db_fields: SnapshotAccountsDbFields,
237    pub full_unpacked_snapshots_dir_and_version: UnpackedSnapshotsDirAndVersion,
238    pub incremental_unpacked_snapshots_dir_and_version: Option<UnpackedSnapshotsDirAndVersion>,
239    pub full_measure_untar: Measure,
240    pub incremental_measure_untar: Option<Measure>,
241    pub next_append_vec_id: AtomicAccountsFileId,
242}
243
244/// Guard type that keeps the unpack directories of snapshots alive.
245/// Once dropped, the unpack directories are removed.
246#[expect(dead_code)]
247#[derive(Debug)]
248pub struct UnarchivedSnapshotsGuard {
249    full_unpack_dir: TempDir,
250    incremental_unpack_dir: Option<TempDir>,
251}
252/// Helper type for passing around the unpacked snapshots dir and the snapshot version together
253#[derive(Debug)]
254pub struct UnpackedSnapshotsDirAndVersion {
255    pub unpacked_snapshots_dir: PathBuf,
256    pub snapshot_version: SnapshotVersion,
257}
258
259/// Helper type for passing around account storage map and next append vec id
260/// for reconstructing accounts from a snapshot
261pub(crate) struct StorageAndNextAccountsFileId {
262    pub storage: AccountStorageMap,
263    pub next_append_vec_id: AtomicAccountsFileId,
264}
265
266/// Purges incomplete bank snapshots
267pub fn purge_incomplete_bank_snapshots(bank_snapshots_dir: impl AsRef<Path>) {
268    let Ok(read_dir_iter) = std::fs::read_dir(&bank_snapshots_dir) else {
269        // If we cannot read the bank snapshots dir, then there's nothing to do
270        return;
271    };
272
273    let is_incomplete = |dir: &PathBuf| !is_bank_snapshot_complete(dir);
274
275    let incomplete_dirs: Vec<_> = read_dir_iter
276        .filter_map(|entry| entry.ok())
277        .map(|entry| entry.path())
278        .filter(|path| path.is_dir())
279        .filter(is_incomplete)
280        .collect();
281
282    // attempt to purge all the incomplete directories; do not exit early
283    for incomplete_dir in incomplete_dirs {
284        let result = purge_bank_snapshot(&incomplete_dir);
285        match result {
286            Ok(_) => info!(
287                "Purged incomplete snapshot dir: {}",
288                incomplete_dir.display()
289            ),
290            Err(err) => warn!("Failed to purge incomplete snapshot dir: {err}"),
291        }
292    }
293}
294
295/// Is the bank snapshot complete?
296fn is_bank_snapshot_complete(bank_snapshot_dir: impl AsRef<Path>) -> bool {
297    let version_path = bank_snapshot_dir
298        .as_ref()
299        .join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
300
301    let Ok(version_file_info) = FileInfo::new_from_path(&version_path) else {
302        // failed to either open or query the file -- snapshot is incomplete
303        return false;
304    };
305
306    let Ok(version_str) = snapshot_version_from_file(version_file_info) else {
307        // failed to read from file -- snapshot is incomplete
308        return false;
309    };
310
311    let Ok(_snapshot_version) = SnapshotVersion::from_str(version_str.as_str()) else {
312        // invalid snapshot version -- snapshot is incomplete
313        return false;
314    };
315
316    // version file is good, so now check the serialized bank and status cache files
317    let Some(slot) = bank_snapshot_dir.as_ref().file_name() else {
318        return false;
319    };
320    let Some(slot) = slot.to_str() else {
321        return false;
322    };
323    for file_name in [slot, snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME] {
324        let file_path = bank_snapshot_dir.as_ref().join(file_name);
325        let Ok(file_info) = FileInfo::new_from_path(file_path) else {
326            // failed to either open or query the file -- snapshot is incomplete
327            return false;
328        };
329        if file_info.size == 0 {
330            // file is empty -- snapshot is incomplete
331            return false;
332        }
333    }
334
335    true
336}
337
338/// Writes files that indicate the bank snapshot is loadable by fastboot
339pub fn mark_bank_snapshot_as_loadable(bank_snapshot_dir: impl AsRef<Path>) -> io::Result<()> {
340    let snapshot_fastboot_version_path = bank_snapshot_dir
341        .as_ref()
342        .join(snapshot_paths::SNAPSHOT_FASTBOOT_VERSION_FILENAME);
343    fs::write(
344        &snapshot_fastboot_version_path,
345        SNAPSHOT_FASTBOOT_VERSION.to_string(),
346    )
347    .map_err(|err| {
348        IoError::other(format!(
349            "failed to write fastboot version file '{}': {err}",
350            snapshot_fastboot_version_path.display(),
351        ))
352    })?;
353    Ok(())
354}
355
356/// Is this bank snapshot loadable?
357fn is_bank_snapshot_loadable(
358    fastboot_version: Option<&Version>,
359) -> std::result::Result<bool, SnapshotFastbootError> {
360    if let Some(fastboot_version) = fastboot_version {
361        is_snapshot_fastboot_compatible(fastboot_version)
362    } else {
363        // No fastboot version file, so this is not a fastbootable
364        Ok(false)
365    }
366}
367
368/// Is the fastboot snapshot version compatible?
369fn is_snapshot_fastboot_compatible(
370    version: &Version,
371) -> std::result::Result<bool, SnapshotFastbootError> {
372    match version.major {
373        // Current format: storages list lives next to the bank snapshot file.
374        3 => Ok(true),
375        // Legacy format: per-storage hardlink dirs. `rebuild_storages_from_snapshot_dir`
376        // migrates them to the new format at load time.
377        2 => Ok(true),
378        v if v > SNAPSHOT_FASTBOOT_VERSION.major => {
379            Err(SnapshotFastbootError::IncompatibleVersion(version.clone()))
380        }
381        // Older format we no longer know how to load — fall back to archive.
382        _ => Ok(false),
383    }
384}
385
386/// Gets the highest, loadable, bank snapshot
387///
388/// The highest bank snapshot is the one with the highest slot.
389pub fn get_highest_loadable_bank_snapshot(
390    snapshot_config: &SnapshotConfig,
391) -> Option<BankSnapshotInfo> {
392    let highest_bank_snapshot = get_highest_bank_snapshot(&snapshot_config.bank_snapshots_dir)?;
393
394    let is_bank_snapshot_loadable =
395        is_bank_snapshot_loadable(highest_bank_snapshot.fastboot_version.as_ref());
396
397    match is_bank_snapshot_loadable {
398        Ok(true) => Some(highest_bank_snapshot),
399        Ok(false) => None,
400        Err(err) => {
401            warn!(
402                "Bank snapshot is not loadable '{}': {err}",
403                highest_bank_snapshot.snapshot_dir.display()
404            );
405            None
406        }
407    }
408}
409
410/// If the validator halts in the middle of `archive_snapshot_package()`, the temporary staging
411/// directory won't be cleaned up.  Call this function to clean them up.
412pub fn remove_tmp_snapshot_archives(snapshot_archives_dir: impl AsRef<Path>) {
413    if let Ok(entries) = std::fs::read_dir(snapshot_archives_dir) {
414        for entry in entries.flatten() {
415            if entry
416                .file_name()
417                .to_str()
418                .map(|file_name| file_name.starts_with(snapshot_paths::TMP_SNAPSHOT_ARCHIVE_PREFIX))
419                .unwrap_or(false)
420            {
421                let path = entry.path();
422                let result = if path.is_dir() {
423                    fs::remove_dir_all(&path)
424                } else {
425                    fs::remove_file(&path)
426                };
427                if let Err(err) = result {
428                    warn!(
429                        "Failed to remove temporary snapshot archive '{}': {err}",
430                        path.display(),
431                    );
432                }
433            }
434        }
435    }
436}
437
438/// Creates an archive based on the bank snapshot and snapshot storages
439pub fn archive_snapshot_package(
440    snapshot_archive_kind: SnapshotArchiveKind,
441    snapshot_slot: Slot,
442    snapshot_hash: SnapshotHash,
443    bank_snapshot_dir: impl AsRef<Path>,
444    mut snapshot_storages: Vec<Arc<AccountStorageEntry>>,
445    snapshot_config: &SnapshotConfig,
446    io_setup: &IoSetupState,
447) -> Result<SnapshotArchiveInfo> {
448    let snapshot_archive_path = match snapshot_archive_kind {
449        SnapshotArchiveKind::Full => snapshot_paths::build_full_snapshot_archive_path(
450            &snapshot_config.full_snapshot_archives_dir,
451            snapshot_slot,
452            &snapshot_hash,
453            snapshot_config.archive_format,
454        ),
455        SnapshotArchiveKind::Incremental(incremental_snapshot_base_slot) => {
456            // After the snapshot has been serialized, it is now safe (and required) to prune all
457            // the storages that are *not* to be archived for this incremental snapshot.
458            snapshot_storages.retain(|storage| storage.slot() > incremental_snapshot_base_slot);
459            snapshot_paths::build_incremental_snapshot_archive_path(
460                &snapshot_config.incremental_snapshot_archives_dir,
461                incremental_snapshot_base_slot,
462                snapshot_slot,
463                &snapshot_hash,
464                snapshot_config.archive_format,
465            )
466        }
467    };
468
469    let snapshot_archive_info = archive_snapshot(
470        snapshot_archive_kind,
471        snapshot_slot,
472        snapshot_hash,
473        snapshot_storages.as_slice(),
474        &bank_snapshot_dir,
475        snapshot_archive_path,
476        snapshot_config.archive_format,
477        io_setup,
478    )?;
479
480    Ok(snapshot_archive_info)
481}
482
483/// Serializes a snapshot into `bank_snapshots_dir`
484pub fn serialize_snapshot(
485    bank_snapshots_dir: impl AsRef<Path>,
486    snapshot_version: SnapshotVersion,
487    bank_snapshot_package: BankSnapshotPackage,
488    snapshot_storages: &[Arc<AccountStorageEntry>],
489    should_finalize: bool,
490    io_setup: &IoSetupState,
491) -> Result<BankSnapshotInfo> {
492    let BankSnapshotPackage {
493        mut bank_fields,
494        bank_hash_stats,
495        status_cache_slot_deltas,
496    } = bank_snapshot_package;
497    let status_cache_slot_deltas = status_cache_slot_deltas.as_slice();
498    let slot = bank_fields.slot;
499
500    // this lambda function is to facilitate converting between
501    // the AddBankSnapshotError and SnapshotError types
502    let do_serialize_snapshot = || {
503        let mut measure_everything = Measure::start("");
504        let bank_snapshot_dir = snapshot_paths::get_bank_snapshot_dir(&bank_snapshots_dir, slot);
505        if bank_snapshot_dir.exists() {
506            return Err(AddBankSnapshotError::SnapshotDirAlreadyExists(
507                bank_snapshot_dir,
508            ));
509        }
510        fs::create_dir_all(&bank_snapshot_dir).map_err(|err| {
511            AddBankSnapshotError::CreateSnapshotDir(err, bank_snapshot_dir.clone())
512        })?;
513
514        // the bank snapshot is stored as bank_snapshots_dir/slot/slot
515        let bank_snapshot_path =
516            bank_snapshot_dir.join(snapshot_paths::get_snapshot_file_name(slot));
517        info!(
518            "Creating bank snapshot for slot {slot} at '{}'",
519            bank_snapshot_path.display(),
520        );
521
522        let bank_snapshot_serializer = move |stream: &mut dyn Write| -> Result<()> {
523            let versioned_epoch_stakes = mem::take(&mut bank_fields.versioned_epoch_stakes);
524            let extra_fields = ExtraFieldsToSerialize {
525                lamports_per_signature: bank_fields.fee_rate_governor.lamports_per_signature,
526                unused_incremental_snapshot_persistence: None,
527                unused_epoch_accounts_hash: None,
528                versioned_epoch_stakes,
529                accounts_lt_hash: Some(bank_fields.accounts_lt_hash.clone().into()),
530                block_id: Some(bank_fields.block_id),
531            };
532            serde_snapshot::serialize_bank_snapshot_into_wincode(
533                stream,
534                bank_fields,
535                bank_hash_stats,
536                extra_fields,
537            )?;
538            Ok(())
539        };
540        let (bank_snapshot_consumed_size, bank_serialize) = measure_time!(
541            serialize_snapshot_data_file(&bank_snapshot_path, io_setup, bank_snapshot_serializer)
542                .map_err(|err| AddBankSnapshotError::SerializeBank(Box::new(err)))?,
543            "bank serialize"
544        );
545
546        let status_cache_path =
547            bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME);
548        let (status_cache_consumed_size, status_cache_serialize_us) = measure_us!(
549            serde_snapshot::serialize_status_cache(
550                status_cache_slot_deltas,
551                &status_cache_path,
552                io_setup,
553            )
554            .map_err(|err| AddBankSnapshotError::SerializeStatusCache(Box::new(err)))?
555        );
556
557        let version_path = bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
558        let (_, write_version_file_us) = measure_us!(
559            fs::write(&version_path, snapshot_version.as_str().as_bytes(),)
560                .map_err(|err| AddBankSnapshotError::WriteSnapshotVersionFile(err, version_path))?
561        );
562
563        let (flush_storages_us, serialize_obsolete_accounts_us, write_storages_list_us) =
564            if should_finalize {
565                let flush_measure = Measure::start("");
566                for storage in snapshot_storages {
567                    storage.flush().map_err(|err| {
568                        AddBankSnapshotError::FlushStorage(err, storage.path().to_path_buf())
569                    })?;
570                    // We're about to mark this snapshot fastboot-loadable. Pin the storage
571                    // file so it outlives the validator-exit Drop chain.
572                    storage.disable_remove_on_drop();
573                }
574                let flush_us = flush_measure.end_as_us();
575
576                let (_, serialize_obsolete_accounts_us) = measure_us!({
577                    write_obsolete_accounts_to_snapshot(
578                        &bank_snapshot_dir,
579                        snapshot_storages,
580                        slot,
581                        io_setup,
582                    )
583                    .map_err(|err| AddBankSnapshotError::SerializeObsoleteAccounts(Box::new(err)))?
584                });
585
586                let (_, write_storages_list_us) = measure_us!(
587                    write_storages_list_to_snapshot(
588                        &bank_snapshot_dir,
589                        snapshot_storages,
590                        io_setup,
591                    )
592                    .map_err(|err| AddBankSnapshotError::WriteStoragesList(Box::new(err)))?
593                );
594
595                mark_bank_snapshot_as_loadable(&bank_snapshot_dir)
596                    .map_err(AddBankSnapshotError::MarkSnapshotLoadable)?;
597
598                (
599                    Some(flush_us),
600                    Some(serialize_obsolete_accounts_us),
601                    Some(write_storages_list_us),
602                )
603            } else {
604                (None, None, None)
605            };
606
607        measure_everything.stop();
608
609        // Monitor sizes because they're capped to MAX_SNAPSHOT_DATA_FILE_SIZE
610        datapoint_info!(
611            "snapshot_bank",
612            ("slot", slot, i64),
613            ("bank_size", bank_snapshot_consumed_size, i64),
614            ("num_storages", snapshot_storages.len(), i64),
615            ("status_cache_size", status_cache_consumed_size, i64),
616            ("flush_storages_us", flush_storages_us, Option<i64>),
617            ("serialize_obsolete_accounts_us", serialize_obsolete_accounts_us, Option<i64>),
618            ("write_storages_list_us", write_storages_list_us, Option<i64>),
619            ("bank_serialize_us", bank_serialize.as_us(), i64),
620            ("status_cache_serialize_us", status_cache_serialize_us, i64),
621            ("write_version_file_us", write_version_file_us, i64),
622            ("total_us", measure_everything.as_us(), i64),
623        );
624
625        info!(
626            "{} for slot {} at {}",
627            bank_serialize,
628            slot,
629            bank_snapshot_path.display(),
630        );
631
632        Ok(BankSnapshotInfo {
633            slot,
634            snapshot_dir: bank_snapshot_dir,
635            snapshot_version,
636            fastboot_version: None,
637        })
638    };
639
640    do_serialize_snapshot().map_err(|err| SnapshotError::AddBankSnapshot(err, slot))
641}
642
643/// Get the bank snapshots in a directory
644pub fn get_bank_snapshots(bank_snapshots_dir: impl AsRef<Path>) -> Vec<BankSnapshotInfo> {
645    let mut bank_snapshots = Vec::default();
646    match fs::read_dir(&bank_snapshots_dir) {
647        Err(err) => {
648            info!(
649                "Unable to read bank snapshots directory '{}': {err}",
650                bank_snapshots_dir.as_ref().display(),
651            );
652        }
653        Ok(paths) => paths
654            .filter_map(|entry| {
655                // check if this entry is a directory and only a Slot
656                // bank snapshots are bank_snapshots_dir/slot/slot
657                entry
658                    .ok()
659                    .filter(|entry| entry.path().is_dir())
660                    .and_then(|entry| {
661                        entry
662                            .path()
663                            .file_name()
664                            .and_then(|file_name| file_name.to_str())
665                            .and_then(|file_name| file_name.parse::<Slot>().ok())
666                    })
667            })
668            .for_each(
669                |slot| match BankSnapshotInfo::new_from_dir(&bank_snapshots_dir, slot) {
670                    Ok(snapshot_info) => bank_snapshots.push(snapshot_info),
671                    // Other threads may be modifying bank snapshots in parallel; only return
672                    // snapshots that are complete as deemed by BankSnapshotInfo::new_from_dir()
673                    Err(err) => debug!("Unable to read bank snapshot for slot {slot}: {err}"),
674                },
675            ),
676    }
677    bank_snapshots
678}
679
680/// Get the bank snapshot with the highest slot in a directory
681///
682/// This function gets the highest bank snapshot of any kind
683pub fn get_highest_bank_snapshot(bank_snapshots_dir: impl AsRef<Path>) -> Option<BankSnapshotInfo> {
684    do_get_highest_bank_snapshot(get_bank_snapshots(&bank_snapshots_dir))
685}
686
687fn do_get_highest_bank_snapshot(
688    mut bank_snapshots: Vec<BankSnapshotInfo>,
689) -> Option<BankSnapshotInfo> {
690    bank_snapshots.sort_unstable();
691    bank_snapshots.into_iter().next_back()
692}
693
694pub fn write_obsolete_accounts_to_snapshot(
695    bank_snapshot_dir: impl AsRef<Path>,
696    snapshot_storages: &[Arc<AccountStorageEntry>],
697    snapshot_slot: Slot,
698    io_setup: &IoSetupState,
699) -> Result<u64> {
700    let obsolete_accounts =
701        SerdeObsoleteAccountsMap::new_from_storages(snapshot_storages, snapshot_slot);
702    serialize_obsolete_accounts(
703        bank_snapshot_dir,
704        &obsolete_accounts,
705        MAX_OBSOLETE_ACCOUNTS_FILE_SIZE,
706        io_setup,
707    )
708}
709
710fn serialize_obsolete_accounts(
711    bank_snapshot_dir: impl AsRef<Path>,
712    obsolete_accounts_map: &SerdeObsoleteAccountsMap,
713    maximum_obsolete_accounts_file_size: u64,
714    io_setup: &IoSetupState,
715) -> Result<u64> {
716    let obsolete_accounts_path = bank_snapshot_dir
717        .as_ref()
718        .join(snapshot_paths::SNAPSHOT_OBSOLETE_ACCOUNTS_FILENAME);
719    let mut file_stream = SizeLimitedWriter::new(
720        large_file_buf_writer(&obsolete_accounts_path, io_setup)?,
721        maximum_obsolete_accounts_file_size,
722    );
723
724    serde_snapshot::serialize_into(&mut file_stream, obsolete_accounts_map).map_err(|err| {
725        IoError::other(format!(
726            "unable to serialize obsolete accounts to file '{}': {err}",
727            obsolete_accounts_path.display(),
728        ))
729    })?;
730
731    Ok(file_stream.bytes_written())
732}
733
734fn deserialize_obsolete_accounts(
735    bank_snapshot_dir: impl AsRef<Path>,
736    maximum_obsolete_accounts_file_size: u64,
737) -> Result<SerdeObsoleteAccountsMap> {
738    let obsolete_accounts_path = bank_snapshot_dir
739        .as_ref()
740        .join(snapshot_paths::SNAPSHOT_OBSOLETE_ACCOUNTS_FILENAME);
741    let obsolete_accounts_reader = ReadAdapter::new(large_file_buf_reader(
742        &obsolete_accounts_path,
743        AUX_SNAPSHOT_FILE_READ_BUF_SIZE,
744        &IoSetupState::default(),
745    )?);
746    // If the file is too large return error
747    let obsolete_accounts_file_metadata = fs::metadata(&obsolete_accounts_path)?;
748    if obsolete_accounts_file_metadata.len() > maximum_obsolete_accounts_file_size {
749        let error_message = format!(
750            "too large obsolete accounts file to deserialize: '{}' has {} bytes (max size is \
751             {maximum_obsolete_accounts_file_size} bytes)",
752            obsolete_accounts_path.display(),
753            obsolete_accounts_file_metadata.len(),
754        );
755        return Err(IoError::other(error_message).into());
756    }
757
758    Ok(serde_snapshot::deserialize_wincode_from(
759        obsolete_accounts_reader,
760    )?)
761}
762
763pub fn write_storages_list_to_snapshot(
764    bank_snapshot_dir: impl AsRef<Path>,
765    snapshot_storages: &[Arc<AccountStorageEntry>],
766    io_setup: &IoSetupState,
767) -> Result<FileSize> {
768    let storages_list = StoragesList::new_from_storages(snapshot_storages);
769    serialize_storages_list_to_snapshot(bank_snapshot_dir, storages_list, io_setup)
770}
771
772fn serialize_storages_list_to_snapshot(
773    bank_snapshot_dir: impl AsRef<Path>,
774    storages_list: StoragesList,
775    io_setup: &IoSetupState,
776) -> Result<FileSize> {
777    let storages_list_path = bank_snapshot_dir
778        .as_ref()
779        .join(snapshot_paths::SNAPSHOT_STORAGES_LIST_FILENAME);
780    let mut file_stream = SizeLimitedWriter::new(
781        large_file_buf_writer(&storages_list_path, io_setup)?,
782        MAX_STORAGES_LIST_FILE_SIZE,
783    );
784    serde_snapshot::serialize_into(&mut file_stream, &storages_list).map_err(|err| {
785        IoError::other(format!(
786            "unable to serialize storages list to file '{}': {err}",
787            storages_list_path.display(),
788        ))
789    })?;
790    Ok(file_stream.bytes_written())
791}
792
793fn deserialize_storages_list(
794    storages_list_path: &Path,
795    maximum_storages_list_file_size: u64,
796) -> Result<StoragesList> {
797    let storages_list_reader = ReadAdapter::new(large_file_buf_reader(
798        storages_list_path,
799        AUX_SNAPSHOT_FILE_READ_BUF_SIZE,
800        &IoSetupState::default(),
801    )?);
802    // If the file is too large return error
803    let storages_list_file_metadata = fs::metadata(storages_list_path)?;
804    if storages_list_file_metadata.len() > maximum_storages_list_file_size {
805        let error_message = format!(
806            "too large storages list file to deserialize: '{}' has {} bytes (max size is \
807             {maximum_storages_list_file_size} bytes)",
808            storages_list_path.display(),
809            storages_list_file_metadata.len(),
810        );
811        return Err(IoError::other(error_message).into());
812    }
813
814    Ok(serde_snapshot::deserialize_wincode_from(
815        storages_list_reader,
816    )?)
817}
818
819pub fn serialize_snapshot_data_file<F>(
820    data_file_path: &Path,
821    io_setup: &IoSetupState,
822    serializer: F,
823) -> Result<u64>
824where
825    F: FnOnce(&mut dyn Write) -> Result<()>,
826{
827    serialize_snapshot_data_file_capped::<F>(
828        data_file_path,
829        MAX_SNAPSHOT_DATA_FILE_SIZE,
830        io_setup,
831        serializer,
832    )
833}
834
835pub fn deserialize_snapshot_data_file<T: Sized>(
836    data_file_path: &Path,
837    deserializer: impl FnOnce(&mut BufReader<std::fs::File>) -> Result<T>,
838) -> Result<T> {
839    let wrapped_deserializer = move |streams: &mut SnapshotStreams<std::fs::File>| -> Result<T> {
840        deserializer(streams.full_snapshot_stream)
841    };
842
843    let wrapped_data_file_path = SnapshotRootPaths {
844        full_snapshot_root_file_path: data_file_path.to_path_buf(),
845        incremental_snapshot_root_file_path: None,
846    };
847
848    deserialize_snapshot_data_files_capped(
849        &wrapped_data_file_path,
850        MAX_SNAPSHOT_DATA_FILE_SIZE,
851        wrapped_deserializer,
852    )
853}
854
855pub fn deserialize_snapshot_data_files<T: Sized>(
856    snapshot_root_paths: &SnapshotRootPaths,
857    deserializer: impl FnOnce(&mut SnapshotStreams<std::fs::File>) -> Result<T>,
858) -> Result<T> {
859    deserialize_snapshot_data_files_capped(
860        snapshot_root_paths,
861        MAX_SNAPSHOT_DATA_FILE_SIZE,
862        deserializer,
863    )
864}
865
866fn serialize_snapshot_data_file_capped<F>(
867    data_file_path: &Path,
868    maximum_file_size: u64,
869    io_setup: &IoSetupState,
870    serializer: F,
871) -> Result<u64>
872where
873    F: FnOnce(&mut dyn Write) -> Result<()>,
874{
875    let mut data_file_stream = SizeLimitedWriter::new(
876        large_file_buf_writer(data_file_path, io_setup)?,
877        maximum_file_size,
878    );
879    serializer(&mut data_file_stream).map_err(|err| {
880        IoError::other(format!(
881            "unable to serialize snapshot data to file '{}': {err}",
882            data_file_path.display(),
883        ))
884    })?;
885    data_file_stream.flush()?;
886    Ok(data_file_stream.bytes_written())
887}
888
889fn deserialize_snapshot_data_files_capped<T: Sized>(
890    snapshot_root_paths: &SnapshotRootPaths,
891    maximum_file_size: u64,
892    deserializer: impl FnOnce(&mut SnapshotStreams<std::fs::File>) -> Result<T>,
893) -> Result<T> {
894    let (full_snapshot_file_size, mut full_snapshot_data_file_stream) =
895        create_snapshot_data_file_stream(
896            &snapshot_root_paths.full_snapshot_root_file_path,
897            maximum_file_size,
898        )?;
899
900    let (incremental_snapshot_file_size, mut incremental_snapshot_data_file_stream) =
901        if let Some(ref incremental_snapshot_root_file_path) =
902            snapshot_root_paths.incremental_snapshot_root_file_path
903        {
904            Some(create_snapshot_data_file_stream(
905                incremental_snapshot_root_file_path,
906                maximum_file_size,
907            )?)
908        } else {
909            None
910        }
911        .unzip();
912
913    let mut snapshot_streams = SnapshotStreams {
914        full_snapshot_stream: &mut full_snapshot_data_file_stream,
915        incremental_snapshot_stream: incremental_snapshot_data_file_stream.as_mut(),
916    };
917    let ret = deserializer(&mut snapshot_streams)?;
918
919    check_deserialize_file_consumed(
920        full_snapshot_file_size,
921        &snapshot_root_paths.full_snapshot_root_file_path,
922        &mut full_snapshot_data_file_stream,
923    )?;
924
925    if let Some(ref incremental_snapshot_root_file_path) =
926        snapshot_root_paths.incremental_snapshot_root_file_path
927    {
928        check_deserialize_file_consumed(
929            incremental_snapshot_file_size.unwrap(),
930            incremental_snapshot_root_file_path,
931            incremental_snapshot_data_file_stream.as_mut().unwrap(),
932        )?;
933    }
934
935    Ok(ret)
936}
937
938/// Before running the deserializer function, perform common operations on the snapshot archive
939/// files, such as checking the file size and opening the file into a stream.
940fn create_snapshot_data_file_stream(
941    snapshot_root_file_path: impl AsRef<Path>,
942    maximum_file_size: u64,
943) -> Result<(u64, BufReader<std::fs::File>)> {
944    let snapshot_file_size = fs::metadata(&snapshot_root_file_path)?.len();
945
946    if snapshot_file_size > maximum_file_size {
947        let error_message = format!(
948            "too large snapshot data file to deserialize: '{}' has {} bytes (max size is {} bytes)",
949            snapshot_root_file_path.as_ref().display(),
950            snapshot_file_size,
951            maximum_file_size,
952        );
953        return Err(IoError::other(error_message).into());
954    }
955
956    let snapshot_data_file = fs::File::open(snapshot_root_file_path)?;
957    let snapshot_data_file_stream = BufReader::new(snapshot_data_file);
958
959    Ok((snapshot_file_size, snapshot_data_file_stream))
960}
961
962/// After running the deserializer function, perform common checks to ensure the snapshot archive
963/// files were consumed correctly.
964fn check_deserialize_file_consumed(
965    file_size: u64,
966    file_path: impl AsRef<Path>,
967    file_stream: &mut BufReader<std::fs::File>,
968) -> Result<()> {
969    let consumed_size = file_stream.stream_position()?;
970
971    if consumed_size != file_size {
972        let error_message = format!(
973            "invalid snapshot data file: '{}' has {} bytes, however consumed {} bytes to \
974             deserialize",
975            file_path.as_ref().display(),
976            file_size,
977            consumed_size,
978        );
979        return Err(IoError::other(error_message).into());
980    }
981
982    Ok(())
983}
984
985/// Unarchives the given full and incremental snapshot archives, as long as they are compatible.
986pub fn verify_and_unarchive_snapshots(
987    bank_snapshots_dir: impl AsRef<Path>,
988    full_snapshot_archive_info: &FullSnapshotArchiveInfo,
989    incremental_snapshot_archive_info: Option<&IncrementalSnapshotArchiveInfo>,
990    account_paths: &[PathBuf],
991    io_setup: &IoSetupState,
992) -> Result<(UnarchivedSnapshots, UnarchivedSnapshotsGuard)> {
993    check_are_snapshots_compatible(
994        full_snapshot_archive_info,
995        incremental_snapshot_archive_info,
996    )?;
997
998    let next_append_vec_id = Arc::new(AtomicAccountsFileId::new(0));
999    let UnarchivedSnapshot {
1000        unpack_dir: full_unpack_dir,
1001        storage: full_storage,
1002        bank_fields: full_bank_fields,
1003        accounts_db_fields: full_accounts_db_fields,
1004        unpacked_snapshots_dir_and_version: full_unpacked_snapshots_dir_and_version,
1005        measure_untar: full_measure_untar,
1006    } = unarchive_snapshot(
1007        &bank_snapshots_dir,
1008        snapshot_paths::TMP_SNAPSHOT_ARCHIVE_PREFIX,
1009        full_snapshot_archive_info.path(),
1010        "snapshot untar",
1011        account_paths,
1012        full_snapshot_archive_info.archive_format(),
1013        next_append_vec_id.clone(),
1014        io_setup,
1015    )?;
1016
1017    let (
1018        incremental_unpack_dir,
1019        incremental_storage,
1020        incremental_bank_fields,
1021        incremental_accounts_db_fields,
1022        incremental_unpacked_snapshots_dir_and_version,
1023        incremental_measure_untar,
1024    ) = if let Some(incremental_snapshot_archive_info) = incremental_snapshot_archive_info {
1025        let UnarchivedSnapshot {
1026            unpack_dir,
1027            storage,
1028            bank_fields,
1029            accounts_db_fields,
1030            unpacked_snapshots_dir_and_version,
1031            measure_untar,
1032        } = unarchive_snapshot(
1033            &bank_snapshots_dir,
1034            snapshot_paths::TMP_SNAPSHOT_ARCHIVE_PREFIX,
1035            incremental_snapshot_archive_info.path(),
1036            "incremental snapshot untar",
1037            account_paths,
1038            incremental_snapshot_archive_info.archive_format(),
1039            next_append_vec_id.clone(),
1040            io_setup,
1041        )?;
1042        (
1043            Some(unpack_dir),
1044            Some(storage),
1045            Some(bank_fields),
1046            Some(accounts_db_fields),
1047            Some(unpacked_snapshots_dir_and_version),
1048            Some(measure_untar),
1049        )
1050    } else {
1051        (None, None, None, None, None, None)
1052    };
1053
1054    let bank_fields = SnapshotBankFields::new(full_bank_fields, incremental_bank_fields);
1055    let accounts_db_fields =
1056        SnapshotAccountsDbFields::new(full_accounts_db_fields, incremental_accounts_db_fields);
1057    let next_append_vec_id = Arc::try_unwrap(next_append_vec_id).unwrap();
1058
1059    Ok((
1060        UnarchivedSnapshots {
1061            full_storage,
1062            incremental_storage,
1063            bank_fields,
1064            accounts_db_fields,
1065            full_unpacked_snapshots_dir_and_version,
1066            incremental_unpacked_snapshots_dir_and_version,
1067            full_measure_untar,
1068            incremental_measure_untar,
1069            next_append_vec_id,
1070        },
1071        UnarchivedSnapshotsGuard {
1072            full_unpack_dir,
1073            incremental_unpack_dir,
1074        },
1075    ))
1076}
1077
1078/// Used to determine if a filename is structured like a version file, bank file, or storage file
1079#[derive(PartialEq, Debug)]
1080enum SnapshotFileKind {
1081    Version,
1082    BankFields,
1083    Storage,
1084}
1085
1086/// Determines `SnapshotFileKind` for `filename` if any
1087fn get_snapshot_file_kind(filename: &str) -> Option<SnapshotFileKind> {
1088    static VERSION_FILE_REGEX: LazyLock<Regex> =
1089        LazyLock::new(|| Regex::new(r"^version$").unwrap());
1090    static BANK_FIELDS_FILE_REGEX: LazyLock<Regex> =
1091        LazyLock::new(|| Regex::new(r"^[0-9]+(\.pre)?$").unwrap());
1092
1093    if VERSION_FILE_REGEX.is_match(filename) {
1094        Some(SnapshotFileKind::Version)
1095    } else if BANK_FIELDS_FILE_REGEX.is_match(filename) {
1096        Some(SnapshotFileKind::BankFields)
1097    } else if get_slot_and_append_vec_id(filename).is_ok() {
1098        Some(SnapshotFileKind::Storage)
1099    } else {
1100        None
1101    }
1102}
1103
1104/// Waits for snapshot file
1105/// Due to parallel unpacking, we may receive some append_vec files before the snapshot file
1106/// This function will push append_vec files into a buffer until we receive the snapshot file
1107fn get_version_and_snapshot_files(
1108    file_receiver: &Receiver<FileInfo>,
1109) -> Result<(FileInfo, FileInfo, Vec<FileInfo>)> {
1110    let mut append_vec_files = Vec::with_capacity(1024);
1111    let mut snapshot_version = None;
1112    let mut snapshot_bank = None;
1113
1114    loop {
1115        if let Ok(file_info) = file_receiver.recv() {
1116            let filename = file_info.path.file_name().unwrap().to_str().unwrap();
1117            match get_snapshot_file_kind(filename) {
1118                Some(SnapshotFileKind::Version) => {
1119                    snapshot_version = Some(file_info);
1120
1121                    // break if we have both the snapshot file and the version file
1122                    if snapshot_bank.is_some() {
1123                        break;
1124                    }
1125                }
1126                Some(SnapshotFileKind::BankFields) => {
1127                    snapshot_bank = Some(file_info);
1128
1129                    // break if we have both the snapshot file and the version file
1130                    if snapshot_version.is_some() {
1131                        break;
1132                    }
1133                }
1134                Some(SnapshotFileKind::Storage) => {
1135                    append_vec_files.push(file_info);
1136                }
1137                None => {} // do nothing for other kinds of files
1138            }
1139        } else {
1140            return Err(SnapshotError::RebuildStorages(
1141                "did not receive snapshot file from unpacking threads".to_string(),
1142            ));
1143        }
1144    }
1145    let snapshot_version = snapshot_version.unwrap();
1146    let snapshot_bank = snapshot_bank.unwrap();
1147
1148    Ok((snapshot_version, snapshot_bank, append_vec_files))
1149}
1150
1151/// Fields and information parsed from the snapshot.
1152struct SnapshotFieldsBundle {
1153    snapshot_version: SnapshotVersion,
1154    bank_fields: BankFieldsToDeserialize,
1155    accounts_db_fields: AccountsDbFields,
1156    append_vec_files: Vec<FileInfo>,
1157}
1158
1159/// Parses fields and information from the snapshot files provided by
1160/// `file_receiver`.
1161fn snapshot_fields_from_files(file_receiver: &Receiver<FileInfo>) -> Result<SnapshotFieldsBundle> {
1162    let (snapshot_version, snapshot_bank, append_vec_files) =
1163        get_version_and_snapshot_files(file_receiver)?;
1164    let snapshot_version_str = snapshot_version_from_file(snapshot_version)?;
1165    let snapshot_version = snapshot_version_str.parse().map_err(|err| {
1166        IoError::other(format!(
1167            "unsupported snapshot version '{snapshot_version_str}': {err}",
1168        ))
1169    })?;
1170
1171    let mut snapshot_stream = BufReader::new(snapshot_bank.file);
1172    let (bank_fields, accounts_db_fields) = match snapshot_version {
1173        SnapshotVersion::V1_2_0 => serde_snapshot::fields_from_stream(&mut snapshot_stream)?,
1174    };
1175
1176    Ok(SnapshotFieldsBundle {
1177        snapshot_version,
1178        bank_fields,
1179        accounts_db_fields,
1180        append_vec_files,
1181    })
1182}
1183
1184/// BankSnapshotInfo::new_from_dir() requires a few meta files to accept a snapshot dir
1185/// as a valid one.  A dir unpacked from an archive lacks these files.  Fill them here to
1186/// allow new_from_dir() checks to pass.  These checks are not needed for unpacked dirs,
1187/// but it is not clean to add another flag to new_from_dir() to skip them.
1188fn create_snapshot_meta_files_for_unarchived_snapshot(unpack_dir: impl AsRef<Path>) -> Result<()> {
1189    let snapshots_dir = unpack_dir.as_ref().join(snapshot_paths::BANK_SNAPSHOTS_DIR);
1190    if !snapshots_dir.is_dir() {
1191        return Err(SnapshotError::NoSnapshotSlotDir(snapshots_dir));
1192    }
1193
1194    // The unpacked dir has a single slot dir, which is the snapshot slot dir.
1195    let slot_dir = std::fs::read_dir(&snapshots_dir)
1196        .map_err(|_| SnapshotError::NoSnapshotSlotDir(snapshots_dir.clone()))?
1197        .find(|entry| entry.as_ref().unwrap().path().is_dir())
1198        .ok_or_else(|| SnapshotError::NoSnapshotSlotDir(snapshots_dir.clone()))?
1199        .map_err(|_| SnapshotError::NoSnapshotSlotDir(snapshots_dir.clone()))?
1200        .path();
1201
1202    let version_file = unpack_dir
1203        .as_ref()
1204        .join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
1205    fs::hard_link(
1206        version_file,
1207        slot_dir.join(snapshot_paths::SNAPSHOT_VERSION_FILENAME),
1208    )?;
1209
1210    let status_cache_file = snapshots_dir.join(snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME);
1211    fs::hard_link(
1212        status_cache_file,
1213        slot_dir.join(snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME),
1214    )?;
1215
1216    Ok(())
1217}
1218
1219/// Perform the common tasks when unarchiving a snapshot.  Handles creating the temporary
1220/// directories, untaring, reading the version file, and then returning those fields plus the
1221/// rebuilt storage
1222#[allow(clippy::too_many_arguments)]
1223fn unarchive_snapshot(
1224    bank_snapshots_dir: impl AsRef<Path>,
1225    unpacked_snapshots_dir_prefix: &'static str,
1226    snapshot_archive_path: impl AsRef<Path>,
1227    measure_name: &'static str,
1228    account_paths: &[PathBuf],
1229    archive_format: ArchiveFormat,
1230    next_append_vec_id: Arc<AtomicAccountsFileId>,
1231    io_setup: &IoSetupState,
1232) -> Result<UnarchivedSnapshot> {
1233    let unpack_dir = tempfile::Builder::new()
1234        .prefix(unpacked_snapshots_dir_prefix)
1235        .tempdir_in(bank_snapshots_dir)?;
1236    let unpacked_snapshots_dir = unpack_dir.path().join(snapshot_paths::BANK_SNAPSHOTS_DIR);
1237
1238    let (file_sender, file_receiver) = crossbeam_channel::unbounded();
1239    thread::scope(|scope| {
1240        let unarchive_handle = streaming_unarchive_snapshot(
1241            scope,
1242            file_sender,
1243            account_paths.to_vec(),
1244            unpack_dir.path().to_path_buf(),
1245            snapshot_archive_path.as_ref().to_path_buf(),
1246            archive_format,
1247            io_setup,
1248        );
1249
1250        let snapshot_result = snapshot_fields_from_files(&file_receiver).and_then(
1251            |SnapshotFieldsBundle {
1252                 snapshot_version,
1253                 bank_fields,
1254                 accounts_db_fields,
1255                 append_vec_files,
1256                 ..
1257             }| {
1258                let (storage, measure_untar) = measure_time!(
1259                    SnapshotStorageRebuilder::rebuild_storages(
1260                        append_vec_files.into_iter().chain(file_receiver),
1261                        next_append_vec_id,
1262                        SnapshotFrom::Archive,
1263                        None,
1264                    )?,
1265                    measure_name
1266                );
1267                info!("{measure_untar}");
1268                create_snapshot_meta_files_for_unarchived_snapshot(&unpack_dir)?;
1269
1270                Ok(UnarchivedSnapshot {
1271                    unpack_dir,
1272                    storage,
1273                    bank_fields,
1274                    accounts_db_fields,
1275                    unpacked_snapshots_dir_and_version: UnpackedSnapshotsDirAndVersion {
1276                        unpacked_snapshots_dir,
1277                        snapshot_version,
1278                    },
1279                    measure_untar,
1280                })
1281            },
1282        );
1283        // Producer errors are usually the root cause (no files -> no reception).
1284        let unarchive_result = unarchive_handle.join().expect("must join unarchive thread");
1285        match (unarchive_result, snapshot_result) {
1286            // Rebuilder closed the receiver early; the producer's send failure is just the
1287            // downstream symptom — surface the rebuilder's error instead.
1288            (Err(SnapshotError::CrossbeamSend(_)), snap @ Err(_)) => snap,
1289            (Err(err), _) => Err(err),
1290            (Ok(()), snap) => snap,
1291        }
1292    })
1293}
1294
1295/// Spawn thread that streams snapshot dir files across channel
1296///
1297/// Follow the flow of streaming_unarchive_snapshot(), but handle the from_dir case.
1298fn spawn_streaming_snapshot_dir_files(
1299    snapshot_file_path: PathBuf,
1300    snapshot_version_path: PathBuf,
1301    account_paths: &[PathBuf],
1302) -> (Receiver<FileInfo>, thread::JoinHandle<Result<()>>) {
1303    let (file_sender, file_receiver) = crossbeam_channel::unbounded();
1304    let account_paths = account_paths.to_vec();
1305
1306    let handle = thread::Builder::new()
1307        .name("solSnapDirFiles".to_string())
1308        .spawn(move || {
1309            let snapshot_bank_file_info = FileInfo::new_from_path(snapshot_file_path)?;
1310            file_sender.send(snapshot_bank_file_info)?;
1311            let snapshot_version_file_info = FileInfo::new_from_path(snapshot_version_path)?;
1312            file_sender.send(snapshot_version_file_info)?;
1313
1314            for account_path in account_paths {
1315                for dir_entry_result in fs::read_dir(account_path)? {
1316                    let dir_entry = dir_entry_result?;
1317                    let path = dir_entry.path();
1318                    let file_info = FileInfo::new_from_path(path)?;
1319                    file_sender.send(file_info)?;
1320                }
1321            }
1322            Ok::<_, SnapshotError>(())
1323        })
1324        .expect("should spawn thread");
1325
1326    (file_receiver, handle)
1327}
1328
1329/// Migrates a legacy (2.0.0) bank snapshot's hardlink-based storages into the new format.
1330///
1331/// Walks `<bank_snapshot_dir>/accounts_hardlinks/`, follows each symlink to its
1332/// `<account_path>/snapshot/<slot>/` target, and renames each storage file there back into
1333/// `<account_path>/run/`. Writes the derived storages list into the bank snapshot dir so the
1334/// load path can always read it from disk. Tears down the legacy directories (the
1335/// `accounts_hardlinks/` symlink dir and the whole `<account_path>/snapshot/` tree) on success
1336/// so subsequent restarts go through the normal new-format path.
1337fn migrate_legacy_hardlinks(bank_snapshot_dir: &Path, account_run_paths: &[PathBuf]) -> Result<()> {
1338    let accounts_hardlinks_dir =
1339        bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_ACCOUNTS_HARDLINKS);
1340    let mut items: Vec<StorageListItem> = Vec::new();
1341
1342    for entry in fs::read_dir(&accounts_hardlinks_dir).map_err(|err| {
1343        IoError::other(format!(
1344            "failed to read legacy accounts hardlinks dir '{}': {err}",
1345            accounts_hardlinks_dir.display(),
1346        ))
1347    })? {
1348        let symlink_path = entry?.path();
1349        let snapshot_slot_dir = fs::read_link(&symlink_path).map_err(|err| {
1350            IoError::other(format!(
1351                "failed to read symlink '{}': {err}",
1352                symlink_path.display(),
1353            ))
1354        })?;
1355        // snapshot_slot_dir = `<X>/snapshot/<slot>/`. The account run dir is its
1356        // grandparent + `run` (i.e. `<X>/run`).
1357        let run_dir = snapshot_slot_dir
1358            .parent()
1359            .and_then(Path::parent)
1360            .ok_or_else(|| {
1361                IoError::other(format!(
1362                    "invalid legacy hardlink target '{}'",
1363                    snapshot_slot_dir.display(),
1364                ))
1365            })?
1366            .join(ACCOUNTS_RUN_DIR);
1367        // The legacy snapshot was taken against the account paths in use at the time. If
1368        // those have changed (e.g. the operator reconfigured `account_paths` while upgrading
1369        // Agave), the run dir we just derived isn't one we're loading into — bail rather than
1370        // silently writing files into a location nobody's reading from.
1371        if !account_run_paths.contains(&run_dir) {
1372            return Err(IoError::other(format!(
1373                "legacy hardlink target '{}' points to run dir '{}' which is not in the current \
1374                 account paths ({:?}); the account paths configuration has changed since this \
1375                 snapshot was taken — load from a snapshot archive instead",
1376                snapshot_slot_dir.display(),
1377                run_dir.display(),
1378                account_run_paths,
1379            ))
1380            .into());
1381        }
1382
1383        for file_entry in fs::read_dir(&snapshot_slot_dir).map_err(|err| {
1384            IoError::other(format!(
1385                "failed to read legacy hardlink dir '{}': {err}",
1386                snapshot_slot_dir.display(),
1387            ))
1388        })? {
1389            let src = file_entry?.path();
1390            let Some(name) = src.file_name().and_then(|n| n.to_str()) else {
1391                continue;
1392            };
1393            let (slot, id) = get_slot_and_append_vec_id(name)?;
1394            let dest = run_dir.join(name);
1395            fs::rename(&src, &dest).map_err(|err| {
1396                IoError::other(format!(
1397                    "failed to migrate legacy storage from '{}' to '{}': {err}",
1398                    src.display(),
1399                    dest.display(),
1400                ))
1401            })?;
1402            items.push(StorageListItem {
1403                slot,
1404                id: id as AccountsFileId,
1405            });
1406        }
1407    }
1408
1409    // Persist the derived list now: migration is destructive (it removes the legacy hardlinks
1410    // below), and writing the storages list is what actually brings the bank snapshot to
1411    // fastboot version >=3 compatibility. Doing it here means the snapshot stays loadable even
1412    // if the validator never performs a proper teardown (e.g. crashes).
1413    serialize_storages_list_to_snapshot(
1414        bank_snapshot_dir,
1415        StoragesList::from_items(items),
1416        &IoSetupState::default(),
1417    )?;
1418
1419    // Tear down the legacy state so we don't repeat this migration: drop the bank snapshot's
1420    // `accounts_hardlinks/` symlink dir and wipe each `<account_path>/snapshot/` tree (catches
1421    // both the per-slot dirs we just emptied and any orphans from older purged snapshots).
1422    fs::remove_dir_all(&accounts_hardlinks_dir).map_err(|err| {
1423        IoError::other(format!(
1424            "failed to remove legacy accounts hardlinks dir '{}': {err}",
1425            accounts_hardlinks_dir.display(),
1426        ))
1427    })?;
1428    wipe_account_snapshot_dirs(account_run_paths);
1429
1430    // Bump the fastboot version so subsequent loads take the normal 3.0+ path instead of
1431    // re-running the migration (which would fail now that the hardlinks dir is gone).
1432    mark_bank_snapshot_as_loadable(bank_snapshot_dir)?;
1433
1434    Ok(())
1435}
1436
1437/// Removes storage files from `account_paths` whose `(slot, id)` pair isn't listed in the
1438/// storages list (i.e. they don't belong to the snapshot being loaded). Files whose names
1439/// don't parse as `<slot>.<id>` storage filenames are left alone.
1440fn prune_stale_storages(account_paths: &[PathBuf], storages_list: StoragesList) -> Result<()> {
1441    let expected_storages = storages_list.into_slot_file_id_set();
1442    for account_path in account_paths {
1443        let read_dir = fs::read_dir(account_path).map_err(|err| {
1444            IoError::other(format!(
1445                "failed to read account path '{}': {err}",
1446                account_path.display(),
1447            ))
1448        })?;
1449        for entry in read_dir {
1450            let path = entry?.path();
1451            let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
1452                continue;
1453            };
1454            let Ok((slot, id)) = get_slot_and_append_vec_id(name) else {
1455                // Not a storage file name — leave it alone.
1456                continue;
1457            };
1458            if !expected_storages.contains(&(slot, id as AccountsFileId)) {
1459                info!(
1460                    "Removing stale storage file '{}' not in storages list",
1461                    path.display(),
1462                );
1463                fs::remove_file(&path)?
1464            }
1465        }
1466    }
1467    Ok(())
1468}
1469
1470/// Performs the common tasks when deserializing a snapshot
1471///
1472/// Handles reading the snapshot file and version file,
1473/// then returning those fields plus the rebuilt storages.
1474pub(crate) fn rebuild_storages_from_snapshot_dir(
1475    snapshot_info: &BankSnapshotInfo,
1476    account_paths: &[PathBuf],
1477    next_append_vec_id: Arc<AtomicAccountsFileId>,
1478) -> Result<(AccountStorageMap, BankFieldsToDeserialize, AccountsDbFields)> {
1479    let bank_snapshot_dir = &snapshot_info.snapshot_dir;
1480
1481    // With fastboot_version >= 2, obsolete accounts are tracked and stored in the snapshot
1482    // Even if obsolete accounts are not enabled, the snapshot may still contain obsolete accounts
1483    // as the feature may have been enabled in previous validator runs.
1484    let obsolete_accounts = snapshot_info
1485        .fastboot_version
1486        .as_ref()
1487        .is_some_and(|fastboot_version| fastboot_version.major >= 2)
1488        .then(|| deserialize_obsolete_accounts(bank_snapshot_dir, MAX_OBSOLETE_ACCOUNTS_FILE_SIZE))
1489        .transpose()
1490        .map_err(|err| {
1491            IoError::other(format!(
1492                "failed to read obsolete accounts file '{}': {err}",
1493                bank_snapshot_dir.display()
1494            ))
1495        })?;
1496
1497    // The bank snapshot lists the storage files belonging to it. Anything else in the account
1498    // paths is from a later (post-snapshot) slot and must be removed before we load — the
1499    // lt hash check at startup verifies the surviving storages.
1500    let storages_list_path =
1501        bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_STORAGES_LIST_FILENAME);
1502    if !storages_list_path.exists() {
1503        // Legacy (2.0.0) bank snapshot: storages live as hardlinks under
1504        // `<account_path>/snapshot/<slot>/`, with symlinks in
1505        // `<bank_snapshot_dir>/accounts_hardlinks/` tying them to the bank snapshot. Move the
1506        // files back into `<account_path>/run/` and write out the storages list so the load
1507        // path below can read it like any other 3.0+ snapshot.
1508        migrate_legacy_hardlinks(bank_snapshot_dir, account_paths)?;
1509    }
1510    let storages_list =
1511        deserialize_storages_list(&storages_list_path, MAX_STORAGES_LIST_FILE_SIZE)?;
1512    prune_stale_storages(account_paths, storages_list)?;
1513
1514    let snapshot_file_path = snapshot_info.snapshot_path();
1515    let snapshot_version_path = bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
1516    let (file_receiver, stream_files_handle) = spawn_streaming_snapshot_dir_files(
1517        snapshot_file_path,
1518        snapshot_version_path,
1519        account_paths,
1520    );
1521
1522    let snapshot_result = snapshot_fields_from_files(&file_receiver).and_then(
1523        |SnapshotFieldsBundle {
1524             bank_fields,
1525             accounts_db_fields,
1526             append_vec_files,
1527             ..
1528         }| {
1529            let storage = SnapshotStorageRebuilder::rebuild_storages(
1530                append_vec_files.into_iter().chain(file_receiver),
1531                next_append_vec_id,
1532                SnapshotFrom::Dir,
1533                obsolete_accounts,
1534            )?;
1535            Ok((storage, bank_fields, accounts_db_fields))
1536        },
1537    );
1538
1539    // Producer errors are usually the root cause (no files -> no reception).
1540    let stream_files_result = stream_files_handle.join().expect("must join dir thread");
1541    match (stream_files_result, snapshot_result) {
1542        // Rebuilder closed the receiver early; the producer's send failure is just the
1543        // downstream symptom — surface the rebuilder's error instead.
1544        (Err(SnapshotError::CrossbeamSend(_)), snap @ Err(_)) => snap,
1545        (Err(err), _) => Err(err),
1546        (Ok(()), snap) => snap,
1547    }
1548}
1549
1550/// Reads the `snapshot_version` from a file. Before opening the file, its size
1551/// is compared to `MAX_SNAPSHOT_VERSION_FILE_SIZE`. If the size exceeds this
1552/// threshold, it is not opened and an error is returned.
1553fn snapshot_version_from_file(mut file_info: FileInfo) -> io::Result<String> {
1554    let file_size = file_info.size;
1555    if file_size > MAX_SNAPSHOT_VERSION_FILE_SIZE {
1556        let error_message = format!(
1557            "snapshot version file too large: '{}' has {} bytes (max size is {} bytes)",
1558            file_info.path.display(),
1559            file_size,
1560            MAX_SNAPSHOT_VERSION_FILE_SIZE,
1561        );
1562        return Err(IoError::other(error_message));
1563    }
1564
1565    // Read snapshot_version from file.
1566    let mut snapshot_version = String::new();
1567    file_info
1568        .file
1569        .read_to_string(&mut snapshot_version)
1570        .map_err(|err| {
1571            IoError::other(format!(
1572                "failed to read snapshot version from file '{}': {err}",
1573                file_info.path.display()
1574            ))
1575        })?;
1576
1577    Ok(snapshot_version.trim().to_string())
1578}
1579
1580/// Check if an incremental snapshot is compatible with a full snapshot.  This is done by checking
1581/// if the incremental snapshot's base slot is the same as the full snapshot's slot.
1582fn check_are_snapshots_compatible(
1583    full_snapshot_archive_info: &FullSnapshotArchiveInfo,
1584    incremental_snapshot_archive_info: Option<&IncrementalSnapshotArchiveInfo>,
1585) -> Result<()> {
1586    if incremental_snapshot_archive_info.is_none() {
1587        return Ok(());
1588    }
1589
1590    let incremental_snapshot_archive_info = incremental_snapshot_archive_info.unwrap();
1591
1592    (full_snapshot_archive_info.slot() == incremental_snapshot_archive_info.base_slot())
1593        .then_some(())
1594        .ok_or_else(|| {
1595            SnapshotError::MismatchedBaseSlot(
1596                full_snapshot_archive_info.slot(),
1597                incremental_snapshot_archive_info.base_slot(),
1598            )
1599        })
1600}
1601
1602pub fn purge_old_snapshot_archives(
1603    full_snapshot_archives_dir: impl AsRef<Path>,
1604    incremental_snapshot_archives_dir: impl AsRef<Path>,
1605    maximum_full_snapshot_archives_to_retain: NonZeroUsize,
1606    maximum_incremental_snapshot_archives_to_retain: NonZeroUsize,
1607) {
1608    info!(
1609        "Purging old full snapshot archives in {}, retaining up to {} full snapshots",
1610        full_snapshot_archives_dir.as_ref().display(),
1611        maximum_full_snapshot_archives_to_retain
1612    );
1613
1614    let mut full_snapshot_archives =
1615        snapshot_paths::full_snapshot_archives_iter(full_snapshot_archives_dir.as_ref())
1616            .collect::<Vec<_>>();
1617    full_snapshot_archives.sort_unstable();
1618    full_snapshot_archives.reverse();
1619
1620    let num_to_retain = full_snapshot_archives
1621        .len()
1622        .min(maximum_full_snapshot_archives_to_retain.get());
1623    trace!(
1624        "There are {} full snapshot archives, retaining {}",
1625        full_snapshot_archives.len(),
1626        num_to_retain,
1627    );
1628
1629    let (full_snapshot_archives_to_retain, full_snapshot_archives_to_remove) =
1630        if full_snapshot_archives.is_empty() {
1631            None
1632        } else {
1633            Some(full_snapshot_archives.split_at(num_to_retain))
1634        }
1635        .unwrap_or_default();
1636
1637    let retained_full_snapshot_slots = full_snapshot_archives_to_retain
1638        .iter()
1639        .map(|ai| ai.slot())
1640        .collect::<HashSet<_>>();
1641
1642    fn remove_archives<T: SnapshotArchiveInfoGetter>(archives: &[T]) {
1643        for path in archives.iter().map(|a| a.path()) {
1644            trace!("Removing snapshot archive: {}", path.display());
1645            let result = fs::remove_file(path);
1646            if let Err(err) = result {
1647                info!(
1648                    "Failed to remove snapshot archive '{}': {err}",
1649                    path.display()
1650                );
1651            }
1652        }
1653    }
1654    remove_archives(full_snapshot_archives_to_remove);
1655
1656    info!(
1657        "Purging old incremental snapshot archives in {}, retaining up to {} incremental snapshots",
1658        incremental_snapshot_archives_dir.as_ref().display(),
1659        maximum_incremental_snapshot_archives_to_retain
1660    );
1661    let mut incremental_snapshot_archives_by_base_slot = HashMap::<Slot, Vec<_>>::new();
1662    for incremental_snapshot_archive in
1663        incremental_snapshot_archives_iter(incremental_snapshot_archives_dir.as_ref())
1664    {
1665        incremental_snapshot_archives_by_base_slot
1666            .entry(incremental_snapshot_archive.base_slot())
1667            .or_default()
1668            .push(incremental_snapshot_archive)
1669    }
1670
1671    let highest_full_snapshot_slot = retained_full_snapshot_slots.iter().max().copied();
1672    for (base_slot, mut incremental_snapshot_archives) in incremental_snapshot_archives_by_base_slot
1673    {
1674        incremental_snapshot_archives.sort_unstable();
1675        let num_to_retain = if Some(base_slot) == highest_full_snapshot_slot {
1676            maximum_incremental_snapshot_archives_to_retain.get()
1677        } else {
1678            usize::from(retained_full_snapshot_slots.contains(&base_slot))
1679        };
1680        trace!(
1681            "There are {} incremental snapshot archives for base slot {}, removing {} of them",
1682            incremental_snapshot_archives.len(),
1683            base_slot,
1684            incremental_snapshot_archives
1685                .len()
1686                .saturating_sub(num_to_retain),
1687        );
1688
1689        incremental_snapshot_archives.truncate(
1690            incremental_snapshot_archives
1691                .len()
1692                .saturating_sub(num_to_retain),
1693        );
1694        remove_archives(&incremental_snapshot_archives);
1695    }
1696}
1697
1698pub fn verify_unpacked_snapshots_dir_and_version(
1699    unpacked_snapshots_dir_and_version: &UnpackedSnapshotsDirAndVersion,
1700) -> Result<(SnapshotVersion, BankSnapshotInfo)> {
1701    info!(
1702        "snapshot version: {}",
1703        unpacked_snapshots_dir_and_version.snapshot_version
1704    );
1705
1706    let snapshot_version = unpacked_snapshots_dir_and_version.snapshot_version;
1707    let mut bank_snapshots =
1708        get_bank_snapshots(&unpacked_snapshots_dir_and_version.unpacked_snapshots_dir);
1709    if bank_snapshots.len() > 1 {
1710        return Err(IoError::other(format!(
1711            "invalid snapshot format: only one snapshot allowed, but found {}",
1712            bank_snapshots.len(),
1713        ))
1714        .into());
1715    }
1716    let root_paths = bank_snapshots.pop().ok_or_else(|| {
1717        IoError::other(format!(
1718            "no snapshots found in snapshots directory '{}'",
1719            unpacked_snapshots_dir_and_version
1720                .unpacked_snapshots_dir
1721                .display(),
1722        ))
1723    })?;
1724    Ok((snapshot_version, root_paths))
1725}
1726
1727#[derive(Debug, Copy, Clone)]
1728/// allow tests to specify what happened to the serialized format
1729pub enum VerifyBank {
1730    /// the bank's serialized format is expected to be identical to what we are comparing against
1731    Deterministic,
1732    /// the serialized bank was 'reserialized' into a non-deterministic format
1733    /// so, deserialize both files and compare deserialized results
1734    NonDeterministic,
1735}
1736
1737/// For each account run dir, wipes the sibling `snapshot/` dir.
1738///
1739/// Validator account paths are laid out as a parent containing two siblings: `run/` (the live
1740/// storage files) and `snapshot/` (legacy per-slot hardlink dirs, pre-3.0). `account_run_paths`
1741/// here holds the `run/` paths, so for each entry we walk up to the parent and wipe the
1742/// `snapshot/` sibling. Nothing new is written under `snapshot/` anymore, so any content is
1743/// either legacy hardlink dirs (written by pre-3.0 validators during fastboot) or orphans
1744/// from purged bank snapshots. Used by the legacy-hardlink migration path and the
1745/// archive-load path to drop that leftover state.
1746pub fn wipe_account_snapshot_dirs(account_run_paths: &[PathBuf]) {
1747    for account_run_path in account_run_paths {
1748        if let Some(parent) = account_run_path.parent() {
1749            move_and_async_delete_path_contents(parent.join(ACCOUNTS_SNAPSHOT_DIR));
1750        }
1751    }
1752}
1753
1754/// Purges all bank snapshots
1755pub fn purge_all_bank_snapshots(bank_snapshots_dir: impl AsRef<Path>) {
1756    let bank_snapshots = get_bank_snapshots(&bank_snapshots_dir);
1757    purge_bank_snapshots(&bank_snapshots);
1758}
1759
1760/// Purges bank snapshots, retaining the newest `num_bank_snapshots_to_retain`
1761pub fn purge_old_bank_snapshots(
1762    bank_snapshots_dir: impl AsRef<Path>,
1763    num_bank_snapshots_to_retain: usize,
1764) {
1765    let mut bank_snapshots = get_bank_snapshots(&bank_snapshots_dir);
1766
1767    bank_snapshots.sort_unstable();
1768    purge_bank_snapshots(
1769        bank_snapshots
1770            .iter()
1771            .rev()
1772            .skip(num_bank_snapshots_to_retain),
1773    );
1774}
1775
1776/// At startup, purge old (i.e. unusable) bank snapshots
1777pub fn purge_old_bank_snapshots_at_startup(bank_snapshots_dir: impl AsRef<Path>) {
1778    purge_old_bank_snapshots(&bank_snapshots_dir, 1);
1779
1780    let highest_bank_snapshot = get_highest_bank_snapshot(&bank_snapshots_dir);
1781    if let Some(highest_bank_snapshot) = highest_bank_snapshot {
1782        debug!(
1783            "Retained bank snapshot for slot {}, and purged the rest.",
1784            highest_bank_snapshot.slot
1785        );
1786    }
1787}
1788
1789/// Purges bank snapshots that are older than `slot`
1790pub fn purge_bank_snapshots_older_than_slot(bank_snapshots_dir: impl AsRef<Path>, slot: Slot) {
1791    let mut bank_snapshots = get_bank_snapshots(&bank_snapshots_dir);
1792    bank_snapshots.retain(|bank_snapshot| bank_snapshot.slot < slot);
1793    purge_bank_snapshots(&bank_snapshots);
1794}
1795
1796/// Purges all `bank_snapshots`
1797///
1798/// Does not exit early if there is an error while purging a bank snapshot.
1799fn purge_bank_snapshots<'a>(bank_snapshots: impl IntoIterator<Item = &'a BankSnapshotInfo>) {
1800    for snapshot_dir in bank_snapshots.into_iter().map(|s| &s.snapshot_dir) {
1801        if purge_bank_snapshot(snapshot_dir).is_err() {
1802            warn!("Failed to purge bank snapshot: {}", snapshot_dir.display());
1803        }
1804    }
1805}
1806
1807/// Remove the bank snapshot at this path
1808pub fn purge_bank_snapshot(bank_snapshot_dir: impl AsRef<Path>) -> Result<()> {
1809    const FN_ERR: &str = "failed to purge bank snapshot";
1810    // Migration: snapshots written by pre-storages-list versions kept an `accounts_hardlinks/`
1811    // subdir of symlinks pointing at hardlink dirs under `<account_path>/snapshot/<slot>/`.
1812    // Follow them so the hardlink dirs don't outlive the owning bank snapshot when we purge at
1813    // runtime (startup-time cleanup catches any leftovers).
1814    let accounts_hardlinks_dir = bank_snapshot_dir
1815        .as_ref()
1816        .join(snapshot_paths::SNAPSHOT_ACCOUNTS_HARDLINKS);
1817    if accounts_hardlinks_dir.is_dir() {
1818        let read_dir = fs::read_dir(&accounts_hardlinks_dir).map_err(|err| {
1819            IoError::other(format!(
1820                "{FN_ERR}: failed to read accounts hardlinks dir '{}': {err}",
1821                accounts_hardlinks_dir.display(),
1822            ))
1823        })?;
1824        for entry in read_dir {
1825            let accounts_hardlink_dir = entry?.path();
1826            let accounts_hardlink_dir = fs::read_link(&accounts_hardlink_dir).map_err(|err| {
1827                IoError::other(format!(
1828                    "{FN_ERR}: failed to read symlink '{}': {err}",
1829                    accounts_hardlink_dir.display(),
1830                ))
1831            })?;
1832            move_and_async_delete_path(&accounts_hardlink_dir);
1833        }
1834    }
1835    fs::remove_dir_all(&bank_snapshot_dir).map_err(|err| {
1836        IoError::other(format!(
1837            "{FN_ERR}: failed to remove dir '{}': {err}",
1838            bank_snapshot_dir.as_ref().display(),
1839        ))
1840    })?;
1841    Ok(())
1842}
1843
1844pub fn should_take_full_snapshot(
1845    block_height: Slot,
1846    full_snapshot_archive_interval_slots: Slot,
1847) -> bool {
1848    block_height.is_multiple_of(full_snapshot_archive_interval_slots)
1849}
1850
1851pub fn should_take_incremental_snapshot(
1852    block_height: Slot,
1853    incremental_snapshot_archive_interval_slots: Slot,
1854    latest_full_snapshot_slot: Option<Slot>,
1855) -> bool {
1856    block_height.is_multiple_of(incremental_snapshot_archive_interval_slots)
1857        && latest_full_snapshot_slot.is_some()
1858}
1859
1860/// Creates an "accounts path" directory for tests
1861///
1862/// This temporary directory will contain the "run" and "snapshot"
1863/// sub-directories required by a validator.
1864#[cfg(feature = "dev-context-only-utils")]
1865pub fn create_tmp_accounts_dir_for_tests() -> (TempDir, PathBuf) {
1866    let tmp_dir = tempfile::TempDir::new().unwrap();
1867    let account_dir = create_accounts_run_and_snapshot_dirs(&tmp_dir).unwrap().0;
1868    (tmp_dir, account_dir)
1869}
1870
1871#[cfg(test)]
1872mod tests {
1873    use {
1874        super::*,
1875        crate::serde_snapshot::{deserialize_wincode_from, serialize_into},
1876        agave_snapshots::{
1877            paths::{
1878                full_snapshot_archives_iter, get_highest_full_snapshot_archive_slot,
1879                get_highest_incremental_snapshot_archive_slot,
1880            },
1881            snapshot_config::{
1882                DEFAULT_MAX_FULL_SNAPSHOT_ARCHIVES_TO_RETAIN,
1883                DEFAULT_MAX_INCREMENTAL_SNAPSHOT_ARCHIVES_TO_RETAIN,
1884            },
1885        },
1886        assert_matches::assert_matches,
1887        solana_accounts_db::accounts_file::{AccountsFile, AccountsFileProvider},
1888        solana_hash::Hash,
1889        std::{convert::TryFrom, mem::size_of},
1890        tempfile::NamedTempFile,
1891        test_case::test_case,
1892    };
1893
1894    #[test]
1895    fn test_serialize_snapshot_data_file_under_limit() {
1896        let temp_dir = tempfile::TempDir::new().unwrap();
1897        let expected_consumed_size = size_of::<u32>() as u64;
1898        let consumed_size = serialize_snapshot_data_file_capped(
1899            &temp_dir.path().join("data-file"),
1900            expected_consumed_size,
1901            &IoSetupState::default(),
1902            |stream| {
1903                serialize_into(stream, &2323_u32)?;
1904                Ok(())
1905            },
1906        )
1907        .unwrap();
1908        assert_eq!(consumed_size, expected_consumed_size);
1909    }
1910
1911    #[test]
1912    fn test_serialize_snapshot_data_file_over_limit() {
1913        let temp_dir = tempfile::TempDir::new().unwrap();
1914        let expected_consumed_size = size_of::<u32>() as u64;
1915        let result = serialize_snapshot_data_file_capped(
1916            &temp_dir.path().join("data-file"),
1917            expected_consumed_size - 1,
1918            &IoSetupState::default(),
1919            |stream| {
1920                serialize_into(stream, &2323_u32)?;
1921                Ok(())
1922            },
1923        );
1924        assert_matches!(result, Err(SnapshotError::Io(ref message)) if message.to_string().contains("bytes would exceed limit of"));
1925    }
1926
1927    #[test]
1928    fn test_deserialize_snapshot_data_file_under_limit() {
1929        let expected_data = 2323_u32;
1930        let expected_consumed_size = size_of::<u32>() as u64;
1931
1932        let temp_dir = tempfile::TempDir::new().unwrap();
1933        serialize_snapshot_data_file_capped(
1934            &temp_dir.path().join("data-file"),
1935            expected_consumed_size,
1936            &IoSetupState::default(),
1937            |stream| {
1938                serialize_into(stream, &expected_data)?;
1939                Ok(())
1940            },
1941        )
1942        .unwrap();
1943
1944        let snapshot_root_paths = SnapshotRootPaths {
1945            full_snapshot_root_file_path: temp_dir.path().join("data-file"),
1946            incremental_snapshot_root_file_path: None,
1947        };
1948
1949        let actual_data = deserialize_snapshot_data_files_capped(
1950            &snapshot_root_paths,
1951            expected_consumed_size,
1952            |stream| {
1953                Ok(deserialize_wincode_from::<_, u32>(
1954                    &mut *stream.full_snapshot_stream,
1955                )?)
1956            },
1957        )
1958        .unwrap();
1959        assert_eq!(actual_data, expected_data);
1960    }
1961
1962    #[test]
1963    fn test_deserialize_snapshot_data_file_over_limit() {
1964        let expected_data = 2323_u32;
1965        let expected_consumed_size = size_of::<u32>() as u64;
1966
1967        let temp_dir = tempfile::TempDir::new().unwrap();
1968        serialize_snapshot_data_file_capped(
1969            &temp_dir.path().join("data-file"),
1970            expected_consumed_size,
1971            &IoSetupState::default(),
1972            |stream| {
1973                serialize_into(stream, &expected_data)?;
1974                Ok(())
1975            },
1976        )
1977        .unwrap();
1978
1979        let snapshot_root_paths = SnapshotRootPaths {
1980            full_snapshot_root_file_path: temp_dir.path().join("data-file"),
1981            incremental_snapshot_root_file_path: None,
1982        };
1983
1984        let result = deserialize_snapshot_data_files_capped(
1985            &snapshot_root_paths,
1986            expected_consumed_size - 1,
1987            |stream| {
1988                Ok(deserialize_wincode_from::<_, u32>(
1989                    &mut *stream.full_snapshot_stream,
1990                )?)
1991            },
1992        );
1993        assert_matches!(result, Err(SnapshotError::Io(ref message)) if message.to_string().starts_with("too large snapshot data file to deserialize"));
1994    }
1995
1996    #[test]
1997    fn test_deserialize_snapshot_data_file_extra_data() {
1998        let expected_data = 2323_u32;
1999        let expected_consumed_size = size_of::<u32>() as u64;
2000
2001        let temp_dir = tempfile::TempDir::new().unwrap();
2002        serialize_snapshot_data_file_capped(
2003            &temp_dir.path().join("data-file"),
2004            expected_consumed_size * 2,
2005            &IoSetupState::default(),
2006            |stream| {
2007                // Write two u32s (in one call, since the wincode writer finalizes on finish) so
2008                // the file has trailing bytes left over after a single-u32 deserialize.
2009                serialize_into(&mut *stream, &(expected_data, expected_data))?;
2010                Ok(())
2011            },
2012        )
2013        .unwrap();
2014
2015        let snapshot_root_paths = SnapshotRootPaths {
2016            full_snapshot_root_file_path: temp_dir.path().join("data-file"),
2017            incremental_snapshot_root_file_path: None,
2018        };
2019
2020        let result = deserialize_snapshot_data_files_capped(
2021            &snapshot_root_paths,
2022            expected_consumed_size * 2,
2023            |stream| {
2024                Ok(deserialize_wincode_from::<_, u32>(
2025                    &mut *stream.full_snapshot_stream,
2026                )?)
2027            },
2028        );
2029        assert_matches!(result, Err(SnapshotError::Io(ref message)) if message.to_string().starts_with("invalid snapshot data file"));
2030    }
2031
2032    #[test]
2033    fn test_snapshot_version_from_file_under_limit() {
2034        let file_content = SnapshotVersion::default().as_str();
2035        let mut file = NamedTempFile::new().unwrap();
2036        file.write_all(file_content.as_bytes()).unwrap();
2037        let file_info = FileInfo::new_from_path(file.path()).unwrap();
2038        let version_from_file = snapshot_version_from_file(file_info).unwrap();
2039        assert_eq!(version_from_file, file_content);
2040    }
2041
2042    #[test]
2043    fn test_snapshot_version_from_file_over_limit() {
2044        let over_limit_size = usize::try_from(MAX_SNAPSHOT_VERSION_FILE_SIZE + 1).unwrap();
2045        let file_content = vec![7u8; over_limit_size];
2046        let mut file = NamedTempFile::new().unwrap();
2047        file.write_all(&file_content).unwrap();
2048        let file_info = FileInfo::new_from_path(file.path()).unwrap();
2049        assert_matches!(
2050            snapshot_version_from_file(file_info),
2051            Err(ref message) if message.to_string().starts_with("snapshot version file too large")
2052        );
2053    }
2054
2055    #[test]
2056    fn test_check_are_snapshots_compatible() {
2057        let slot1: Slot = 1234;
2058        let slot2: Slot = 5678;
2059        let slot3: Slot = 999_999;
2060
2061        let full_snapshot_archive_info = FullSnapshotArchiveInfo::new_from_path(PathBuf::from(
2062            format!("/dir/snapshot-{}-{}.tar.zst", slot1, Hash::new_unique()),
2063        ))
2064        .unwrap();
2065
2066        assert!(check_are_snapshots_compatible(&full_snapshot_archive_info, None,).is_ok());
2067
2068        let incremental_snapshot_archive_info =
2069            IncrementalSnapshotArchiveInfo::new_from_path(PathBuf::from(format!(
2070                "/dir/incremental-snapshot-{}-{}-{}.tar.zst",
2071                slot1,
2072                slot2,
2073                Hash::new_unique()
2074            )))
2075            .unwrap();
2076
2077        assert!(
2078            check_are_snapshots_compatible(
2079                &full_snapshot_archive_info,
2080                Some(&incremental_snapshot_archive_info)
2081            )
2082            .is_ok()
2083        );
2084
2085        let incremental_snapshot_archive_info =
2086            IncrementalSnapshotArchiveInfo::new_from_path(PathBuf::from(format!(
2087                "/dir/incremental-snapshot-{}-{}-{}.tar.zst",
2088                slot2,
2089                slot3,
2090                Hash::new_unique()
2091            )))
2092            .unwrap();
2093
2094        assert!(
2095            check_are_snapshots_compatible(
2096                &full_snapshot_archive_info,
2097                Some(&incremental_snapshot_archive_info)
2098            )
2099            .is_err()
2100        );
2101    }
2102
2103    /// A test heler function that creates bank snapshot files
2104    fn common_create_bank_snapshot_files(
2105        bank_snapshots_dir: &Path,
2106        min_slot: Slot,
2107        max_slot: Slot,
2108    ) {
2109        for slot in min_slot..max_slot {
2110            let snapshot_dir = snapshot_paths::get_bank_snapshot_dir(bank_snapshots_dir, slot);
2111            fs::create_dir_all(&snapshot_dir).unwrap();
2112
2113            let snapshot_filename = snapshot_paths::get_snapshot_file_name(slot);
2114            let snapshot_path = snapshot_dir.join(snapshot_filename);
2115            fs::File::create(snapshot_path).unwrap();
2116
2117            let status_cache_file =
2118                snapshot_dir.join(snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME);
2119            fs::File::create(status_cache_file).unwrap();
2120
2121            let version_path = snapshot_dir.join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
2122            fs::write(version_path, SnapshotVersion::default().as_str().as_bytes()).unwrap();
2123        }
2124    }
2125
2126    #[test]
2127    fn test_get_bank_snapshots() {
2128        let temp_snapshots_dir = tempfile::TempDir::new().unwrap();
2129        let min_slot = 10;
2130        let max_slot = 20;
2131        common_create_bank_snapshot_files(temp_snapshots_dir.path(), min_slot, max_slot);
2132
2133        let bank_snapshots = get_bank_snapshots(temp_snapshots_dir.path());
2134        assert_eq!(bank_snapshots.len() as Slot, max_slot - min_slot);
2135    }
2136
2137    #[test]
2138    fn test_get_highest_bank_snapshot() {
2139        let temp_snapshots_dir = tempfile::TempDir::new().unwrap();
2140        let min_slot = 99;
2141        let max_slot = 123;
2142        common_create_bank_snapshot_files(temp_snapshots_dir.path(), min_slot, max_slot);
2143
2144        let highest_bank_snapshot = get_highest_bank_snapshot(temp_snapshots_dir.path());
2145        assert!(highest_bank_snapshot.is_some());
2146        assert_eq!(highest_bank_snapshot.unwrap().slot, max_slot - 1);
2147    }
2148
2149    /// A test helper function that creates full and incremental snapshot archive files.  Creates
2150    /// full snapshot files in the range (`min_full_snapshot_slot`, `max_full_snapshot_slot`], and
2151    /// incremental snapshot files in the range (`min_incremental_snapshot_slot`,
2152    /// `max_incremental_snapshot_slot`].  Additionally, "bad" files are created for both full and
2153    /// incremental snapshots to ensure the tests properly filter them out.
2154    fn common_create_snapshot_archive_files(
2155        full_snapshot_archives_dir: &Path,
2156        incremental_snapshot_archives_dir: &Path,
2157        min_full_snapshot_slot: Slot,
2158        max_full_snapshot_slot: Slot,
2159        min_incremental_snapshot_slot: Slot,
2160        max_incremental_snapshot_slot: Slot,
2161    ) {
2162        fs::create_dir_all(full_snapshot_archives_dir).unwrap();
2163        fs::create_dir_all(incremental_snapshot_archives_dir).unwrap();
2164        for full_snapshot_slot in min_full_snapshot_slot..max_full_snapshot_slot {
2165            for incremental_snapshot_slot in
2166                min_incremental_snapshot_slot..max_incremental_snapshot_slot
2167            {
2168                let snapshot_filename = format!(
2169                    "incremental-snapshot-{}-{}-{}.tar.zst",
2170                    full_snapshot_slot,
2171                    incremental_snapshot_slot,
2172                    Hash::default()
2173                );
2174                let snapshot_filepath = incremental_snapshot_archives_dir.join(snapshot_filename);
2175                fs::File::create(snapshot_filepath).unwrap();
2176            }
2177
2178            let snapshot_filename = format!(
2179                "snapshot-{}-{}.tar.zst",
2180                full_snapshot_slot,
2181                Hash::default()
2182            );
2183            let snapshot_filepath = full_snapshot_archives_dir.join(snapshot_filename);
2184            fs::File::create(snapshot_filepath).unwrap();
2185
2186            // Add in an incremental snapshot with a bad filename and high slot to ensure filename are filtered and sorted correctly
2187            let bad_filename = format!(
2188                "incremental-snapshot-{}-{}-bad!hash.tar.zst",
2189                full_snapshot_slot,
2190                max_incremental_snapshot_slot + 1,
2191            );
2192            let bad_filepath = incremental_snapshot_archives_dir.join(bad_filename);
2193            fs::File::create(bad_filepath).unwrap();
2194        }
2195
2196        // Add in a snapshot with a bad filename and high slot to ensure filename are filtered and
2197        // sorted correctly
2198        let bad_filename = format!("snapshot-{}-bad!hash.tar.zst", max_full_snapshot_slot + 1);
2199        let bad_filepath = full_snapshot_archives_dir.join(bad_filename);
2200        fs::File::create(bad_filepath).unwrap();
2201    }
2202
2203    #[test]
2204    fn test_get_full_snapshot_archives() {
2205        let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2206        let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2207        let min_slot = 123;
2208        let max_slot = 456;
2209        common_create_snapshot_archive_files(
2210            full_snapshot_archives_dir.path(),
2211            incremental_snapshot_archives_dir.path(),
2212            min_slot,
2213            max_slot,
2214            0,
2215            0,
2216        );
2217
2218        let snapshot_archives =
2219            full_snapshot_archives_iter(full_snapshot_archives_dir.path()).collect::<Vec<_>>();
2220        assert_eq!(snapshot_archives.len() as Slot, max_slot - min_slot);
2221    }
2222
2223    #[test]
2224    fn test_get_full_snapshot_archives_remote() {
2225        let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2226        let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2227        let min_slot = 123;
2228        let max_slot = 456;
2229        common_create_snapshot_archive_files(
2230            &full_snapshot_archives_dir
2231                .path()
2232                .join(snapshot_paths::SNAPSHOT_ARCHIVE_DOWNLOAD_DIR),
2233            &incremental_snapshot_archives_dir
2234                .path()
2235                .join(snapshot_paths::SNAPSHOT_ARCHIVE_DOWNLOAD_DIR),
2236            min_slot,
2237            max_slot,
2238            0,
2239            0,
2240        );
2241
2242        let snapshot_archives =
2243            full_snapshot_archives_iter(full_snapshot_archives_dir.path()).collect::<Vec<_>>();
2244        assert_eq!(snapshot_archives.len() as Slot, max_slot - min_slot);
2245        assert!(snapshot_archives.iter().all(|info| info.is_remote()));
2246    }
2247
2248    #[test]
2249    fn test_get_incremental_snapshot_archives() {
2250        let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2251        let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2252        let min_full_snapshot_slot = 12;
2253        let max_full_snapshot_slot = 23;
2254        let min_incremental_snapshot_slot = 34;
2255        let max_incremental_snapshot_slot = 45;
2256        common_create_snapshot_archive_files(
2257            full_snapshot_archives_dir.path(),
2258            incremental_snapshot_archives_dir.path(),
2259            min_full_snapshot_slot,
2260            max_full_snapshot_slot,
2261            min_incremental_snapshot_slot,
2262            max_incremental_snapshot_slot,
2263        );
2264
2265        let incremental_snapshot_archives =
2266            incremental_snapshot_archives_iter(incremental_snapshot_archives_dir.path())
2267                .collect::<Vec<_>>();
2268        assert_eq!(
2269            incremental_snapshot_archives.len() as Slot,
2270            (max_full_snapshot_slot - min_full_snapshot_slot)
2271                * (max_incremental_snapshot_slot - min_incremental_snapshot_slot)
2272        );
2273    }
2274
2275    #[test]
2276    fn test_get_incremental_snapshot_archives_remote() {
2277        let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2278        let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2279        let min_full_snapshot_slot = 12;
2280        let max_full_snapshot_slot = 23;
2281        let min_incremental_snapshot_slot = 34;
2282        let max_incremental_snapshot_slot = 45;
2283        common_create_snapshot_archive_files(
2284            &full_snapshot_archives_dir
2285                .path()
2286                .join(snapshot_paths::SNAPSHOT_ARCHIVE_DOWNLOAD_DIR),
2287            &incremental_snapshot_archives_dir
2288                .path()
2289                .join(snapshot_paths::SNAPSHOT_ARCHIVE_DOWNLOAD_DIR),
2290            min_full_snapshot_slot,
2291            max_full_snapshot_slot,
2292            min_incremental_snapshot_slot,
2293            max_incremental_snapshot_slot,
2294        );
2295
2296        let incremental_snapshot_archives =
2297            incremental_snapshot_archives_iter(incremental_snapshot_archives_dir.path())
2298                .collect::<Vec<_>>();
2299        assert_eq!(
2300            incremental_snapshot_archives.len() as Slot,
2301            (max_full_snapshot_slot - min_full_snapshot_slot)
2302                * (max_incremental_snapshot_slot - min_incremental_snapshot_slot)
2303        );
2304        assert!(
2305            incremental_snapshot_archives
2306                .iter()
2307                .all(|info| info.is_remote())
2308        );
2309    }
2310
2311    #[test]
2312    fn test_get_highest_full_snapshot_archive_slot() {
2313        let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2314        let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2315        let min_slot = 123;
2316        let max_slot = 456;
2317        common_create_snapshot_archive_files(
2318            full_snapshot_archives_dir.path(),
2319            incremental_snapshot_archives_dir.path(),
2320            min_slot,
2321            max_slot,
2322            0,
2323            0,
2324        );
2325
2326        assert_eq!(
2327            get_highest_full_snapshot_archive_slot(full_snapshot_archives_dir.path()),
2328            Some(max_slot - 1)
2329        );
2330    }
2331
2332    #[test]
2333    fn test_get_highest_incremental_snapshot_slot() {
2334        let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2335        let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2336        let min_full_snapshot_slot = 12;
2337        let max_full_snapshot_slot = 23;
2338        let min_incremental_snapshot_slot = 34;
2339        let max_incremental_snapshot_slot = 45;
2340        common_create_snapshot_archive_files(
2341            full_snapshot_archives_dir.path(),
2342            incremental_snapshot_archives_dir.path(),
2343            min_full_snapshot_slot,
2344            max_full_snapshot_slot,
2345            min_incremental_snapshot_slot,
2346            max_incremental_snapshot_slot,
2347        );
2348
2349        for full_snapshot_slot in min_full_snapshot_slot..max_full_snapshot_slot {
2350            assert_eq!(
2351                get_highest_incremental_snapshot_archive_slot(
2352                    incremental_snapshot_archives_dir.path(),
2353                    full_snapshot_slot
2354                ),
2355                Some(max_incremental_snapshot_slot - 1)
2356            );
2357        }
2358
2359        assert_eq!(
2360            get_highest_incremental_snapshot_archive_slot(
2361                incremental_snapshot_archives_dir.path(),
2362                max_full_snapshot_slot
2363            ),
2364            None
2365        );
2366    }
2367
2368    fn common_test_purge_old_snapshot_archives(
2369        snapshot_names: &[&String],
2370        maximum_full_snapshot_archives_to_retain: NonZeroUsize,
2371        maximum_incremental_snapshot_archives_to_retain: NonZeroUsize,
2372        expected_snapshots: &[&String],
2373    ) {
2374        let temp_snap_dir = tempfile::TempDir::new().unwrap();
2375
2376        for snap_name in snapshot_names {
2377            let snap_path = temp_snap_dir.path().join(snap_name);
2378            let mut _snap_file = fs::File::create(snap_path);
2379        }
2380        purge_old_snapshot_archives(
2381            temp_snap_dir.path(),
2382            temp_snap_dir.path(),
2383            maximum_full_snapshot_archives_to_retain,
2384            maximum_incremental_snapshot_archives_to_retain,
2385        );
2386
2387        let mut retained_snaps = HashSet::new();
2388        for entry in fs::read_dir(temp_snap_dir.path()).unwrap() {
2389            let entry_path_buf = entry.unwrap().path();
2390            let entry_path = entry_path_buf.as_path();
2391            let snapshot_name = entry_path
2392                .file_name()
2393                .unwrap()
2394                .to_str()
2395                .unwrap()
2396                .to_string();
2397            retained_snaps.insert(snapshot_name);
2398        }
2399
2400        for snap_name in expected_snapshots {
2401            assert!(
2402                retained_snaps.contains(snap_name.as_str()),
2403                "{snap_name} not found"
2404            );
2405        }
2406        assert_eq!(retained_snaps.len(), expected_snapshots.len());
2407    }
2408
2409    #[test]
2410    fn test_purge_old_full_snapshot_archives() {
2411        let snap1_name = format!("snapshot-1-{}.tar.zst", Hash::default());
2412        let snap2_name = format!("snapshot-3-{}.tar.zst", Hash::default());
2413        let snap3_name = format!("snapshot-50-{}.tar.zst", Hash::default());
2414        let snapshot_names = vec![&snap1_name, &snap2_name, &snap3_name];
2415
2416        // expecting only the newest to be retained
2417        let expected_snapshots = vec![&snap3_name];
2418        common_test_purge_old_snapshot_archives(
2419            &snapshot_names,
2420            NonZeroUsize::new(1).unwrap(),
2421            DEFAULT_MAX_INCREMENTAL_SNAPSHOT_ARCHIVES_TO_RETAIN,
2422            &expected_snapshots,
2423        );
2424
2425        // retaining 2, expecting the 2 newest to be retained
2426        let expected_snapshots = vec![&snap2_name, &snap3_name];
2427        common_test_purge_old_snapshot_archives(
2428            &snapshot_names,
2429            NonZeroUsize::new(2).unwrap(),
2430            DEFAULT_MAX_INCREMENTAL_SNAPSHOT_ARCHIVES_TO_RETAIN,
2431            &expected_snapshots,
2432        );
2433
2434        // retaining 3, all three should be retained
2435        let expected_snapshots = vec![&snap1_name, &snap2_name, &snap3_name];
2436        common_test_purge_old_snapshot_archives(
2437            &snapshot_names,
2438            NonZeroUsize::new(3).unwrap(),
2439            DEFAULT_MAX_INCREMENTAL_SNAPSHOT_ARCHIVES_TO_RETAIN,
2440            &expected_snapshots,
2441        );
2442    }
2443
2444    /// Mimic a running node's behavior w.r.t. purging old snapshot archives.  Take snapshots in a
2445    /// loop, and periodically purge old snapshot archives.  After purging, check to make sure the
2446    /// snapshot archives on disk are correct.
2447    #[test]
2448    fn test_purge_old_full_snapshot_archives_in_the_loop() {
2449        let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2450        let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2451        let maximum_snapshots_to_retain = NonZeroUsize::new(5).unwrap();
2452        let starting_slot: Slot = 42;
2453
2454        for slot in (starting_slot..).take(100) {
2455            let full_snapshot_archive_file_name =
2456                format!("snapshot-{}-{}.tar.zst", slot, Hash::default());
2457            let full_snapshot_archive_path = full_snapshot_archives_dir
2458                .as_ref()
2459                .join(full_snapshot_archive_file_name);
2460            fs::File::create(full_snapshot_archive_path).unwrap();
2461
2462            // don't purge-and-check until enough snapshot archives have been created
2463            if slot < starting_slot + maximum_snapshots_to_retain.get() as Slot {
2464                continue;
2465            }
2466
2467            // purge infrequently, so there will always be snapshot archives to purge
2468            if slot % (maximum_snapshots_to_retain.get() as Slot * 2) != 0 {
2469                continue;
2470            }
2471
2472            purge_old_snapshot_archives(
2473                &full_snapshot_archives_dir,
2474                &incremental_snapshot_archives_dir,
2475                maximum_snapshots_to_retain,
2476                NonZeroUsize::new(usize::MAX).unwrap(),
2477            );
2478            let mut full_snapshot_archives =
2479                full_snapshot_archives_iter(full_snapshot_archives_dir.path()).collect::<Vec<_>>();
2480            full_snapshot_archives.sort_unstable();
2481            assert_eq!(
2482                full_snapshot_archives.len(),
2483                maximum_snapshots_to_retain.get()
2484            );
2485            assert_eq!(full_snapshot_archives.last().unwrap().slot(), slot);
2486            for (i, full_snapshot_archive) in full_snapshot_archives.iter().rev().enumerate() {
2487                assert_eq!(full_snapshot_archive.slot(), slot - i as Slot);
2488            }
2489        }
2490    }
2491
2492    #[test]
2493    fn test_purge_old_incremental_snapshot_archives() {
2494        let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2495        let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2496        let starting_slot = 100_000;
2497
2498        let maximum_incremental_snapshot_archives_to_retain =
2499            DEFAULT_MAX_INCREMENTAL_SNAPSHOT_ARCHIVES_TO_RETAIN;
2500        let maximum_full_snapshot_archives_to_retain = DEFAULT_MAX_FULL_SNAPSHOT_ARCHIVES_TO_RETAIN;
2501
2502        let incremental_snapshot_interval = 100;
2503        let num_incremental_snapshots_per_full_snapshot =
2504            maximum_incremental_snapshot_archives_to_retain.get() * 2;
2505        let full_snapshot_interval =
2506            incremental_snapshot_interval * num_incremental_snapshots_per_full_snapshot;
2507
2508        let mut snapshot_filenames = vec![];
2509        (starting_slot..)
2510            .step_by(full_snapshot_interval)
2511            .take(
2512                maximum_full_snapshot_archives_to_retain
2513                    .checked_mul(NonZeroUsize::new(2).unwrap())
2514                    .unwrap()
2515                    .get(),
2516            )
2517            .for_each(|full_snapshot_slot| {
2518                let snapshot_filename = format!(
2519                    "snapshot-{}-{}.tar.zst",
2520                    full_snapshot_slot,
2521                    Hash::default()
2522                );
2523                let snapshot_path = full_snapshot_archives_dir.path().join(&snapshot_filename);
2524                fs::File::create(snapshot_path).unwrap();
2525                snapshot_filenames.push(snapshot_filename);
2526
2527                (full_snapshot_slot..)
2528                    .step_by(incremental_snapshot_interval)
2529                    .take(num_incremental_snapshots_per_full_snapshot)
2530                    .skip(1)
2531                    .for_each(|incremental_snapshot_slot| {
2532                        let snapshot_filename = format!(
2533                            "incremental-snapshot-{}-{}-{}.tar.zst",
2534                            full_snapshot_slot,
2535                            incremental_snapshot_slot,
2536                            Hash::default()
2537                        );
2538                        let snapshot_path = incremental_snapshot_archives_dir
2539                            .path()
2540                            .join(&snapshot_filename);
2541                        fs::File::create(snapshot_path).unwrap();
2542                        snapshot_filenames.push(snapshot_filename);
2543                    });
2544            });
2545
2546        purge_old_snapshot_archives(
2547            full_snapshot_archives_dir.path(),
2548            incremental_snapshot_archives_dir.path(),
2549            maximum_full_snapshot_archives_to_retain,
2550            maximum_incremental_snapshot_archives_to_retain,
2551        );
2552
2553        // Ensure correct number of full snapshot archives are purged/retained
2554        let mut remaining_full_snapshot_archives =
2555            full_snapshot_archives_iter(full_snapshot_archives_dir.path()).collect::<Vec<_>>();
2556        assert_eq!(
2557            remaining_full_snapshot_archives.len(),
2558            maximum_full_snapshot_archives_to_retain.get(),
2559        );
2560        remaining_full_snapshot_archives.sort_unstable();
2561        let latest_full_snapshot_archive_slot =
2562            remaining_full_snapshot_archives.last().unwrap().slot();
2563
2564        // Ensure correct number of incremental snapshot archives are purged/retained
2565        // For each additional full snapshot archive, one additional (the newest)
2566        // incremental snapshot archive is retained. This is accounted for by the
2567        // `+ maximum_full_snapshot_archives_to_retain.saturating_sub(1)`
2568        let mut remaining_incremental_snapshot_archives =
2569            incremental_snapshot_archives_iter(incremental_snapshot_archives_dir.path())
2570                .collect::<Vec<_>>();
2571        assert_eq!(
2572            remaining_incremental_snapshot_archives.len(),
2573            maximum_incremental_snapshot_archives_to_retain
2574                .get()
2575                .saturating_add(
2576                    maximum_full_snapshot_archives_to_retain
2577                        .get()
2578                        .saturating_sub(1)
2579                )
2580        );
2581        remaining_incremental_snapshot_archives.sort_unstable();
2582        remaining_incremental_snapshot_archives.reverse();
2583
2584        // Ensure there exists one incremental snapshot all but the latest full snapshot
2585        for i in (1..maximum_full_snapshot_archives_to_retain.get()).rev() {
2586            let incremental_snapshot_archive =
2587                remaining_incremental_snapshot_archives.pop().unwrap();
2588
2589            let expected_base_slot =
2590                latest_full_snapshot_archive_slot - (i * full_snapshot_interval) as u64;
2591            assert_eq!(incremental_snapshot_archive.base_slot(), expected_base_slot);
2592            let expected_slot = expected_base_slot
2593                + (full_snapshot_interval - incremental_snapshot_interval) as u64;
2594            assert_eq!(incremental_snapshot_archive.slot(), expected_slot);
2595        }
2596
2597        // Ensure all remaining incremental snapshots are only for the latest full snapshot
2598        for incremental_snapshot_archive in &remaining_incremental_snapshot_archives {
2599            assert_eq!(
2600                incremental_snapshot_archive.base_slot(),
2601                latest_full_snapshot_archive_slot
2602            );
2603        }
2604
2605        // Ensure the remaining incremental snapshots are at the right slot
2606        let expected_remaining_incremental_snapshot_archive_slots =
2607            (latest_full_snapshot_archive_slot..)
2608                .step_by(incremental_snapshot_interval)
2609                .take(num_incremental_snapshots_per_full_snapshot)
2610                .skip(
2611                    num_incremental_snapshots_per_full_snapshot
2612                        - maximum_incremental_snapshot_archives_to_retain.get(),
2613                )
2614                .collect::<HashSet<_>>();
2615
2616        let actual_remaining_incremental_snapshot_archive_slots =
2617            remaining_incremental_snapshot_archives
2618                .iter()
2619                .map(|snapshot| snapshot.slot())
2620                .collect::<HashSet<_>>();
2621        assert_eq!(
2622            actual_remaining_incremental_snapshot_archive_slots,
2623            expected_remaining_incremental_snapshot_archive_slots
2624        );
2625    }
2626
2627    #[test]
2628    fn test_purge_all_incremental_snapshot_archives_when_no_full_snapshot_archives() {
2629        let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2630        let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2631
2632        for snapshot_filenames in [
2633            format!("incremental-snapshot-100-120-{}.tar.zst", Hash::default()),
2634            format!("incremental-snapshot-100-140-{}.tar.zst", Hash::default()),
2635            format!("incremental-snapshot-100-160-{}.tar.zst", Hash::default()),
2636            format!("incremental-snapshot-100-180-{}.tar.zst", Hash::default()),
2637            format!("incremental-snapshot-200-220-{}.tar.zst", Hash::default()),
2638            format!("incremental-snapshot-200-240-{}.tar.zst", Hash::default()),
2639            format!("incremental-snapshot-200-260-{}.tar.zst", Hash::default()),
2640            format!("incremental-snapshot-200-280-{}.tar.zst", Hash::default()),
2641        ] {
2642            let snapshot_path = incremental_snapshot_archives_dir
2643                .path()
2644                .join(snapshot_filenames);
2645            fs::File::create(snapshot_path).unwrap();
2646        }
2647
2648        purge_old_snapshot_archives(
2649            full_snapshot_archives_dir.path(),
2650            incremental_snapshot_archives_dir.path(),
2651            NonZeroUsize::new(usize::MAX).unwrap(),
2652            NonZeroUsize::new(usize::MAX).unwrap(),
2653        );
2654
2655        let remaining_incremental_snapshot_archives =
2656            incremental_snapshot_archives_iter(incremental_snapshot_archives_dir.path())
2657                .collect::<Vec<_>>();
2658        assert!(remaining_incremental_snapshot_archives.is_empty());
2659    }
2660
2661    #[test]
2662    fn test_get_snapshot_file_kind() {
2663        assert_eq!(None, get_snapshot_file_kind("file.txt"));
2664        assert_eq!(
2665            Some(SnapshotFileKind::Version),
2666            get_snapshot_file_kind(snapshot_paths::SNAPSHOT_VERSION_FILENAME)
2667        );
2668        assert_eq!(
2669            Some(SnapshotFileKind::BankFields),
2670            get_snapshot_file_kind("1234")
2671        );
2672        assert_eq!(
2673            Some(SnapshotFileKind::Storage),
2674            get_snapshot_file_kind("1000.999")
2675        );
2676    }
2677
2678    #[test_case(0)]
2679    #[test_case(1)]
2680    #[test_case(10)]
2681    fn test_serialize_deserialize_account_storage_entries(num_storages: u64) {
2682        let temp_dir = tempfile::tempdir().unwrap();
2683        let bank_snapshot_dir = temp_dir.path();
2684        let storage_dir = tempfile::tempdir().unwrap();
2685        let snapshot_slot = num_storages + 1 as Slot;
2686
2687        // Create AccountStorageEntries
2688        let mut snapshot_storages = Vec::new();
2689        for i in 0..num_storages {
2690            let storage = Arc::new(AccountStorageEntry::new(
2691                storage_dir.path(),
2692                i,        // Incrementing slot
2693                i as u32, // Incrementing id
2694                1024,
2695                AccountsFileProvider::AppendVec,
2696            ));
2697            snapshot_storages.push(storage);
2698        }
2699
2700        // write obsolete accounts to snapshot
2701        write_obsolete_accounts_to_snapshot(
2702            bank_snapshot_dir,
2703            &snapshot_storages,
2704            snapshot_slot,
2705            &IoSetupState::default(),
2706        )
2707        .unwrap();
2708
2709        // Deserialize
2710        let mut deserialized_accounts =
2711            deserialize_obsolete_accounts(bank_snapshot_dir, MAX_OBSOLETE_ACCOUNTS_FILE_SIZE)
2712                .unwrap()
2713                .into_hashmap();
2714
2715        // Verify
2716        for storage in &snapshot_storages {
2717            let obsolete_accounts = deserialized_accounts.remove(&storage.slot()).unwrap();
2718            assert!(obsolete_accounts.into_tuple().2 == 0);
2719        }
2720    }
2721
2722    #[test]
2723    #[should_panic(expected = "bytes would exceed limit of 100")]
2724    fn test_serialize_obsolete_accounts_too_large_file() {
2725        let temp_dir = tempfile::tempdir().unwrap();
2726        let bank_snapshot_dir = temp_dir.path();
2727        let storage_dir = tempfile::tempdir().unwrap();
2728        let num_storages = 10;
2729        let snapshot_slot = num_storages + 1 as Slot;
2730
2731        // Create AccountStorageEntries
2732        let mut snapshot_storages = Vec::new();
2733        for i in 0..num_storages {
2734            let storage = Arc::new(AccountStorageEntry::new(
2735                storage_dir.path(),
2736                i,        // Incrementing slot
2737                i as u32, // Incrementing id
2738                1024,
2739                AccountsFileProvider::AppendVec,
2740            ));
2741            snapshot_storages.push(storage);
2742        }
2743
2744        // write obsolete accounts to snapshot
2745        let obsolete_accounts =
2746            SerdeObsoleteAccountsMap::new_from_storages(&snapshot_storages, snapshot_slot);
2747
2748        // Limit the file size to something low for the test
2749        serialize_obsolete_accounts(
2750            bank_snapshot_dir,
2751            &obsolete_accounts,
2752            100,
2753            &IoSetupState::default(),
2754        )
2755        .unwrap();
2756    }
2757
2758    #[test]
2759    #[should_panic(expected = "too large obsolete accounts file to deserialize")]
2760    fn test_deserialize_obsolete_accounts_too_large_file() {
2761        let temp_dir = tempfile::tempdir().unwrap();
2762        let bank_snapshot_dir = temp_dir.path();
2763        let storage_dir = tempfile::tempdir().unwrap();
2764        let num_storages = 10;
2765        let snapshot_slot = num_storages + 1 as Slot;
2766
2767        // Create AccountStorageEntries
2768        let mut snapshot_storages = Vec::new();
2769        for i in 0..num_storages {
2770            let storage = Arc::new(AccountStorageEntry::new(
2771                storage_dir.path(),
2772                i,        // Incrementing slot
2773                i as u32, // Incrementing id
2774                1024,
2775                AccountsFileProvider::AppendVec,
2776            ));
2777            snapshot_storages.push(storage);
2778        }
2779
2780        // Write obsolete accounts to snapshot
2781        write_obsolete_accounts_to_snapshot(
2782            bank_snapshot_dir,
2783            &snapshot_storages,
2784            snapshot_slot,
2785            &IoSetupState::default(),
2786        )
2787        .unwrap();
2788
2789        // Set a very low maximum file size for deserialization
2790        // This should panic
2791        deserialize_obsolete_accounts(bank_snapshot_dir, 100).unwrap();
2792    }
2793
2794    #[test]
2795    fn test_is_bank_snapshot_complete() {
2796        let temp_dir = TempDir::new().unwrap();
2797        let slot = 123;
2798        let bank_snapshot_dir = temp_dir.as_ref().join(slot.to_string());
2799        fs::create_dir(&bank_snapshot_dir).unwrap();
2800
2801        let version_path = bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
2802        let serialized_bank_path = bank_snapshot_dir.join(slot.to_string());
2803        let status_cache_path =
2804            bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME);
2805
2806        // scenario 1: no version file
2807        assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2808
2809        // scenario 2: bad version file (too large)
2810        let too_large = format!(
2811            "{:v>width$}",
2812            "hi",
2813            width = (MAX_SNAPSHOT_VERSION_FILE_SIZE + 1) as usize,
2814        );
2815        fs::write(&version_path, too_large).unwrap();
2816        assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2817
2818        // scenario 3: bad version
2819        fs::remove_file(&version_path).unwrap();
2820        let bad_version = String::from("v0.0.0");
2821        fs::write(&version_path, bad_version).unwrap();
2822        assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2823
2824        // scenario 4: empty version
2825        fs::remove_file(&version_path).unwrap();
2826        fs::File::create_new(&version_path).unwrap();
2827        assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2828
2829        // write a "good" version file so we can check the next file
2830        fs::remove_file(&version_path).unwrap();
2831        fs::write(&version_path, SnapshotVersion::default().as_str()).unwrap();
2832
2833        // scenario 5: no serialized bank file
2834        assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2835
2836        // scenario 6: empty serialized bank
2837        fs::File::create_new(&serialized_bank_path).unwrap();
2838        assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2839
2840        // write a "good" serialized bank file so we can check the next file
2841        fs::remove_file(&serialized_bank_path).unwrap();
2842        fs::write(&serialized_bank_path, "serialized bank").unwrap();
2843
2844        // scenario 7: no status cache file
2845        assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2846
2847        // scenario 8: empty status cache
2848        fs::File::create_new(&status_cache_path).unwrap();
2849        assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2850
2851        // write a "good" status cache file so we can check for all good
2852        fs::remove_file(&status_cache_path).unwrap();
2853        fs::write(&status_cache_path, "status cache").unwrap();
2854
2855        // scenario 9: all good
2856        assert!(is_bank_snapshot_complete(bank_snapshot_dir));
2857    }
2858
2859    #[test]
2860    fn test_prune_stale_storages() {
2861        let account_path = tempfile::TempDir::new().unwrap();
2862        // Files that belong to the snapshot.
2863        let keep_a = account_path.path().join(AccountsFile::file_name(100, 1));
2864        let keep_b = account_path.path().join(AccountsFile::file_name(200, 2));
2865        // A stale storage file that should be removed.
2866        let stale = account_path.path().join(AccountsFile::file_name(300, 3));
2867        // A non-storage filename — should be left alone.
2868        let untouched = account_path.path().join("something_else.txt");
2869        for path in [&keep_a, &keep_b, &stale, &untouched] {
2870            fs::write(path, b"x").unwrap();
2871        }
2872
2873        let storages_list = StoragesList::from_items(vec![
2874            StorageListItem { slot: 100, id: 1 },
2875            StorageListItem { slot: 200, id: 2 },
2876        ]);
2877        prune_stale_storages(
2878            std::slice::from_ref(&account_path.path().to_path_buf()),
2879            storages_list,
2880        )
2881        .unwrap();
2882
2883        assert!(keep_a.exists(), "expected storage file was deleted");
2884        assert!(keep_b.exists(), "expected storage file was deleted");
2885        assert!(!stale.exists(), "stale storage file was not removed");
2886        assert!(untouched.exists(), "non-storage file was wrongly removed");
2887    }
2888}