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
69pub const MAX_OBSOLETE_ACCOUNTS_FILE_SIZE: u64 = 1024 * 1024 * 1024 * 12; pub const MAX_STORAGES_LIST_FILE_SIZE: u64 = 100 * 1024 * 1024; pub const MAX_SNAPSHOT_DATA_FILE_SIZE: u64 = 32 * 1024 * 1024 * 1024; const MAX_SNAPSHOT_VERSION_FILE_SIZE: u64 = 8; const AUX_SNAPSHOT_FILE_READ_BUF_SIZE: usize = 4 * 1024 * 1024;
83
84const SNAPSHOT_FASTBOOT_VERSION: Version = Version::new(3, 0, 0);
97
98#[derive(PartialEq, Eq, Debug)]
101pub struct BankSnapshotInfo {
102 pub slot: Slot,
104 pub snapshot_dir: PathBuf,
106 pub snapshot_version: SnapshotVersion,
108 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
118impl 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 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 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 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#[derive(Clone, Copy, Debug, Eq, PartialEq)]
204pub enum SnapshotFrom {
205 Archive,
207 Dir,
209}
210
211#[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#[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#[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#[expect(dead_code)]
247#[derive(Debug)]
248pub struct UnarchivedSnapshotsGuard {
249 full_unpack_dir: TempDir,
250 incremental_unpack_dir: Option<TempDir>,
251}
252#[derive(Debug)]
254pub struct UnpackedSnapshotsDirAndVersion {
255 pub unpacked_snapshots_dir: PathBuf,
256 pub snapshot_version: SnapshotVersion,
257}
258
259pub(crate) struct StorageAndNextAccountsFileId {
262 pub storage: AccountStorageMap,
263 pub next_append_vec_id: AtomicAccountsFileId,
264}
265
266pub 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 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 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
295fn 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 return false;
304 };
305
306 let Ok(version_str) = snapshot_version_from_file(version_file_info) else {
307 return false;
309 };
310
311 let Ok(_snapshot_version) = SnapshotVersion::from_str(version_str.as_str()) else {
312 return false;
314 };
315
316 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 return false;
328 };
329 if file_info.size == 0 {
330 return false;
332 }
333 }
334
335 true
336}
337
338pub 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
356fn 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 Ok(false)
365 }
366}
367
368fn is_snapshot_fastboot_compatible(
370 version: &Version,
371) -> std::result::Result<bool, SnapshotFastbootError> {
372 match version.major {
373 3 => Ok(true),
375 2 => Ok(true),
378 v if v > SNAPSHOT_FASTBOOT_VERSION.major => {
379 Err(SnapshotFastbootError::IncompatibleVersion(version.clone()))
380 }
381 _ => Ok(false),
383 }
384}
385
386pub 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
410pub 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
438pub 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 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
483pub 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 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 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 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 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
644pub 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 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 Err(err) => debug!("Unable to read bank snapshot for slot {slot}: {err}"),
675 },
676 ),
677 }
678 bank_snapshots
679}
680
681pub 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 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 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
939fn 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
963fn 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
986pub 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#[derive(PartialEq, Debug)]
1081enum SnapshotFileKind {
1082 Version,
1083 BankFields,
1084 Storage,
1085}
1086
1087fn 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
1105fn 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 if snapshot_bank.is_some() {
1124 break;
1125 }
1126 }
1127 Some(SnapshotFileKind::BankFields) => {
1128 snapshot_bank = Some(file_info);
1129
1130 if snapshot_version.is_some() {
1132 break;
1133 }
1134 }
1135 Some(SnapshotFileKind::Storage) => {
1136 append_vec_files.push(file_info);
1137 }
1138 None => {} }
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
1152struct SnapshotFieldsBundle {
1154 snapshot_version: SnapshotVersion,
1155 bank_fields: BankFieldsToDeserialize,
1156 accounts_db_fields: AccountsDbFields,
1157 append_vec_files: Vec<FileInfo>,
1158}
1159
1160fn 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
1185fn 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 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#[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 let unarchive_result = unarchive_handle.join().expect("must join unarchive thread");
1286 match (unarchive_result, snapshot_result) {
1287 (Err(SnapshotError::CrossbeamSend(_)), snap @ Err(_)) => snap,
1290 (Err(err), _) => Err(err),
1291 (Ok(()), snap) => snap,
1292 }
1293 })
1294}
1295
1296fn 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
1330fn 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 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 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 serialize_storages_list_to_snapshot(
1415 bank_snapshot_dir,
1416 StoragesList::from_items(items),
1417 &IoSetupState::default(),
1418 )?;
1419
1420 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 mark_bank_snapshot_as_loadable(bank_snapshot_dir)?;
1434
1435 Ok(())
1436}
1437
1438fn 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 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
1471pub(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 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 let storages_list_path =
1502 bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_STORAGES_LIST_FILENAME);
1503 if !storages_list_path.exists() {
1504 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 let stream_files_result = stream_files_handle.join().expect("must join dir thread");
1542 match (stream_files_result, snapshot_result) {
1543 (Err(SnapshotError::CrossbeamSend(_)), snap @ Err(_)) => snap,
1546 (Err(err), _) => Err(err),
1547 (Ok(()), snap) => snap,
1548 }
1549}
1550
1551fn 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 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
1581fn 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)]
1729pub enum VerifyBank {
1731 Deterministic,
1733 NonDeterministic,
1736}
1737
1738pub 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
1755pub 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
1761pub 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
1777pub 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
1790pub 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
1797fn 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
1808pub fn purge_bank_snapshot(bank_snapshot_dir: impl AsRef<Path>) -> Result<()> {
1810 const FN_ERR: &str = "failed to purge bank snapshot";
1811 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#[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 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 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 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 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 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 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 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 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 #[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 if slot < starting_slot + maximum_snapshots_to_retain.get() as Slot {
2465 continue;
2466 }
2467
2468 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 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 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 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 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 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 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, i as u32, 1024,
2696 AccountsFileProvider::AppendVec,
2697 ));
2698 snapshot_storages.push(storage);
2699 }
2700
2701 write_obsolete_accounts_to_snapshot(
2703 bank_snapshot_dir,
2704 &snapshot_storages,
2705 snapshot_slot,
2706 &IoSetupState::default(),
2707 )
2708 .unwrap();
2709
2710 let mut deserialized_accounts =
2712 deserialize_obsolete_accounts(bank_snapshot_dir, MAX_OBSOLETE_ACCOUNTS_FILE_SIZE)
2713 .unwrap()
2714 .into_hashmap();
2715
2716 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 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, i as u32, 1024,
2740 AccountsFileProvider::AppendVec,
2741 ));
2742 snapshot_storages.push(storage);
2743 }
2744
2745 let obsolete_accounts =
2747 SerdeObsoleteAccountsMap::new_from_storages(&snapshot_storages, snapshot_slot);
2748
2749 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 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, i as u32, 1024,
2776 AccountsFileProvider::AppendVec,
2777 ));
2778 snapshot_storages.push(storage);
2779 }
2780
2781 write_obsolete_accounts_to_snapshot(
2783 bank_snapshot_dir,
2784 &snapshot_storages,
2785 snapshot_slot,
2786 &IoSetupState::default(),
2787 )
2788 .unwrap();
2789
2790 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 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2809
2810 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 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 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 fs::remove_file(&version_path).unwrap();
2832 fs::write(&version_path, SnapshotVersion::default().as_str()).unwrap();
2833
2834 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2836
2837 fs::File::create_new(&serialized_bank_path).unwrap();
2839 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2840
2841 fs::remove_file(&serialized_bank_path).unwrap();
2843 fs::write(&serialized_bank_path, "serialized bank").unwrap();
2844
2845 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2847
2848 fs::File::create_new(&status_cache_path).unwrap();
2850 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2851
2852 fs::remove_file(&status_cache_path).unwrap();
2854 fs::write(&status_cache_path, "status cache").unwrap();
2855
2856 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 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 let stale = account_path.path().join(AccountsFile::file_name(300, 3));
2868 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}