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