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