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, StartupHints,
9 StorageListItem, 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_STARTUP_HINTS_FILE_SIZE: u64 = 4096;
80pub 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;
85
86const SNAPSHOT_FASTBOOT_VERSION: Version = Version::new(3, 1, 0);
101
102#[derive(PartialEq, Eq, Debug)]
105pub struct BankSnapshotInfo {
106 pub slot: Slot,
108 pub snapshot_dir: PathBuf,
110 pub snapshot_version: SnapshotVersion,
112 pub fastboot_version: Option<Version>,
114}
115
116impl PartialOrd for BankSnapshotInfo {
117 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
118 Some(self.cmp(other))
119 }
120}
121
122impl Ord for BankSnapshotInfo {
124 fn cmp(&self, other: &Self) -> Ordering {
125 self.slot.cmp(&other.slot)
126 }
127}
128
129impl BankSnapshotInfo {
130 pub fn new_from_dir(
131 bank_snapshots_dir: impl AsRef<Path>,
132 slot: Slot,
133 ) -> std::result::Result<BankSnapshotInfo, SnapshotNewFromDirError> {
134 let bank_snapshot_dir = snapshot_paths::get_bank_snapshot_dir(&bank_snapshots_dir, slot);
137
138 if !bank_snapshot_dir.is_dir() {
139 return Err(SnapshotNewFromDirError::InvalidBankSnapshotDir(
140 bank_snapshot_dir,
141 ));
142 }
143
144 let version_path = bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
151 let version_file_info = FileInfo::new_from_path(&version_path)
152 .map_err(|err| SnapshotNewFromDirError::IncompleteDir(err, version_path))?;
153 let version_str = snapshot_version_from_file(version_file_info).map_err(|err| {
154 SnapshotNewFromDirError::IncompleteDir(err, bank_snapshot_dir.clone())
155 })?;
156
157 let snapshot_version = SnapshotVersion::from_str(version_str.as_str())
158 .or(Err(SnapshotNewFromDirError::InvalidVersion(version_str)))?;
159
160 let status_cache_file =
161 bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME);
162 if !status_cache_file.is_file() {
163 return Err(SnapshotNewFromDirError::MissingStatusCacheFile(
164 status_cache_file,
165 ));
166 }
167
168 let bank_snapshot_path =
169 bank_snapshot_dir.join(snapshot_paths::get_snapshot_file_name(slot));
170 if !bank_snapshot_path.is_file() {
171 return Err(SnapshotNewFromDirError::MissingSnapshotFile(
172 bank_snapshot_dir,
173 ));
174 };
175
176 let snapshot_fastboot_version_path =
177 bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_FASTBOOT_VERSION_FILENAME);
178
179 let fastboot_version = fs::read_to_string(&snapshot_fastboot_version_path)
183 .ok()
184 .map(|version_string| {
185 Version::from_str(version_string.trim())
186 .map_err(|_| SnapshotNewFromDirError::InvalidFastbootVersion(version_string))
187 })
188 .transpose()?;
189
190 Ok(BankSnapshotInfo {
191 slot,
192 snapshot_dir: bank_snapshot_dir,
193 snapshot_version,
194 fastboot_version,
195 })
196 }
197
198 pub fn snapshot_path(&self) -> PathBuf {
199 self.snapshot_dir
200 .join(snapshot_paths::get_snapshot_file_name(self.slot))
201 }
202}
203
204#[derive(Clone, Copy, Debug, Eq, PartialEq)]
208pub enum SnapshotFrom {
209 Archive,
211 Dir,
213}
214
215#[derive(Debug)]
218pub struct SnapshotRootPaths {
219 pub full_snapshot_root_file_path: PathBuf,
220 pub incremental_snapshot_root_file_path: Option<PathBuf>,
221}
222
223#[derive(Debug)]
225pub struct UnarchivedSnapshot {
226 unpack_dir: TempDir,
227 pub storage: AccountStorageMap,
228 pub bank_fields: BankFieldsToDeserialize,
229 pub(crate) accounts_db_fields: AccountsDbFields,
230 pub unpacked_snapshots_dir_and_version: UnpackedSnapshotsDirAndVersion,
231 pub measure_untar: Measure,
232}
233
234#[derive(Debug)]
236pub struct UnarchivedSnapshots {
237 pub full_storage: AccountStorageMap,
238 pub incremental_storage: Option<AccountStorageMap>,
239 pub bank_fields: SnapshotBankFields,
240 pub accounts_db_fields: SnapshotAccountsDbFields,
241 pub full_unpacked_snapshots_dir_and_version: UnpackedSnapshotsDirAndVersion,
242 pub incremental_unpacked_snapshots_dir_and_version: Option<UnpackedSnapshotsDirAndVersion>,
243 pub full_measure_untar: Measure,
244 pub incremental_measure_untar: Option<Measure>,
245 pub next_append_vec_id: AtomicAccountsFileId,
246}
247
248#[expect(dead_code)]
251#[derive(Debug)]
252pub struct UnarchivedSnapshotsGuard {
253 full_unpack_dir: TempDir,
254 incremental_unpack_dir: Option<TempDir>,
255}
256#[derive(Debug)]
258pub struct UnpackedSnapshotsDirAndVersion {
259 pub unpacked_snapshots_dir: PathBuf,
260 pub snapshot_version: SnapshotVersion,
261}
262
263pub(crate) struct StorageAndNextAccountsFileId {
266 pub storage: AccountStorageMap,
267 pub next_append_vec_id: AtomicAccountsFileId,
268}
269
270pub fn purge_incomplete_bank_snapshots(bank_snapshots_dir: impl AsRef<Path>) {
272 let Ok(read_dir_iter) = std::fs::read_dir(&bank_snapshots_dir) else {
273 return;
275 };
276
277 let is_incomplete = |dir: &PathBuf| !is_bank_snapshot_complete(dir);
278
279 let incomplete_dirs: Vec<_> = read_dir_iter
280 .filter_map(|entry| entry.ok())
281 .map(|entry| entry.path())
282 .filter(|path| path.is_dir())
283 .filter(is_incomplete)
284 .collect();
285
286 for incomplete_dir in incomplete_dirs {
288 let result = purge_bank_snapshot(&incomplete_dir);
289 match result {
290 Ok(_) => info!(
291 "Purged incomplete snapshot dir: {}",
292 incomplete_dir.display()
293 ),
294 Err(err) => warn!("Failed to purge incomplete snapshot dir: {err}"),
295 }
296 }
297}
298
299fn is_bank_snapshot_complete(bank_snapshot_dir: impl AsRef<Path>) -> bool {
301 let version_path = bank_snapshot_dir
302 .as_ref()
303 .join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
304
305 let Ok(version_file_info) = FileInfo::new_from_path(&version_path) else {
306 return false;
308 };
309
310 let Ok(version_str) = snapshot_version_from_file(version_file_info) else {
311 return false;
313 };
314
315 let Ok(_snapshot_version) = SnapshotVersion::from_str(version_str.as_str()) else {
316 return false;
318 };
319
320 let Some(slot) = bank_snapshot_dir.as_ref().file_name() else {
322 return false;
323 };
324 let Some(slot) = slot.to_str() else {
325 return false;
326 };
327 for file_name in [slot, snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME] {
328 let file_path = bank_snapshot_dir.as_ref().join(file_name);
329 let Ok(file_info) = FileInfo::new_from_path(file_path) else {
330 return false;
332 };
333 if file_info.size == 0 {
334 return false;
336 }
337 }
338
339 true
340}
341
342pub fn mark_bank_snapshot_as_loadable(bank_snapshot_dir: impl AsRef<Path>) -> io::Result<()> {
344 let snapshot_fastboot_version_path = bank_snapshot_dir
345 .as_ref()
346 .join(snapshot_paths::SNAPSHOT_FASTBOOT_VERSION_FILENAME);
347 fs::write(
348 &snapshot_fastboot_version_path,
349 SNAPSHOT_FASTBOOT_VERSION.to_string(),
350 )
351 .map_err(|err| {
352 IoError::other(format!(
353 "failed to write fastboot version file '{}': {err}",
354 snapshot_fastboot_version_path.display(),
355 ))
356 })?;
357 Ok(())
358}
359
360fn is_bank_snapshot_loadable(
362 fastboot_version: Option<&Version>,
363) -> std::result::Result<bool, SnapshotFastbootError> {
364 if let Some(fastboot_version) = fastboot_version {
365 is_snapshot_fastboot_compatible(fastboot_version)
366 } else {
367 Ok(false)
369 }
370}
371
372fn is_snapshot_fastboot_compatible(
374 version: &Version,
375) -> std::result::Result<bool, SnapshotFastbootError> {
376 match version.major {
377 3 => Ok(true),
379 2 => Ok(true),
382 v if v > SNAPSHOT_FASTBOOT_VERSION.major => {
383 Err(SnapshotFastbootError::IncompatibleVersion(version.clone()))
384 }
385 _ => Ok(false),
387 }
388}
389
390pub fn get_highest_loadable_bank_snapshot(
394 snapshot_config: &SnapshotConfig,
395) -> Option<BankSnapshotInfo> {
396 let highest_bank_snapshot = get_highest_bank_snapshot(&snapshot_config.bank_snapshots_dir)?;
397
398 let is_bank_snapshot_loadable =
399 is_bank_snapshot_loadable(highest_bank_snapshot.fastboot_version.as_ref());
400
401 match is_bank_snapshot_loadable {
402 Ok(true) => Some(highest_bank_snapshot),
403 Ok(false) => None,
404 Err(err) => {
405 warn!(
406 "Bank snapshot is not loadable '{}': {err}",
407 highest_bank_snapshot.snapshot_dir.display()
408 );
409 None
410 }
411 }
412}
413
414pub fn remove_tmp_snapshot_archives(snapshot_archives_dir: impl AsRef<Path>) {
417 if let Ok(entries) = std::fs::read_dir(snapshot_archives_dir) {
418 for entry in entries.flatten() {
419 if entry
420 .file_name()
421 .to_str()
422 .map(|file_name| file_name.starts_with(snapshot_paths::TMP_SNAPSHOT_ARCHIVE_PREFIX))
423 .unwrap_or(false)
424 {
425 let path = entry.path();
426 let result = if path.is_dir() {
427 fs::remove_dir_all(&path)
428 } else {
429 fs::remove_file(&path)
430 };
431 if let Err(err) = result {
432 warn!(
433 "Failed to remove temporary snapshot archive '{}': {err}",
434 path.display(),
435 );
436 }
437 }
438 }
439 }
440}
441
442pub fn archive_snapshot_package(
444 snapshot_archive_kind: SnapshotArchiveKind,
445 snapshot_slot: Slot,
446 snapshot_hash: SnapshotHash,
447 bank_snapshot_dir: impl AsRef<Path>,
448 mut snapshot_storages: Vec<Arc<AccountStorageEntry>>,
449 snapshot_config: &SnapshotConfig,
450 io_setup: &IoSetupState,
451) -> Result<SnapshotArchiveInfo> {
452 let snapshot_archive_path = match snapshot_archive_kind {
453 SnapshotArchiveKind::Full => snapshot_paths::build_full_snapshot_archive_path(
454 &snapshot_config.full_snapshot_archives_dir,
455 snapshot_slot,
456 &snapshot_hash,
457 snapshot_config.archive_format,
458 ),
459 SnapshotArchiveKind::Incremental(incremental_snapshot_base_slot) => {
460 snapshot_storages.retain(|storage| storage.slot() > incremental_snapshot_base_slot);
463 snapshot_paths::build_incremental_snapshot_archive_path(
464 &snapshot_config.incremental_snapshot_archives_dir,
465 incremental_snapshot_base_slot,
466 snapshot_slot,
467 &snapshot_hash,
468 snapshot_config.archive_format,
469 )
470 }
471 };
472
473 let snapshot_archive_info = archive_snapshot(
474 snapshot_archive_kind,
475 snapshot_slot,
476 snapshot_hash,
477 snapshot_storages.as_slice(),
478 &bank_snapshot_dir,
479 snapshot_archive_path,
480 snapshot_config.archive_format,
481 io_setup,
482 )?;
483
484 Ok(snapshot_archive_info)
485}
486
487pub fn serialize_snapshot(
489 bank_snapshots_dir: impl AsRef<Path>,
490 snapshot_version: SnapshotVersion,
491 bank_snapshot_package: BankSnapshotPackage,
492 snapshot_storages: &[Arc<AccountStorageEntry>],
493 should_finalize: bool,
494 startup_hints: &StartupHints,
495 io_setup: &IoSetupState,
496) -> Result<BankSnapshotInfo> {
497 let BankSnapshotPackage {
498 mut bank_fields,
499 bank_hash_stats,
500 status_cache_slot_deltas,
501 } = bank_snapshot_package;
502 let status_cache_slot_deltas = status_cache_slot_deltas.as_slice();
503 let slot = bank_fields.slot;
504
505 let do_serialize_snapshot = || {
508 let mut measure_everything = Measure::start("");
509 let bank_snapshot_dir = snapshot_paths::get_bank_snapshot_dir(&bank_snapshots_dir, slot);
510 if bank_snapshot_dir.exists() {
511 return Err(AddBankSnapshotError::SnapshotDirAlreadyExists(
512 bank_snapshot_dir,
513 ));
514 }
515 fs::create_dir_all(&bank_snapshot_dir).map_err(|err| {
516 AddBankSnapshotError::CreateSnapshotDir(err, bank_snapshot_dir.clone())
517 })?;
518
519 let bank_snapshot_path =
521 bank_snapshot_dir.join(snapshot_paths::get_snapshot_file_name(slot));
522 info!(
523 "Creating bank snapshot for slot {slot} at '{}'",
524 bank_snapshot_path.display(),
525 );
526
527 let bank_snapshot_serializer = move |stream: &mut dyn Write| -> Result<()> {
528 let versioned_epoch_stakes = mem::take(&mut bank_fields.versioned_epoch_stakes);
529 let extra_fields = ExtraFieldsToSerialize {
530 lamports_per_signature: bank_fields.fee_rate_governor.lamports_per_signature,
531 unused_incremental_snapshot_persistence: None,
532 unused_epoch_accounts_hash: None,
533 versioned_epoch_stakes,
534 accounts_lt_hash: Some(bank_fields.accounts_lt_hash.clone().into()),
535 block_id: Some(bank_fields.block_id),
536 };
537 serde_snapshot::serialize_bank_snapshot_into_wincode(
538 stream,
539 bank_fields,
540 bank_hash_stats,
541 extra_fields,
542 )?;
543 Ok(())
544 };
545 let (bank_snapshot_consumed_size, bank_serialize) = measure_time!(
546 serialize_snapshot_data_file(&bank_snapshot_path, io_setup, bank_snapshot_serializer)
547 .map_err(|err| AddBankSnapshotError::SerializeBank(Box::new(err)))?,
548 "bank serialize"
549 );
550
551 let status_cache_path =
552 bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME);
553 let (status_cache_consumed_size, status_cache_serialize_us) = measure_us!(
554 serde_snapshot::serialize_status_cache(
555 status_cache_slot_deltas,
556 &status_cache_path,
557 io_setup,
558 )
559 .map_err(|err| AddBankSnapshotError::SerializeStatusCache(Box::new(err)))?
560 );
561
562 let version_path = bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
563 let (_, write_version_file_us) = measure_us!(
564 fs::write(&version_path, snapshot_version.as_str().as_bytes(),)
565 .map_err(|err| AddBankSnapshotError::WriteSnapshotVersionFile(err, version_path))?
566 );
567
568 let (flush_storages_us, serialize_obsolete_accounts_us, write_storages_list_us) =
569 if should_finalize {
570 let flush_measure = Measure::start("");
571 for storage in snapshot_storages {
572 storage.flush().map_err(|err| {
573 AddBankSnapshotError::FlushStorage(err, storage.path().to_path_buf())
574 })?;
575 storage.disable_remove_on_drop();
578 }
579 let flush_us = flush_measure.end_as_us();
580
581 let (_, serialize_obsolete_accounts_us) = measure_us!({
582 write_obsolete_accounts_to_snapshot(
583 &bank_snapshot_dir,
584 snapshot_storages,
585 slot,
586 io_setup,
587 )
588 .map_err(|err| AddBankSnapshotError::SerializeObsoleteAccounts(Box::new(err)))?
589 });
590
591 let (_, write_storages_list_us) = measure_us!(
592 write_storages_list_to_snapshot(
593 &bank_snapshot_dir,
594 snapshot_storages,
595 io_setup,
596 )
597 .map_err(|err| AddBankSnapshotError::WriteStoragesList(Box::new(err)))?
598 );
599
600 write_startup_hints_to_snapshot(&bank_snapshot_dir, startup_hints, io_setup)
601 .map_err(|err| AddBankSnapshotError::WriteStartupHints(Box::new(err)))?;
602
603 mark_bank_snapshot_as_loadable(&bank_snapshot_dir)
604 .map_err(AddBankSnapshotError::MarkSnapshotLoadable)?;
605
606 (
607 Some(flush_us),
608 Some(serialize_obsolete_accounts_us),
609 Some(write_storages_list_us),
610 )
611 } else {
612 (None, None, None)
613 };
614
615 measure_everything.stop();
616
617 datapoint_info!(
619 "snapshot_bank",
620 ("slot", slot, i64),
621 ("bank_size", bank_snapshot_consumed_size, i64),
622 ("num_storages", snapshot_storages.len(), i64),
623 ("status_cache_size", status_cache_consumed_size, i64),
624 ("flush_storages_us", flush_storages_us, Option<i64>),
625 ("serialize_obsolete_accounts_us", serialize_obsolete_accounts_us, Option<i64>),
626 ("write_storages_list_us", write_storages_list_us, Option<i64>),
627 ("bank_serialize_us", bank_serialize.as_us(), i64),
628 ("status_cache_serialize_us", status_cache_serialize_us, i64),
629 ("write_version_file_us", write_version_file_us, i64),
630 ("total_us", measure_everything.as_us(), i64),
631 );
632
633 info!(
634 "{} for slot {} at {}",
635 bank_serialize,
636 slot,
637 bank_snapshot_path.display(),
638 );
639
640 Ok(BankSnapshotInfo {
641 slot,
642 snapshot_dir: bank_snapshot_dir,
643 snapshot_version,
644 fastboot_version: None,
645 })
646 };
647
648 do_serialize_snapshot().map_err(|err| SnapshotError::AddBankSnapshot(err, slot))
649}
650
651pub fn get_bank_snapshots(bank_snapshots_dir: impl AsRef<Path>) -> Vec<BankSnapshotInfo> {
653 let mut bank_snapshots = Vec::default();
654 match fs::read_dir(&bank_snapshots_dir) {
655 Err(err) => {
656 info!(
657 "Unable to read bank snapshots directory '{}': {err}",
658 bank_snapshots_dir.as_ref().display(),
659 );
660 }
661 Ok(paths) => paths
662 .filter_map(|entry| {
663 entry
666 .ok()
667 .filter(|entry| entry.path().is_dir())
668 .and_then(|entry| {
669 entry
670 .path()
671 .file_name()
672 .and_then(|file_name| file_name.to_str())
673 .and_then(|file_name| file_name.parse::<Slot>().ok())
674 })
675 })
676 .for_each(
677 |slot| match BankSnapshotInfo::new_from_dir(&bank_snapshots_dir, slot) {
678 Ok(snapshot_info) => bank_snapshots.push(snapshot_info),
679 Err(err) => debug!("Unable to read bank snapshot for slot {slot}: {err}"),
682 },
683 ),
684 }
685 bank_snapshots
686}
687
688pub fn get_highest_bank_snapshot(bank_snapshots_dir: impl AsRef<Path>) -> Option<BankSnapshotInfo> {
692 do_get_highest_bank_snapshot(get_bank_snapshots(&bank_snapshots_dir))
693}
694
695fn do_get_highest_bank_snapshot(
696 mut bank_snapshots: Vec<BankSnapshotInfo>,
697) -> Option<BankSnapshotInfo> {
698 bank_snapshots.sort_unstable();
699 bank_snapshots.into_iter().next_back()
700}
701
702pub fn write_obsolete_accounts_to_snapshot(
703 bank_snapshot_dir: impl AsRef<Path>,
704 snapshot_storages: &[Arc<AccountStorageEntry>],
705 snapshot_slot: Slot,
706 io_setup: &IoSetupState,
707) -> Result<u64> {
708 let obsolete_accounts =
709 SerdeObsoleteAccountsMap::new_from_storages(snapshot_storages, snapshot_slot);
710 serialize_obsolete_accounts(
711 bank_snapshot_dir,
712 &obsolete_accounts,
713 MAX_OBSOLETE_ACCOUNTS_FILE_SIZE,
714 io_setup,
715 )
716}
717
718fn serialize_obsolete_accounts(
719 bank_snapshot_dir: impl AsRef<Path>,
720 obsolete_accounts_map: &SerdeObsoleteAccountsMap,
721 maximum_obsolete_accounts_file_size: u64,
722 io_setup: &IoSetupState,
723) -> Result<u64> {
724 let obsolete_accounts_path = bank_snapshot_dir
725 .as_ref()
726 .join(snapshot_paths::SNAPSHOT_OBSOLETE_ACCOUNTS_FILENAME);
727 let mut file_stream = SizeLimitedWriter::new(
728 large_file_buf_writer(&obsolete_accounts_path, io_setup)?,
729 maximum_obsolete_accounts_file_size,
730 );
731
732 serde_snapshot::serialize_into(&mut file_stream, obsolete_accounts_map).map_err(|err| {
733 IoError::other(format!(
734 "unable to serialize obsolete accounts to file '{}': {err}",
735 obsolete_accounts_path.display(),
736 ))
737 })?;
738
739 Ok(file_stream.bytes_written())
740}
741
742fn deserialize_obsolete_accounts(
743 bank_snapshot_dir: impl AsRef<Path>,
744 maximum_obsolete_accounts_file_size: u64,
745) -> Result<SerdeObsoleteAccountsMap> {
746 let obsolete_accounts_path = bank_snapshot_dir
747 .as_ref()
748 .join(snapshot_paths::SNAPSHOT_OBSOLETE_ACCOUNTS_FILENAME);
749 let obsolete_accounts_reader = ReadAdapter::new(large_file_buf_reader(
750 &obsolete_accounts_path,
751 AUX_SNAPSHOT_FILE_READ_BUF_SIZE,
752 &IoSetupState::default(),
753 )?);
754 let obsolete_accounts_file_metadata = fs::metadata(&obsolete_accounts_path)?;
756 if obsolete_accounts_file_metadata.len() > maximum_obsolete_accounts_file_size {
757 let error_message = format!(
758 "too large obsolete accounts file to deserialize: '{}' has {} bytes (max size is \
759 {maximum_obsolete_accounts_file_size} bytes)",
760 obsolete_accounts_path.display(),
761 obsolete_accounts_file_metadata.len(),
762 );
763 return Err(IoError::other(error_message).into());
764 }
765
766 Ok(serde_snapshot::deserialize_wincode_from(
767 obsolete_accounts_reader,
768 )?)
769}
770
771pub fn write_storages_list_to_snapshot(
772 bank_snapshot_dir: impl AsRef<Path>,
773 snapshot_storages: &[Arc<AccountStorageEntry>],
774 io_setup: &IoSetupState,
775) -> Result<FileSize> {
776 let storages_list = StoragesList::new_from_storages(snapshot_storages);
777 serialize_storages_list_to_snapshot(bank_snapshot_dir, storages_list, io_setup)
778}
779
780fn serialize_storages_list_to_snapshot(
781 bank_snapshot_dir: impl AsRef<Path>,
782 storages_list: StoragesList,
783 io_setup: &IoSetupState,
784) -> Result<FileSize> {
785 let storages_list_path = bank_snapshot_dir
786 .as_ref()
787 .join(snapshot_paths::SNAPSHOT_STORAGES_LIST_FILENAME);
788 let mut file_stream = SizeLimitedWriter::new(
789 large_file_buf_writer(&storages_list_path, io_setup)?,
790 MAX_STORAGES_LIST_FILE_SIZE,
791 );
792 serde_snapshot::serialize_into(&mut file_stream, &storages_list).map_err(|err| {
793 IoError::other(format!(
794 "unable to serialize storages list to file '{}': {err}",
795 storages_list_path.display(),
796 ))
797 })?;
798 Ok(file_stream.bytes_written())
799}
800
801fn deserialize_storages_list(
802 storages_list_path: &Path,
803 maximum_storages_list_file_size: u64,
804) -> Result<StoragesList> {
805 let storages_list_reader = ReadAdapter::new(large_file_buf_reader(
806 storages_list_path,
807 AUX_SNAPSHOT_FILE_READ_BUF_SIZE,
808 &IoSetupState::default(),
809 )?);
810 let storages_list_file_metadata = fs::metadata(storages_list_path)?;
812 if storages_list_file_metadata.len() > maximum_storages_list_file_size {
813 let error_message = format!(
814 "too large storages list file to deserialize: '{}' has {} bytes (max size is \
815 {maximum_storages_list_file_size} bytes)",
816 storages_list_path.display(),
817 storages_list_file_metadata.len(),
818 );
819 return Err(IoError::other(error_message).into());
820 }
821
822 Ok(serde_snapshot::deserialize_wincode_from(
823 storages_list_reader,
824 )?)
825}
826
827pub fn write_startup_hints_to_snapshot(
828 bank_snapshot_dir: impl AsRef<Path>,
829 startup_hints: &StartupHints,
830 io_setup: &IoSetupState,
831) -> Result<FileSize> {
832 let startup_hints_path = bank_snapshot_dir
833 .as_ref()
834 .join(snapshot_paths::SNAPSHOT_STARTUP_HINTS_FILENAME);
835 let mut file_stream = SizeLimitedWriter::new(
836 large_file_buf_writer(&startup_hints_path, io_setup)?,
837 MAX_STARTUP_HINTS_FILE_SIZE,
838 );
839 serde_snapshot::serialize_into(&mut file_stream, startup_hints).map_err(|err| {
840 IoError::other(format!(
841 "unable to serialize startup hints to file '{}': {err}",
842 startup_hints_path.display(),
843 ))
844 })?;
845 Ok(file_stream.bytes_written())
846}
847
848pub fn read_startup_hints(bank_snapshot_dir: impl AsRef<Path>) -> Result<Option<StartupHints>> {
853 let startup_hints_path = bank_snapshot_dir
854 .as_ref()
855 .join(snapshot_paths::SNAPSHOT_STARTUP_HINTS_FILENAME);
856 if !startup_hints_path.exists() {
857 return Ok(None);
858 }
859
860 let startup_hints_file_metadata = fs::metadata(&startup_hints_path)?;
861 if startup_hints_file_metadata.len() > MAX_STARTUP_HINTS_FILE_SIZE {
862 let error_message = format!(
863 "too large startup hints file to deserialize: '{}' has {} bytes (max size is \
864 {MAX_STARTUP_HINTS_FILE_SIZE} bytes)",
865 startup_hints_path.display(),
866 startup_hints_file_metadata.len(),
867 );
868 return Err(IoError::other(error_message).into());
869 }
870 let startup_hints_reader = ReadAdapter::new(large_file_buf_reader(
871 &startup_hints_path,
872 AUX_SNAPSHOT_FILE_READ_BUF_SIZE,
873 &IoSetupState::default(),
874 )?);
875
876 Ok(StartupHints::read_from(startup_hints_reader)?)
877}
878
879pub fn serialize_snapshot_data_file<F>(
880 data_file_path: &Path,
881 io_setup: &IoSetupState,
882 serializer: F,
883) -> Result<u64>
884where
885 F: FnOnce(&mut dyn Write) -> Result<()>,
886{
887 serialize_snapshot_data_file_capped::<F>(
888 data_file_path,
889 MAX_SNAPSHOT_DATA_FILE_SIZE,
890 io_setup,
891 serializer,
892 )
893}
894
895pub fn deserialize_snapshot_data_file<T: Sized>(
896 data_file_path: &Path,
897 deserializer: impl FnOnce(&mut BufReader<std::fs::File>) -> Result<T>,
898) -> Result<T> {
899 let wrapped_deserializer = move |streams: &mut SnapshotStreams<std::fs::File>| -> Result<T> {
900 deserializer(streams.full_snapshot_stream)
901 };
902
903 let wrapped_data_file_path = SnapshotRootPaths {
904 full_snapshot_root_file_path: data_file_path.to_path_buf(),
905 incremental_snapshot_root_file_path: None,
906 };
907
908 deserialize_snapshot_data_files_capped(
909 &wrapped_data_file_path,
910 MAX_SNAPSHOT_DATA_FILE_SIZE,
911 wrapped_deserializer,
912 )
913}
914
915pub fn deserialize_snapshot_data_files<T: Sized>(
916 snapshot_root_paths: &SnapshotRootPaths,
917 deserializer: impl FnOnce(&mut SnapshotStreams<std::fs::File>) -> Result<T>,
918) -> Result<T> {
919 deserialize_snapshot_data_files_capped(
920 snapshot_root_paths,
921 MAX_SNAPSHOT_DATA_FILE_SIZE,
922 deserializer,
923 )
924}
925
926fn serialize_snapshot_data_file_capped<F>(
927 data_file_path: &Path,
928 maximum_file_size: u64,
929 io_setup: &IoSetupState,
930 serializer: F,
931) -> Result<u64>
932where
933 F: FnOnce(&mut dyn Write) -> Result<()>,
934{
935 let mut data_file_stream = SizeLimitedWriter::new(
936 large_file_buf_writer(data_file_path, io_setup)?,
937 maximum_file_size,
938 );
939 serializer(&mut data_file_stream).map_err(|err| {
940 IoError::other(format!(
941 "unable to serialize snapshot data to file '{}': {err}",
942 data_file_path.display(),
943 ))
944 })?;
945 data_file_stream.flush()?;
946 Ok(data_file_stream.bytes_written())
947}
948
949fn deserialize_snapshot_data_files_capped<T: Sized>(
950 snapshot_root_paths: &SnapshotRootPaths,
951 maximum_file_size: u64,
952 deserializer: impl FnOnce(&mut SnapshotStreams<std::fs::File>) -> Result<T>,
953) -> Result<T> {
954 let (full_snapshot_file_size, mut full_snapshot_data_file_stream) =
955 create_snapshot_data_file_stream(
956 &snapshot_root_paths.full_snapshot_root_file_path,
957 maximum_file_size,
958 )?;
959
960 let (incremental_snapshot_file_size, mut incremental_snapshot_data_file_stream) =
961 if let Some(ref incremental_snapshot_root_file_path) =
962 snapshot_root_paths.incremental_snapshot_root_file_path
963 {
964 Some(create_snapshot_data_file_stream(
965 incremental_snapshot_root_file_path,
966 maximum_file_size,
967 )?)
968 } else {
969 None
970 }
971 .unzip();
972
973 let mut snapshot_streams = SnapshotStreams {
974 full_snapshot_stream: &mut full_snapshot_data_file_stream,
975 incremental_snapshot_stream: incremental_snapshot_data_file_stream.as_mut(),
976 };
977 let ret = deserializer(&mut snapshot_streams)?;
978
979 check_deserialize_file_consumed(
980 full_snapshot_file_size,
981 &snapshot_root_paths.full_snapshot_root_file_path,
982 &mut full_snapshot_data_file_stream,
983 )?;
984
985 if let Some(ref incremental_snapshot_root_file_path) =
986 snapshot_root_paths.incremental_snapshot_root_file_path
987 {
988 check_deserialize_file_consumed(
989 incremental_snapshot_file_size.unwrap(),
990 incremental_snapshot_root_file_path,
991 incremental_snapshot_data_file_stream.as_mut().unwrap(),
992 )?;
993 }
994
995 Ok(ret)
996}
997
998fn create_snapshot_data_file_stream(
1001 snapshot_root_file_path: impl AsRef<Path>,
1002 maximum_file_size: u64,
1003) -> Result<(u64, BufReader<std::fs::File>)> {
1004 let snapshot_file_size = fs::metadata(&snapshot_root_file_path)?.len();
1005
1006 if snapshot_file_size > maximum_file_size {
1007 let error_message = format!(
1008 "too large snapshot data file to deserialize: '{}' has {} bytes (max size is {} bytes)",
1009 snapshot_root_file_path.as_ref().display(),
1010 snapshot_file_size,
1011 maximum_file_size,
1012 );
1013 return Err(IoError::other(error_message).into());
1014 }
1015
1016 let snapshot_data_file = fs::File::open(snapshot_root_file_path)?;
1017 let snapshot_data_file_stream = BufReader::new(snapshot_data_file);
1018
1019 Ok((snapshot_file_size, snapshot_data_file_stream))
1020}
1021
1022fn check_deserialize_file_consumed(
1025 file_size: u64,
1026 file_path: impl AsRef<Path>,
1027 file_stream: &mut BufReader<std::fs::File>,
1028) -> Result<()> {
1029 let consumed_size = file_stream.stream_position()?;
1030
1031 if consumed_size != file_size {
1032 let error_message = format!(
1033 "invalid snapshot data file: '{}' has {} bytes, however consumed {} bytes to \
1034 deserialize",
1035 file_path.as_ref().display(),
1036 file_size,
1037 consumed_size,
1038 );
1039 return Err(IoError::other(error_message).into());
1040 }
1041
1042 Ok(())
1043}
1044
1045pub fn verify_and_unarchive_snapshots(
1047 bank_snapshots_dir: impl AsRef<Path>,
1048 full_snapshot_archive_info: &FullSnapshotArchiveInfo,
1049 incremental_snapshot_archive_info: Option<&IncrementalSnapshotArchiveInfo>,
1050 account_paths: &[PathBuf],
1051 io_setup: &IoSetupState,
1052) -> Result<(UnarchivedSnapshots, UnarchivedSnapshotsGuard)> {
1053 check_are_snapshots_compatible(
1054 full_snapshot_archive_info,
1055 incremental_snapshot_archive_info,
1056 )?;
1057
1058 let next_append_vec_id = Arc::new(AtomicAccountsFileId::new(0));
1059 let UnarchivedSnapshot {
1060 unpack_dir: full_unpack_dir,
1061 storage: full_storage,
1062 bank_fields: full_bank_fields,
1063 accounts_db_fields: full_accounts_db_fields,
1064 unpacked_snapshots_dir_and_version: full_unpacked_snapshots_dir_and_version,
1065 measure_untar: full_measure_untar,
1066 } = unarchive_snapshot(
1067 &bank_snapshots_dir,
1068 snapshot_paths::TMP_SNAPSHOT_ARCHIVE_PREFIX,
1069 full_snapshot_archive_info.path(),
1070 "snapshot untar",
1071 account_paths,
1072 full_snapshot_archive_info.archive_format(),
1073 next_append_vec_id.clone(),
1074 io_setup,
1075 )?;
1076
1077 let (
1078 incremental_unpack_dir,
1079 incremental_storage,
1080 incremental_bank_fields,
1081 incremental_accounts_db_fields,
1082 incremental_unpacked_snapshots_dir_and_version,
1083 incremental_measure_untar,
1084 ) = if let Some(incremental_snapshot_archive_info) = incremental_snapshot_archive_info {
1085 let UnarchivedSnapshot {
1086 unpack_dir,
1087 storage,
1088 bank_fields,
1089 accounts_db_fields,
1090 unpacked_snapshots_dir_and_version,
1091 measure_untar,
1092 } = unarchive_snapshot(
1093 &bank_snapshots_dir,
1094 snapshot_paths::TMP_SNAPSHOT_ARCHIVE_PREFIX,
1095 incremental_snapshot_archive_info.path(),
1096 "incremental snapshot untar",
1097 account_paths,
1098 incremental_snapshot_archive_info.archive_format(),
1099 next_append_vec_id.clone(),
1100 io_setup,
1101 )?;
1102 (
1103 Some(unpack_dir),
1104 Some(storage),
1105 Some(bank_fields),
1106 Some(accounts_db_fields),
1107 Some(unpacked_snapshots_dir_and_version),
1108 Some(measure_untar),
1109 )
1110 } else {
1111 (None, None, None, None, None, None)
1112 };
1113
1114 let bank_fields = SnapshotBankFields::new(full_bank_fields, incremental_bank_fields);
1115 let accounts_db_fields =
1116 SnapshotAccountsDbFields::new(full_accounts_db_fields, incremental_accounts_db_fields);
1117 let next_append_vec_id = Arc::try_unwrap(next_append_vec_id).unwrap();
1118
1119 Ok((
1120 UnarchivedSnapshots {
1121 full_storage,
1122 incremental_storage,
1123 bank_fields,
1124 accounts_db_fields,
1125 full_unpacked_snapshots_dir_and_version,
1126 incremental_unpacked_snapshots_dir_and_version,
1127 full_measure_untar,
1128 incremental_measure_untar,
1129 next_append_vec_id,
1130 },
1131 UnarchivedSnapshotsGuard {
1132 full_unpack_dir,
1133 incremental_unpack_dir,
1134 },
1135 ))
1136}
1137
1138#[derive(PartialEq, Debug)]
1140enum SnapshotFileKind {
1141 Version,
1142 BankFields,
1143 Storage,
1144}
1145
1146fn get_snapshot_file_kind(filename: &str) -> Option<SnapshotFileKind> {
1148 static VERSION_FILE_REGEX: LazyLock<Regex> =
1149 LazyLock::new(|| Regex::new(r"^version$").unwrap());
1150 static BANK_FIELDS_FILE_REGEX: LazyLock<Regex> =
1151 LazyLock::new(|| Regex::new(r"^[0-9]+(\.pre)?$").unwrap());
1152
1153 if VERSION_FILE_REGEX.is_match(filename) {
1154 Some(SnapshotFileKind::Version)
1155 } else if BANK_FIELDS_FILE_REGEX.is_match(filename) {
1156 Some(SnapshotFileKind::BankFields)
1157 } else if get_slot_and_append_vec_id(filename).is_ok() {
1158 Some(SnapshotFileKind::Storage)
1159 } else {
1160 None
1161 }
1162}
1163
1164fn get_version_and_snapshot_files(
1168 file_receiver: &Receiver<FileInfo>,
1169) -> Result<(FileInfo, FileInfo, Vec<FileInfo>)> {
1170 let mut append_vec_files = Vec::with_capacity(1024);
1171 let mut snapshot_version = None;
1172 let mut snapshot_bank = None;
1173
1174 loop {
1175 if let Ok(file_info) = file_receiver.recv() {
1176 let filename = file_info.path.file_name().unwrap().to_str().unwrap();
1177 match get_snapshot_file_kind(filename) {
1178 Some(SnapshotFileKind::Version) => {
1179 snapshot_version = Some(file_info);
1180
1181 if snapshot_bank.is_some() {
1183 break;
1184 }
1185 }
1186 Some(SnapshotFileKind::BankFields) => {
1187 snapshot_bank = Some(file_info);
1188
1189 if snapshot_version.is_some() {
1191 break;
1192 }
1193 }
1194 Some(SnapshotFileKind::Storage) => {
1195 append_vec_files.push(file_info);
1196 }
1197 None => {} }
1199 } else {
1200 return Err(SnapshotError::RebuildStorages(
1201 "did not receive snapshot file from unpacking threads".to_string(),
1202 ));
1203 }
1204 }
1205 let snapshot_version = snapshot_version.unwrap();
1206 let snapshot_bank = snapshot_bank.unwrap();
1207
1208 Ok((snapshot_version, snapshot_bank, append_vec_files))
1209}
1210
1211struct SnapshotFieldsBundle {
1213 snapshot_version: SnapshotVersion,
1214 bank_fields: BankFieldsToDeserialize,
1215 accounts_db_fields: AccountsDbFields,
1216 append_vec_files: Vec<FileInfo>,
1217}
1218
1219fn snapshot_fields_from_files(file_receiver: &Receiver<FileInfo>) -> Result<SnapshotFieldsBundle> {
1222 let (snapshot_version, snapshot_bank, append_vec_files) =
1223 get_version_and_snapshot_files(file_receiver)?;
1224 let snapshot_version_str = snapshot_version_from_file(snapshot_version)?;
1225 let snapshot_version = snapshot_version_str.parse().map_err(|err| {
1226 IoError::other(format!(
1227 "unsupported snapshot version '{snapshot_version_str}': {err}",
1228 ))
1229 })?;
1230
1231 let mut snapshot_stream = BufReader::new(snapshot_bank.file);
1232 let (bank_fields, accounts_db_fields) = match snapshot_version {
1233 SnapshotVersion::V1_2_0 => serde_snapshot::fields_from_stream(&mut snapshot_stream)?,
1234 };
1235
1236 Ok(SnapshotFieldsBundle {
1237 snapshot_version,
1238 bank_fields,
1239 accounts_db_fields,
1240 append_vec_files,
1241 })
1242}
1243
1244fn create_snapshot_meta_files_for_unarchived_snapshot(unpack_dir: impl AsRef<Path>) -> Result<()> {
1249 let snapshots_dir = unpack_dir.as_ref().join(snapshot_paths::BANK_SNAPSHOTS_DIR);
1250 if !snapshots_dir.is_dir() {
1251 return Err(SnapshotError::NoSnapshotSlotDir(snapshots_dir));
1252 }
1253
1254 let slot_dir = std::fs::read_dir(&snapshots_dir)
1256 .map_err(|_| SnapshotError::NoSnapshotSlotDir(snapshots_dir.clone()))?
1257 .find(|entry| entry.as_ref().unwrap().path().is_dir())
1258 .ok_or_else(|| SnapshotError::NoSnapshotSlotDir(snapshots_dir.clone()))?
1259 .map_err(|_| SnapshotError::NoSnapshotSlotDir(snapshots_dir.clone()))?
1260 .path();
1261
1262 let version_file = unpack_dir
1263 .as_ref()
1264 .join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
1265 fs::hard_link(
1266 version_file,
1267 slot_dir.join(snapshot_paths::SNAPSHOT_VERSION_FILENAME),
1268 )?;
1269
1270 let status_cache_file = snapshots_dir.join(snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME);
1271 fs::hard_link(
1272 status_cache_file,
1273 slot_dir.join(snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME),
1274 )?;
1275
1276 Ok(())
1277}
1278
1279#[allow(clippy::too_many_arguments)]
1283fn unarchive_snapshot(
1284 bank_snapshots_dir: impl AsRef<Path>,
1285 unpacked_snapshots_dir_prefix: &'static str,
1286 snapshot_archive_path: impl AsRef<Path>,
1287 measure_name: &'static str,
1288 account_paths: &[PathBuf],
1289 archive_format: ArchiveFormat,
1290 next_append_vec_id: Arc<AtomicAccountsFileId>,
1291 io_setup: &IoSetupState,
1292) -> Result<UnarchivedSnapshot> {
1293 let unpack_dir = tempfile::Builder::new()
1294 .prefix(unpacked_snapshots_dir_prefix)
1295 .tempdir_in(bank_snapshots_dir)?;
1296 let unpacked_snapshots_dir = unpack_dir.path().join(snapshot_paths::BANK_SNAPSHOTS_DIR);
1297
1298 let (file_sender, file_receiver) = crossbeam_channel::unbounded();
1299 thread::scope(|scope| {
1300 let unarchive_handle = streaming_unarchive_snapshot(
1301 scope,
1302 file_sender,
1303 account_paths.to_vec(),
1304 unpack_dir.path().to_path_buf(),
1305 snapshot_archive_path.as_ref().to_path_buf(),
1306 archive_format,
1307 io_setup,
1308 );
1309
1310 let snapshot_result = snapshot_fields_from_files(&file_receiver).and_then(
1311 |SnapshotFieldsBundle {
1312 snapshot_version,
1313 bank_fields,
1314 accounts_db_fields,
1315 append_vec_files,
1316 ..
1317 }| {
1318 let (storage, measure_untar) = measure_time!(
1319 SnapshotStorageRebuilder::rebuild_storages(
1320 append_vec_files.into_iter().chain(file_receiver),
1321 next_append_vec_id,
1322 SnapshotFrom::Archive,
1323 None,
1324 )?,
1325 measure_name
1326 );
1327 info!("{measure_untar}");
1328 create_snapshot_meta_files_for_unarchived_snapshot(&unpack_dir)?;
1329
1330 Ok(UnarchivedSnapshot {
1331 unpack_dir,
1332 storage,
1333 bank_fields,
1334 accounts_db_fields,
1335 unpacked_snapshots_dir_and_version: UnpackedSnapshotsDirAndVersion {
1336 unpacked_snapshots_dir,
1337 snapshot_version,
1338 },
1339 measure_untar,
1340 })
1341 },
1342 );
1343 let unarchive_result = unarchive_handle.join().expect("must join unarchive thread");
1345 match (unarchive_result, snapshot_result) {
1346 (Err(SnapshotError::CrossbeamSend(_)), snap @ Err(_)) => snap,
1349 (Err(err), _) => Err(err),
1350 (Ok(()), snap) => snap,
1351 }
1352 })
1353}
1354
1355fn spawn_streaming_snapshot_dir_files(
1359 snapshot_file_path: PathBuf,
1360 snapshot_version_path: PathBuf,
1361 account_paths: &[PathBuf],
1362) -> (Receiver<FileInfo>, thread::JoinHandle<Result<()>>) {
1363 let (file_sender, file_receiver) = crossbeam_channel::unbounded();
1364 let account_paths = account_paths.to_vec();
1365
1366 let handle = thread::Builder::new()
1367 .name("solSnapDirFiles".to_string())
1368 .spawn(move || {
1369 let snapshot_bank_file_info = FileInfo::new_from_path(snapshot_file_path)?;
1370 file_sender.send(snapshot_bank_file_info)?;
1371 let snapshot_version_file_info = FileInfo::new_from_path(snapshot_version_path)?;
1372 file_sender.send(snapshot_version_file_info)?;
1373
1374 for account_path in account_paths {
1375 for dir_entry_result in fs::read_dir(account_path)? {
1376 let dir_entry = dir_entry_result?;
1377 let path = dir_entry.path();
1378 let file_info = FileInfo::new_from_path(path)?;
1379 file_sender.send(file_info)?;
1380 }
1381 }
1382 Ok::<_, SnapshotError>(())
1383 })
1384 .expect("should spawn thread");
1385
1386 (file_receiver, handle)
1387}
1388
1389fn migrate_legacy_hardlinks(bank_snapshot_dir: &Path, account_run_paths: &[PathBuf]) -> Result<()> {
1398 let accounts_hardlinks_dir =
1399 bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_ACCOUNTS_HARDLINKS);
1400 let mut items: Vec<StorageListItem> = Vec::new();
1401
1402 for entry in fs::read_dir(&accounts_hardlinks_dir).map_err(|err| {
1403 IoError::other(format!(
1404 "failed to read legacy accounts hardlinks dir '{}': {err}",
1405 accounts_hardlinks_dir.display(),
1406 ))
1407 })? {
1408 let symlink_path = entry?.path();
1409 let snapshot_slot_dir = fs::read_link(&symlink_path).map_err(|err| {
1410 IoError::other(format!(
1411 "failed to read symlink '{}': {err}",
1412 symlink_path.display(),
1413 ))
1414 })?;
1415 let run_dir = snapshot_slot_dir
1418 .parent()
1419 .and_then(Path::parent)
1420 .ok_or_else(|| {
1421 IoError::other(format!(
1422 "invalid legacy hardlink target '{}'",
1423 snapshot_slot_dir.display(),
1424 ))
1425 })?
1426 .join(ACCOUNTS_RUN_DIR);
1427 if !account_run_paths.contains(&run_dir) {
1432 return Err(IoError::other(format!(
1433 "legacy hardlink target '{}' points to run dir '{}' which is not in the current \
1434 account paths ({:?}); the account paths configuration has changed since this \
1435 snapshot was taken — load from a snapshot archive instead",
1436 snapshot_slot_dir.display(),
1437 run_dir.display(),
1438 account_run_paths,
1439 ))
1440 .into());
1441 }
1442
1443 for file_entry in fs::read_dir(&snapshot_slot_dir).map_err(|err| {
1444 IoError::other(format!(
1445 "failed to read legacy hardlink dir '{}': {err}",
1446 snapshot_slot_dir.display(),
1447 ))
1448 })? {
1449 let src = file_entry?.path();
1450 let Some(name) = src.file_name().and_then(|n| n.to_str()) else {
1451 continue;
1452 };
1453 let (slot, id) = get_slot_and_append_vec_id(name)?;
1454 let dest = run_dir.join(name);
1455 fs::rename(&src, &dest).map_err(|err| {
1456 IoError::other(format!(
1457 "failed to migrate legacy storage from '{}' to '{}': {err}",
1458 src.display(),
1459 dest.display(),
1460 ))
1461 })?;
1462 items.push(StorageListItem {
1463 slot,
1464 id: id as AccountsFileId,
1465 });
1466 }
1467 }
1468
1469 serialize_storages_list_to_snapshot(
1474 bank_snapshot_dir,
1475 StoragesList::from_items(items),
1476 &IoSetupState::default(),
1477 )?;
1478
1479 fs::remove_dir_all(&accounts_hardlinks_dir).map_err(|err| {
1483 IoError::other(format!(
1484 "failed to remove legacy accounts hardlinks dir '{}': {err}",
1485 accounts_hardlinks_dir.display(),
1486 ))
1487 })?;
1488 wipe_account_snapshot_dirs(account_run_paths);
1489
1490 mark_bank_snapshot_as_loadable(bank_snapshot_dir)?;
1493
1494 Ok(())
1495}
1496
1497fn prune_stale_storages(account_paths: &[PathBuf], storages_list: StoragesList) -> Result<()> {
1501 let expected_storages = storages_list.into_slot_file_id_set();
1502 for account_path in account_paths {
1503 let read_dir = fs::read_dir(account_path).map_err(|err| {
1504 IoError::other(format!(
1505 "failed to read account path '{}': {err}",
1506 account_path.display(),
1507 ))
1508 })?;
1509 for entry in read_dir {
1510 let path = entry?.path();
1511 let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
1512 continue;
1513 };
1514 let Ok((slot, id)) = get_slot_and_append_vec_id(name) else {
1515 continue;
1517 };
1518 if !expected_storages.contains(&(slot, id as AccountsFileId)) {
1519 info!(
1520 "Removing stale storage file '{}' not in storages list",
1521 path.display(),
1522 );
1523 fs::remove_file(&path)?
1524 }
1525 }
1526 }
1527 Ok(())
1528}
1529
1530pub(crate) fn rebuild_storages_from_snapshot_dir(
1535 snapshot_info: &BankSnapshotInfo,
1536 account_paths: &[PathBuf],
1537 next_append_vec_id: Arc<AtomicAccountsFileId>,
1538) -> Result<(AccountStorageMap, BankFieldsToDeserialize, AccountsDbFields)> {
1539 let bank_snapshot_dir = &snapshot_info.snapshot_dir;
1540
1541 let obsolete_accounts = snapshot_info
1545 .fastboot_version
1546 .as_ref()
1547 .is_some_and(|fastboot_version| fastboot_version.major >= 2)
1548 .then(|| deserialize_obsolete_accounts(bank_snapshot_dir, MAX_OBSOLETE_ACCOUNTS_FILE_SIZE))
1549 .transpose()
1550 .map_err(|err| {
1551 IoError::other(format!(
1552 "failed to read obsolete accounts file '{}': {err}",
1553 bank_snapshot_dir.display()
1554 ))
1555 })?;
1556
1557 let storages_list_path =
1561 bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_STORAGES_LIST_FILENAME);
1562 if !storages_list_path.exists() {
1563 migrate_legacy_hardlinks(bank_snapshot_dir, account_paths)?;
1569 }
1570 let storages_list =
1571 deserialize_storages_list(&storages_list_path, MAX_STORAGES_LIST_FILE_SIZE)?;
1572 prune_stale_storages(account_paths, storages_list)?;
1573
1574 let snapshot_file_path = snapshot_info.snapshot_path();
1575 let snapshot_version_path = bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
1576 let (file_receiver, stream_files_handle) = spawn_streaming_snapshot_dir_files(
1577 snapshot_file_path,
1578 snapshot_version_path,
1579 account_paths,
1580 );
1581
1582 let snapshot_result = snapshot_fields_from_files(&file_receiver).and_then(
1583 |SnapshotFieldsBundle {
1584 bank_fields,
1585 accounts_db_fields,
1586 append_vec_files,
1587 ..
1588 }| {
1589 let storage = SnapshotStorageRebuilder::rebuild_storages(
1590 append_vec_files.into_iter().chain(file_receiver),
1591 next_append_vec_id,
1592 SnapshotFrom::Dir,
1593 obsolete_accounts,
1594 )?;
1595 Ok((storage, bank_fields, accounts_db_fields))
1596 },
1597 );
1598
1599 let stream_files_result = stream_files_handle.join().expect("must join dir thread");
1601 match (stream_files_result, snapshot_result) {
1602 (Err(SnapshotError::CrossbeamSend(_)), snap @ Err(_)) => snap,
1605 (Err(err), _) => Err(err),
1606 (Ok(()), snap) => snap,
1607 }
1608}
1609
1610fn snapshot_version_from_file(mut file_info: FileInfo) -> io::Result<String> {
1614 let file_size = file_info.size;
1615 if file_size > MAX_SNAPSHOT_VERSION_FILE_SIZE {
1616 let error_message = format!(
1617 "snapshot version file too large: '{}' has {} bytes (max size is {} bytes)",
1618 file_info.path.display(),
1619 file_size,
1620 MAX_SNAPSHOT_VERSION_FILE_SIZE,
1621 );
1622 return Err(IoError::other(error_message));
1623 }
1624
1625 let mut snapshot_version = String::new();
1627 file_info
1628 .file
1629 .read_to_string(&mut snapshot_version)
1630 .map_err(|err| {
1631 IoError::other(format!(
1632 "failed to read snapshot version from file '{}': {err}",
1633 file_info.path.display()
1634 ))
1635 })?;
1636
1637 Ok(snapshot_version.trim().to_string())
1638}
1639
1640fn check_are_snapshots_compatible(
1643 full_snapshot_archive_info: &FullSnapshotArchiveInfo,
1644 incremental_snapshot_archive_info: Option<&IncrementalSnapshotArchiveInfo>,
1645) -> Result<()> {
1646 if incremental_snapshot_archive_info.is_none() {
1647 return Ok(());
1648 }
1649
1650 let incremental_snapshot_archive_info = incremental_snapshot_archive_info.unwrap();
1651
1652 (full_snapshot_archive_info.slot() == incremental_snapshot_archive_info.base_slot())
1653 .then_some(())
1654 .ok_or_else(|| {
1655 SnapshotError::MismatchedBaseSlot(
1656 full_snapshot_archive_info.slot(),
1657 incremental_snapshot_archive_info.base_slot(),
1658 )
1659 })
1660}
1661
1662pub fn purge_old_snapshot_archives(
1663 full_snapshot_archives_dir: impl AsRef<Path>,
1664 incremental_snapshot_archives_dir: impl AsRef<Path>,
1665 maximum_full_snapshot_archives_to_retain: NonZeroUsize,
1666 maximum_incremental_snapshot_archives_to_retain: NonZeroUsize,
1667) {
1668 info!(
1669 "Purging old full snapshot archives in {}, retaining up to {} full snapshots",
1670 full_snapshot_archives_dir.as_ref().display(),
1671 maximum_full_snapshot_archives_to_retain
1672 );
1673
1674 let mut full_snapshot_archives =
1675 snapshot_paths::full_snapshot_archives_iter(full_snapshot_archives_dir.as_ref())
1676 .collect::<Vec<_>>();
1677 full_snapshot_archives.sort_unstable();
1678 full_snapshot_archives.reverse();
1679
1680 let num_to_retain = full_snapshot_archives
1681 .len()
1682 .min(maximum_full_snapshot_archives_to_retain.get());
1683 trace!(
1684 "There are {} full snapshot archives, retaining {}",
1685 full_snapshot_archives.len(),
1686 num_to_retain,
1687 );
1688
1689 let (full_snapshot_archives_to_retain, full_snapshot_archives_to_remove) =
1690 if full_snapshot_archives.is_empty() {
1691 None
1692 } else {
1693 Some(full_snapshot_archives.split_at(num_to_retain))
1694 }
1695 .unwrap_or_default();
1696
1697 let retained_full_snapshot_slots = full_snapshot_archives_to_retain
1698 .iter()
1699 .map(|ai| ai.slot())
1700 .collect::<HashSet<_>>();
1701
1702 fn remove_archives<T: SnapshotArchiveInfoGetter>(archives: &[T]) {
1703 for path in archives.iter().map(|a| a.path()) {
1704 trace!("Removing snapshot archive: {}", path.display());
1705 let result = fs::remove_file(path);
1706 if let Err(err) = result {
1707 info!(
1708 "Failed to remove snapshot archive '{}': {err}",
1709 path.display()
1710 );
1711 }
1712 }
1713 }
1714 remove_archives(full_snapshot_archives_to_remove);
1715
1716 info!(
1717 "Purging old incremental snapshot archives in {}, retaining up to {} incremental snapshots",
1718 incremental_snapshot_archives_dir.as_ref().display(),
1719 maximum_incremental_snapshot_archives_to_retain
1720 );
1721 let mut incremental_snapshot_archives_by_base_slot = HashMap::<Slot, Vec<_>>::new();
1722 for incremental_snapshot_archive in
1723 incremental_snapshot_archives_iter(incremental_snapshot_archives_dir.as_ref())
1724 {
1725 incremental_snapshot_archives_by_base_slot
1726 .entry(incremental_snapshot_archive.base_slot())
1727 .or_default()
1728 .push(incremental_snapshot_archive)
1729 }
1730
1731 let highest_full_snapshot_slot = retained_full_snapshot_slots.iter().max().copied();
1732 for (base_slot, mut incremental_snapshot_archives) in incremental_snapshot_archives_by_base_slot
1733 {
1734 incremental_snapshot_archives.sort_unstable();
1735 let num_to_retain = if Some(base_slot) == highest_full_snapshot_slot {
1736 maximum_incremental_snapshot_archives_to_retain.get()
1737 } else {
1738 usize::from(retained_full_snapshot_slots.contains(&base_slot))
1739 };
1740 trace!(
1741 "There are {} incremental snapshot archives for base slot {}, removing {} of them",
1742 incremental_snapshot_archives.len(),
1743 base_slot,
1744 incremental_snapshot_archives
1745 .len()
1746 .saturating_sub(num_to_retain),
1747 );
1748
1749 incremental_snapshot_archives.truncate(
1750 incremental_snapshot_archives
1751 .len()
1752 .saturating_sub(num_to_retain),
1753 );
1754 remove_archives(&incremental_snapshot_archives);
1755 }
1756}
1757
1758pub fn verify_unpacked_snapshots_dir_and_version(
1759 unpacked_snapshots_dir_and_version: &UnpackedSnapshotsDirAndVersion,
1760) -> Result<(SnapshotVersion, BankSnapshotInfo)> {
1761 info!(
1762 "snapshot version: {}",
1763 unpacked_snapshots_dir_and_version.snapshot_version
1764 );
1765
1766 let snapshot_version = unpacked_snapshots_dir_and_version.snapshot_version;
1767 let mut bank_snapshots =
1768 get_bank_snapshots(&unpacked_snapshots_dir_and_version.unpacked_snapshots_dir);
1769 if bank_snapshots.len() > 1 {
1770 return Err(IoError::other(format!(
1771 "invalid snapshot format: only one snapshot allowed, but found {}",
1772 bank_snapshots.len(),
1773 ))
1774 .into());
1775 }
1776 let root_paths = bank_snapshots.pop().ok_or_else(|| {
1777 IoError::other(format!(
1778 "no snapshots found in snapshots directory '{}'",
1779 unpacked_snapshots_dir_and_version
1780 .unpacked_snapshots_dir
1781 .display(),
1782 ))
1783 })?;
1784 Ok((snapshot_version, root_paths))
1785}
1786
1787#[derive(Debug, Copy, Clone)]
1788pub enum VerifyBank {
1790 Deterministic,
1792 NonDeterministic,
1795}
1796
1797pub fn wipe_account_snapshot_dirs(account_run_paths: &[PathBuf]) {
1807 for account_run_path in account_run_paths {
1808 if let Some(parent) = account_run_path.parent() {
1809 move_and_async_delete_path_contents(parent.join(ACCOUNTS_SNAPSHOT_DIR));
1810 }
1811 }
1812}
1813
1814pub fn purge_all_bank_snapshots(bank_snapshots_dir: impl AsRef<Path>) {
1816 let bank_snapshots = get_bank_snapshots(&bank_snapshots_dir);
1817 purge_bank_snapshots(&bank_snapshots);
1818}
1819
1820pub fn purge_old_bank_snapshots(
1822 bank_snapshots_dir: impl AsRef<Path>,
1823 num_bank_snapshots_to_retain: usize,
1824) {
1825 let mut bank_snapshots = get_bank_snapshots(&bank_snapshots_dir);
1826
1827 bank_snapshots.sort_unstable();
1828 purge_bank_snapshots(
1829 bank_snapshots
1830 .iter()
1831 .rev()
1832 .skip(num_bank_snapshots_to_retain),
1833 );
1834}
1835
1836pub fn purge_old_bank_snapshots_at_startup(bank_snapshots_dir: impl AsRef<Path>) {
1838 purge_old_bank_snapshots(&bank_snapshots_dir, 1);
1839
1840 let highest_bank_snapshot = get_highest_bank_snapshot(&bank_snapshots_dir);
1841 if let Some(highest_bank_snapshot) = highest_bank_snapshot {
1842 debug!(
1843 "Retained bank snapshot for slot {}, and purged the rest.",
1844 highest_bank_snapshot.slot
1845 );
1846 }
1847}
1848
1849pub fn purge_bank_snapshots_older_than_slot(bank_snapshots_dir: impl AsRef<Path>, slot: Slot) {
1851 let mut bank_snapshots = get_bank_snapshots(&bank_snapshots_dir);
1852 bank_snapshots.retain(|bank_snapshot| bank_snapshot.slot < slot);
1853 purge_bank_snapshots(&bank_snapshots);
1854}
1855
1856fn purge_bank_snapshots<'a>(bank_snapshots: impl IntoIterator<Item = &'a BankSnapshotInfo>) {
1860 for snapshot_dir in bank_snapshots.into_iter().map(|s| &s.snapshot_dir) {
1861 if purge_bank_snapshot(snapshot_dir).is_err() {
1862 warn!("Failed to purge bank snapshot: {}", snapshot_dir.display());
1863 }
1864 }
1865}
1866
1867pub fn purge_bank_snapshot(bank_snapshot_dir: impl AsRef<Path>) -> Result<()> {
1869 const FN_ERR: &str = "failed to purge bank snapshot";
1870 let accounts_hardlinks_dir = bank_snapshot_dir
1875 .as_ref()
1876 .join(snapshot_paths::SNAPSHOT_ACCOUNTS_HARDLINKS);
1877 if accounts_hardlinks_dir.is_dir() {
1878 let read_dir = fs::read_dir(&accounts_hardlinks_dir).map_err(|err| {
1879 IoError::other(format!(
1880 "{FN_ERR}: failed to read accounts hardlinks dir '{}': {err}",
1881 accounts_hardlinks_dir.display(),
1882 ))
1883 })?;
1884 for entry in read_dir {
1885 let accounts_hardlink_dir = entry?.path();
1886 let accounts_hardlink_dir = fs::read_link(&accounts_hardlink_dir).map_err(|err| {
1887 IoError::other(format!(
1888 "{FN_ERR}: failed to read symlink '{}': {err}",
1889 accounts_hardlink_dir.display(),
1890 ))
1891 })?;
1892 move_and_async_delete_path(&accounts_hardlink_dir);
1893 }
1894 }
1895 fs::remove_dir_all(&bank_snapshot_dir).map_err(|err| {
1896 IoError::other(format!(
1897 "{FN_ERR}: failed to remove dir '{}': {err}",
1898 bank_snapshot_dir.as_ref().display(),
1899 ))
1900 })?;
1901 Ok(())
1902}
1903
1904pub fn should_take_full_snapshot(
1905 block_height: Slot,
1906 full_snapshot_archive_interval_slots: Slot,
1907) -> bool {
1908 block_height.is_multiple_of(full_snapshot_archive_interval_slots)
1909}
1910
1911pub fn should_take_incremental_snapshot(
1912 block_height: Slot,
1913 incremental_snapshot_archive_interval_slots: Slot,
1914 latest_full_snapshot_slot: Option<Slot>,
1915) -> bool {
1916 block_height.is_multiple_of(incremental_snapshot_archive_interval_slots)
1917 && latest_full_snapshot_slot.is_some()
1918}
1919
1920#[cfg(feature = "dev-context-only-utils")]
1925pub fn create_tmp_accounts_dir_for_tests() -> (TempDir, PathBuf) {
1926 let tmp_dir = tempfile::TempDir::new().unwrap();
1927 let account_dir = create_accounts_run_and_snapshot_dirs(&tmp_dir).unwrap().0;
1928 (tmp_dir, account_dir)
1929}
1930
1931#[cfg(test)]
1932mod tests {
1933 use {
1934 super::*,
1935 crate::serde_snapshot::{deserialize_wincode_from, serialize_into},
1936 agave_snapshots::{
1937 paths::{
1938 full_snapshot_archives_iter, get_highest_full_snapshot_archive_slot,
1939 get_highest_incremental_snapshot_archive_slot,
1940 },
1941 snapshot_config::{
1942 DEFAULT_MAX_FULL_SNAPSHOT_ARCHIVES_TO_RETAIN,
1943 DEFAULT_MAX_INCREMENTAL_SNAPSHOT_ARCHIVES_TO_RETAIN,
1944 },
1945 },
1946 assert_matches::assert_matches,
1947 solana_accounts_db::accounts_file::{AccountsFile, AccountsFileProvider},
1948 solana_hash::Hash,
1949 std::{convert::TryFrom, mem::size_of},
1950 tempfile::NamedTempFile,
1951 test_case::test_case,
1952 };
1953
1954 #[test]
1955 fn test_serialize_snapshot_data_file_under_limit() {
1956 let temp_dir = tempfile::TempDir::new().unwrap();
1957 let expected_consumed_size = size_of::<u32>() as u64;
1958 let consumed_size = serialize_snapshot_data_file_capped(
1959 &temp_dir.path().join("data-file"),
1960 expected_consumed_size,
1961 &IoSetupState::default(),
1962 |stream| {
1963 serialize_into(stream, &2323_u32)?;
1964 Ok(())
1965 },
1966 )
1967 .unwrap();
1968 assert_eq!(consumed_size, expected_consumed_size);
1969 }
1970
1971 #[test]
1972 fn test_serialize_snapshot_data_file_over_limit() {
1973 let temp_dir = tempfile::TempDir::new().unwrap();
1974 let expected_consumed_size = size_of::<u32>() as u64;
1975 let result = serialize_snapshot_data_file_capped(
1976 &temp_dir.path().join("data-file"),
1977 expected_consumed_size - 1,
1978 &IoSetupState::default(),
1979 |stream| {
1980 serialize_into(stream, &2323_u32)?;
1981 Ok(())
1982 },
1983 );
1984 assert_matches!(result, Err(SnapshotError::Io(ref message)) if message.to_string().contains("bytes would exceed limit of"));
1985 }
1986
1987 #[test]
1988 fn test_deserialize_snapshot_data_file_under_limit() {
1989 let expected_data = 2323_u32;
1990 let expected_consumed_size = size_of::<u32>() as u64;
1991
1992 let temp_dir = tempfile::TempDir::new().unwrap();
1993 serialize_snapshot_data_file_capped(
1994 &temp_dir.path().join("data-file"),
1995 expected_consumed_size,
1996 &IoSetupState::default(),
1997 |stream| {
1998 serialize_into(stream, &expected_data)?;
1999 Ok(())
2000 },
2001 )
2002 .unwrap();
2003
2004 let snapshot_root_paths = SnapshotRootPaths {
2005 full_snapshot_root_file_path: temp_dir.path().join("data-file"),
2006 incremental_snapshot_root_file_path: None,
2007 };
2008
2009 let actual_data = deserialize_snapshot_data_files_capped(
2010 &snapshot_root_paths,
2011 expected_consumed_size,
2012 |stream| {
2013 Ok(deserialize_wincode_from::<_, u32>(
2014 &mut *stream.full_snapshot_stream,
2015 )?)
2016 },
2017 )
2018 .unwrap();
2019 assert_eq!(actual_data, expected_data);
2020 }
2021
2022 #[test]
2023 fn test_deserialize_snapshot_data_file_over_limit() {
2024 let expected_data = 2323_u32;
2025 let expected_consumed_size = size_of::<u32>() as u64;
2026
2027 let temp_dir = tempfile::TempDir::new().unwrap();
2028 serialize_snapshot_data_file_capped(
2029 &temp_dir.path().join("data-file"),
2030 expected_consumed_size,
2031 &IoSetupState::default(),
2032 |stream| {
2033 serialize_into(stream, &expected_data)?;
2034 Ok(())
2035 },
2036 )
2037 .unwrap();
2038
2039 let snapshot_root_paths = SnapshotRootPaths {
2040 full_snapshot_root_file_path: temp_dir.path().join("data-file"),
2041 incremental_snapshot_root_file_path: None,
2042 };
2043
2044 let result = deserialize_snapshot_data_files_capped(
2045 &snapshot_root_paths,
2046 expected_consumed_size - 1,
2047 |stream| {
2048 Ok(deserialize_wincode_from::<_, u32>(
2049 &mut *stream.full_snapshot_stream,
2050 )?)
2051 },
2052 );
2053 assert_matches!(result, Err(SnapshotError::Io(ref message)) if message.to_string().starts_with("too large snapshot data file to deserialize"));
2054 }
2055
2056 #[test]
2057 fn test_deserialize_snapshot_data_file_extra_data() {
2058 let expected_data = 2323_u32;
2059 let expected_consumed_size = size_of::<u32>() as u64;
2060
2061 let temp_dir = tempfile::TempDir::new().unwrap();
2062 serialize_snapshot_data_file_capped(
2063 &temp_dir.path().join("data-file"),
2064 expected_consumed_size * 2,
2065 &IoSetupState::default(),
2066 |stream| {
2067 serialize_into(&mut *stream, &(expected_data, expected_data))?;
2070 Ok(())
2071 },
2072 )
2073 .unwrap();
2074
2075 let snapshot_root_paths = SnapshotRootPaths {
2076 full_snapshot_root_file_path: temp_dir.path().join("data-file"),
2077 incremental_snapshot_root_file_path: None,
2078 };
2079
2080 let result = deserialize_snapshot_data_files_capped(
2081 &snapshot_root_paths,
2082 expected_consumed_size * 2,
2083 |stream| {
2084 Ok(deserialize_wincode_from::<_, u32>(
2085 &mut *stream.full_snapshot_stream,
2086 )?)
2087 },
2088 );
2089 assert_matches!(result, Err(SnapshotError::Io(ref message)) if message.to_string().starts_with("invalid snapshot data file"));
2090 }
2091
2092 #[test]
2093 fn test_snapshot_version_from_file_under_limit() {
2094 let file_content = SnapshotVersion::default().as_str();
2095 let mut file = NamedTempFile::new().unwrap();
2096 file.write_all(file_content.as_bytes()).unwrap();
2097 let file_info = FileInfo::new_from_path(file.path()).unwrap();
2098 let version_from_file = snapshot_version_from_file(file_info).unwrap();
2099 assert_eq!(version_from_file, file_content);
2100 }
2101
2102 #[test]
2103 fn test_snapshot_version_from_file_over_limit() {
2104 let over_limit_size = usize::try_from(MAX_SNAPSHOT_VERSION_FILE_SIZE + 1).unwrap();
2105 let file_content = vec![7u8; over_limit_size];
2106 let mut file = NamedTempFile::new().unwrap();
2107 file.write_all(&file_content).unwrap();
2108 let file_info = FileInfo::new_from_path(file.path()).unwrap();
2109 assert_matches!(
2110 snapshot_version_from_file(file_info),
2111 Err(ref message) if message.to_string().starts_with("snapshot version file too large")
2112 );
2113 }
2114
2115 #[test]
2116 fn test_check_are_snapshots_compatible() {
2117 let slot1: Slot = 1234;
2118 let slot2: Slot = 5678;
2119 let slot3: Slot = 999_999;
2120
2121 let full_snapshot_archive_info = FullSnapshotArchiveInfo::new_from_path(PathBuf::from(
2122 format!("/dir/snapshot-{}-{}.tar.zst", slot1, Hash::new_unique()),
2123 ))
2124 .unwrap();
2125
2126 assert!(check_are_snapshots_compatible(&full_snapshot_archive_info, None,).is_ok());
2127
2128 let incremental_snapshot_archive_info =
2129 IncrementalSnapshotArchiveInfo::new_from_path(PathBuf::from(format!(
2130 "/dir/incremental-snapshot-{}-{}-{}.tar.zst",
2131 slot1,
2132 slot2,
2133 Hash::new_unique()
2134 )))
2135 .unwrap();
2136
2137 assert!(
2138 check_are_snapshots_compatible(
2139 &full_snapshot_archive_info,
2140 Some(&incremental_snapshot_archive_info)
2141 )
2142 .is_ok()
2143 );
2144
2145 let incremental_snapshot_archive_info =
2146 IncrementalSnapshotArchiveInfo::new_from_path(PathBuf::from(format!(
2147 "/dir/incremental-snapshot-{}-{}-{}.tar.zst",
2148 slot2,
2149 slot3,
2150 Hash::new_unique()
2151 )))
2152 .unwrap();
2153
2154 assert!(
2155 check_are_snapshots_compatible(
2156 &full_snapshot_archive_info,
2157 Some(&incremental_snapshot_archive_info)
2158 )
2159 .is_err()
2160 );
2161 }
2162
2163 fn common_create_bank_snapshot_files(
2165 bank_snapshots_dir: &Path,
2166 min_slot: Slot,
2167 max_slot: Slot,
2168 ) {
2169 for slot in min_slot..max_slot {
2170 let snapshot_dir = snapshot_paths::get_bank_snapshot_dir(bank_snapshots_dir, slot);
2171 fs::create_dir_all(&snapshot_dir).unwrap();
2172
2173 let snapshot_filename = snapshot_paths::get_snapshot_file_name(slot);
2174 let snapshot_path = snapshot_dir.join(snapshot_filename);
2175 fs::File::create(snapshot_path).unwrap();
2176
2177 let status_cache_file =
2178 snapshot_dir.join(snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME);
2179 fs::File::create(status_cache_file).unwrap();
2180
2181 let version_path = snapshot_dir.join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
2182 fs::write(version_path, SnapshotVersion::default().as_str().as_bytes()).unwrap();
2183 }
2184 }
2185
2186 #[test]
2187 fn test_get_bank_snapshots() {
2188 let temp_snapshots_dir = tempfile::TempDir::new().unwrap();
2189 let min_slot = 10;
2190 let max_slot = 20;
2191 common_create_bank_snapshot_files(temp_snapshots_dir.path(), min_slot, max_slot);
2192
2193 let bank_snapshots = get_bank_snapshots(temp_snapshots_dir.path());
2194 assert_eq!(bank_snapshots.len() as Slot, max_slot - min_slot);
2195 }
2196
2197 #[test]
2198 fn test_get_highest_bank_snapshot() {
2199 let temp_snapshots_dir = tempfile::TempDir::new().unwrap();
2200 let min_slot = 99;
2201 let max_slot = 123;
2202 common_create_bank_snapshot_files(temp_snapshots_dir.path(), min_slot, max_slot);
2203
2204 let highest_bank_snapshot = get_highest_bank_snapshot(temp_snapshots_dir.path());
2205 assert!(highest_bank_snapshot.is_some());
2206 assert_eq!(highest_bank_snapshot.unwrap().slot, max_slot - 1);
2207 }
2208
2209 fn common_create_snapshot_archive_files(
2215 full_snapshot_archives_dir: &Path,
2216 incremental_snapshot_archives_dir: &Path,
2217 min_full_snapshot_slot: Slot,
2218 max_full_snapshot_slot: Slot,
2219 min_incremental_snapshot_slot: Slot,
2220 max_incremental_snapshot_slot: Slot,
2221 ) {
2222 fs::create_dir_all(full_snapshot_archives_dir).unwrap();
2223 fs::create_dir_all(incremental_snapshot_archives_dir).unwrap();
2224 for full_snapshot_slot in min_full_snapshot_slot..max_full_snapshot_slot {
2225 for incremental_snapshot_slot in
2226 min_incremental_snapshot_slot..max_incremental_snapshot_slot
2227 {
2228 let snapshot_filename = format!(
2229 "incremental-snapshot-{}-{}-{}.tar.zst",
2230 full_snapshot_slot,
2231 incremental_snapshot_slot,
2232 Hash::default()
2233 );
2234 let snapshot_filepath = incremental_snapshot_archives_dir.join(snapshot_filename);
2235 fs::File::create(snapshot_filepath).unwrap();
2236 }
2237
2238 let snapshot_filename = format!(
2239 "snapshot-{}-{}.tar.zst",
2240 full_snapshot_slot,
2241 Hash::default()
2242 );
2243 let snapshot_filepath = full_snapshot_archives_dir.join(snapshot_filename);
2244 fs::File::create(snapshot_filepath).unwrap();
2245
2246 let bad_filename = format!(
2248 "incremental-snapshot-{}-{}-bad!hash.tar.zst",
2249 full_snapshot_slot,
2250 max_incremental_snapshot_slot + 1,
2251 );
2252 let bad_filepath = incremental_snapshot_archives_dir.join(bad_filename);
2253 fs::File::create(bad_filepath).unwrap();
2254 }
2255
2256 let bad_filename = format!("snapshot-{}-bad!hash.tar.zst", max_full_snapshot_slot + 1);
2259 let bad_filepath = full_snapshot_archives_dir.join(bad_filename);
2260 fs::File::create(bad_filepath).unwrap();
2261 }
2262
2263 #[test]
2264 fn test_get_full_snapshot_archives() {
2265 let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2266 let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2267 let min_slot = 123;
2268 let max_slot = 456;
2269 common_create_snapshot_archive_files(
2270 full_snapshot_archives_dir.path(),
2271 incremental_snapshot_archives_dir.path(),
2272 min_slot,
2273 max_slot,
2274 0,
2275 0,
2276 );
2277
2278 let snapshot_archives =
2279 full_snapshot_archives_iter(full_snapshot_archives_dir.path()).collect::<Vec<_>>();
2280 assert_eq!(snapshot_archives.len() as Slot, max_slot - min_slot);
2281 }
2282
2283 #[test]
2284 fn test_get_full_snapshot_archives_remote() {
2285 let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2286 let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2287 let min_slot = 123;
2288 let max_slot = 456;
2289 common_create_snapshot_archive_files(
2290 &full_snapshot_archives_dir
2291 .path()
2292 .join(snapshot_paths::SNAPSHOT_ARCHIVE_DOWNLOAD_DIR),
2293 &incremental_snapshot_archives_dir
2294 .path()
2295 .join(snapshot_paths::SNAPSHOT_ARCHIVE_DOWNLOAD_DIR),
2296 min_slot,
2297 max_slot,
2298 0,
2299 0,
2300 );
2301
2302 let snapshot_archives =
2303 full_snapshot_archives_iter(full_snapshot_archives_dir.path()).collect::<Vec<_>>();
2304 assert_eq!(snapshot_archives.len() as Slot, max_slot - min_slot);
2305 assert!(snapshot_archives.iter().all(|info| info.is_remote()));
2306 }
2307
2308 #[test]
2309 fn test_get_incremental_snapshot_archives() {
2310 let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2311 let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2312 let min_full_snapshot_slot = 12;
2313 let max_full_snapshot_slot = 23;
2314 let min_incremental_snapshot_slot = 34;
2315 let max_incremental_snapshot_slot = 45;
2316 common_create_snapshot_archive_files(
2317 full_snapshot_archives_dir.path(),
2318 incremental_snapshot_archives_dir.path(),
2319 min_full_snapshot_slot,
2320 max_full_snapshot_slot,
2321 min_incremental_snapshot_slot,
2322 max_incremental_snapshot_slot,
2323 );
2324
2325 let incremental_snapshot_archives =
2326 incremental_snapshot_archives_iter(incremental_snapshot_archives_dir.path())
2327 .collect::<Vec<_>>();
2328 assert_eq!(
2329 incremental_snapshot_archives.len() as Slot,
2330 (max_full_snapshot_slot - min_full_snapshot_slot)
2331 * (max_incremental_snapshot_slot - min_incremental_snapshot_slot)
2332 );
2333 }
2334
2335 #[test]
2336 fn test_get_incremental_snapshot_archives_remote() {
2337 let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2338 let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2339 let min_full_snapshot_slot = 12;
2340 let max_full_snapshot_slot = 23;
2341 let min_incremental_snapshot_slot = 34;
2342 let max_incremental_snapshot_slot = 45;
2343 common_create_snapshot_archive_files(
2344 &full_snapshot_archives_dir
2345 .path()
2346 .join(snapshot_paths::SNAPSHOT_ARCHIVE_DOWNLOAD_DIR),
2347 &incremental_snapshot_archives_dir
2348 .path()
2349 .join(snapshot_paths::SNAPSHOT_ARCHIVE_DOWNLOAD_DIR),
2350 min_full_snapshot_slot,
2351 max_full_snapshot_slot,
2352 min_incremental_snapshot_slot,
2353 max_incremental_snapshot_slot,
2354 );
2355
2356 let incremental_snapshot_archives =
2357 incremental_snapshot_archives_iter(incremental_snapshot_archives_dir.path())
2358 .collect::<Vec<_>>();
2359 assert_eq!(
2360 incremental_snapshot_archives.len() as Slot,
2361 (max_full_snapshot_slot - min_full_snapshot_slot)
2362 * (max_incremental_snapshot_slot - min_incremental_snapshot_slot)
2363 );
2364 assert!(
2365 incremental_snapshot_archives
2366 .iter()
2367 .all(|info| info.is_remote())
2368 );
2369 }
2370
2371 #[test]
2372 fn test_get_highest_full_snapshot_archive_slot() {
2373 let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2374 let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2375 let min_slot = 123;
2376 let max_slot = 456;
2377 common_create_snapshot_archive_files(
2378 full_snapshot_archives_dir.path(),
2379 incremental_snapshot_archives_dir.path(),
2380 min_slot,
2381 max_slot,
2382 0,
2383 0,
2384 );
2385
2386 assert_eq!(
2387 get_highest_full_snapshot_archive_slot(full_snapshot_archives_dir.path()),
2388 Some(max_slot - 1)
2389 );
2390 }
2391
2392 #[test]
2393 fn test_get_highest_incremental_snapshot_slot() {
2394 let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2395 let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2396 let min_full_snapshot_slot = 12;
2397 let max_full_snapshot_slot = 23;
2398 let min_incremental_snapshot_slot = 34;
2399 let max_incremental_snapshot_slot = 45;
2400 common_create_snapshot_archive_files(
2401 full_snapshot_archives_dir.path(),
2402 incremental_snapshot_archives_dir.path(),
2403 min_full_snapshot_slot,
2404 max_full_snapshot_slot,
2405 min_incremental_snapshot_slot,
2406 max_incremental_snapshot_slot,
2407 );
2408
2409 for full_snapshot_slot in min_full_snapshot_slot..max_full_snapshot_slot {
2410 assert_eq!(
2411 get_highest_incremental_snapshot_archive_slot(
2412 incremental_snapshot_archives_dir.path(),
2413 full_snapshot_slot
2414 ),
2415 Some(max_incremental_snapshot_slot - 1)
2416 );
2417 }
2418
2419 assert_eq!(
2420 get_highest_incremental_snapshot_archive_slot(
2421 incremental_snapshot_archives_dir.path(),
2422 max_full_snapshot_slot
2423 ),
2424 None
2425 );
2426 }
2427
2428 fn common_test_purge_old_snapshot_archives(
2429 snapshot_names: &[&String],
2430 maximum_full_snapshot_archives_to_retain: NonZeroUsize,
2431 maximum_incremental_snapshot_archives_to_retain: NonZeroUsize,
2432 expected_snapshots: &[&String],
2433 ) {
2434 let temp_snap_dir = tempfile::TempDir::new().unwrap();
2435
2436 for snap_name in snapshot_names {
2437 let snap_path = temp_snap_dir.path().join(snap_name);
2438 let mut _snap_file = fs::File::create(snap_path);
2439 }
2440 purge_old_snapshot_archives(
2441 temp_snap_dir.path(),
2442 temp_snap_dir.path(),
2443 maximum_full_snapshot_archives_to_retain,
2444 maximum_incremental_snapshot_archives_to_retain,
2445 );
2446
2447 let mut retained_snaps = HashSet::new();
2448 for entry in fs::read_dir(temp_snap_dir.path()).unwrap() {
2449 let entry_path_buf = entry.unwrap().path();
2450 let entry_path = entry_path_buf.as_path();
2451 let snapshot_name = entry_path
2452 .file_name()
2453 .unwrap()
2454 .to_str()
2455 .unwrap()
2456 .to_string();
2457 retained_snaps.insert(snapshot_name);
2458 }
2459
2460 for snap_name in expected_snapshots {
2461 assert!(
2462 retained_snaps.contains(snap_name.as_str()),
2463 "{snap_name} not found"
2464 );
2465 }
2466 assert_eq!(retained_snaps.len(), expected_snapshots.len());
2467 }
2468
2469 #[test]
2470 fn test_purge_old_full_snapshot_archives() {
2471 let snap1_name = format!("snapshot-1-{}.tar.zst", Hash::default());
2472 let snap2_name = format!("snapshot-3-{}.tar.zst", Hash::default());
2473 let snap3_name = format!("snapshot-50-{}.tar.zst", Hash::default());
2474 let snapshot_names = vec![&snap1_name, &snap2_name, &snap3_name];
2475
2476 let expected_snapshots = vec![&snap3_name];
2478 common_test_purge_old_snapshot_archives(
2479 &snapshot_names,
2480 NonZeroUsize::new(1).unwrap(),
2481 DEFAULT_MAX_INCREMENTAL_SNAPSHOT_ARCHIVES_TO_RETAIN,
2482 &expected_snapshots,
2483 );
2484
2485 let expected_snapshots = vec![&snap2_name, &snap3_name];
2487 common_test_purge_old_snapshot_archives(
2488 &snapshot_names,
2489 NonZeroUsize::new(2).unwrap(),
2490 DEFAULT_MAX_INCREMENTAL_SNAPSHOT_ARCHIVES_TO_RETAIN,
2491 &expected_snapshots,
2492 );
2493
2494 let expected_snapshots = vec![&snap1_name, &snap2_name, &snap3_name];
2496 common_test_purge_old_snapshot_archives(
2497 &snapshot_names,
2498 NonZeroUsize::new(3).unwrap(),
2499 DEFAULT_MAX_INCREMENTAL_SNAPSHOT_ARCHIVES_TO_RETAIN,
2500 &expected_snapshots,
2501 );
2502 }
2503
2504 #[test]
2508 fn test_purge_old_full_snapshot_archives_in_the_loop() {
2509 let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2510 let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2511 let maximum_snapshots_to_retain = NonZeroUsize::new(5).unwrap();
2512 let starting_slot: Slot = 42;
2513
2514 for slot in (starting_slot..).take(100) {
2515 let full_snapshot_archive_file_name =
2516 format!("snapshot-{}-{}.tar.zst", slot, Hash::default());
2517 let full_snapshot_archive_path = full_snapshot_archives_dir
2518 .as_ref()
2519 .join(full_snapshot_archive_file_name);
2520 fs::File::create(full_snapshot_archive_path).unwrap();
2521
2522 if slot < starting_slot + maximum_snapshots_to_retain.get() as Slot {
2524 continue;
2525 }
2526
2527 if slot % (maximum_snapshots_to_retain.get() as Slot * 2) != 0 {
2529 continue;
2530 }
2531
2532 purge_old_snapshot_archives(
2533 &full_snapshot_archives_dir,
2534 &incremental_snapshot_archives_dir,
2535 maximum_snapshots_to_retain,
2536 NonZeroUsize::new(usize::MAX).unwrap(),
2537 );
2538 let mut full_snapshot_archives =
2539 full_snapshot_archives_iter(full_snapshot_archives_dir.path()).collect::<Vec<_>>();
2540 full_snapshot_archives.sort_unstable();
2541 assert_eq!(
2542 full_snapshot_archives.len(),
2543 maximum_snapshots_to_retain.get()
2544 );
2545 assert_eq!(full_snapshot_archives.last().unwrap().slot(), slot);
2546 for (i, full_snapshot_archive) in full_snapshot_archives.iter().rev().enumerate() {
2547 assert_eq!(full_snapshot_archive.slot(), slot - i as Slot);
2548 }
2549 }
2550 }
2551
2552 #[test]
2553 fn test_purge_old_incremental_snapshot_archives() {
2554 let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2555 let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2556 let starting_slot = 100_000;
2557
2558 let maximum_incremental_snapshot_archives_to_retain =
2559 DEFAULT_MAX_INCREMENTAL_SNAPSHOT_ARCHIVES_TO_RETAIN;
2560 let maximum_full_snapshot_archives_to_retain = DEFAULT_MAX_FULL_SNAPSHOT_ARCHIVES_TO_RETAIN;
2561
2562 let incremental_snapshot_interval = 100;
2563 let num_incremental_snapshots_per_full_snapshot =
2564 maximum_incremental_snapshot_archives_to_retain.get() * 2;
2565 let full_snapshot_interval =
2566 incremental_snapshot_interval * num_incremental_snapshots_per_full_snapshot;
2567
2568 let mut snapshot_filenames = vec![];
2569 (starting_slot..)
2570 .step_by(full_snapshot_interval)
2571 .take(
2572 maximum_full_snapshot_archives_to_retain
2573 .checked_mul(NonZeroUsize::new(2).unwrap())
2574 .unwrap()
2575 .get(),
2576 )
2577 .for_each(|full_snapshot_slot| {
2578 let snapshot_filename = format!(
2579 "snapshot-{}-{}.tar.zst",
2580 full_snapshot_slot,
2581 Hash::default()
2582 );
2583 let snapshot_path = full_snapshot_archives_dir.path().join(&snapshot_filename);
2584 fs::File::create(snapshot_path).unwrap();
2585 snapshot_filenames.push(snapshot_filename);
2586
2587 (full_snapshot_slot..)
2588 .step_by(incremental_snapshot_interval)
2589 .take(num_incremental_snapshots_per_full_snapshot)
2590 .skip(1)
2591 .for_each(|incremental_snapshot_slot| {
2592 let snapshot_filename = format!(
2593 "incremental-snapshot-{}-{}-{}.tar.zst",
2594 full_snapshot_slot,
2595 incremental_snapshot_slot,
2596 Hash::default()
2597 );
2598 let snapshot_path = incremental_snapshot_archives_dir
2599 .path()
2600 .join(&snapshot_filename);
2601 fs::File::create(snapshot_path).unwrap();
2602 snapshot_filenames.push(snapshot_filename);
2603 });
2604 });
2605
2606 purge_old_snapshot_archives(
2607 full_snapshot_archives_dir.path(),
2608 incremental_snapshot_archives_dir.path(),
2609 maximum_full_snapshot_archives_to_retain,
2610 maximum_incremental_snapshot_archives_to_retain,
2611 );
2612
2613 let mut remaining_full_snapshot_archives =
2615 full_snapshot_archives_iter(full_snapshot_archives_dir.path()).collect::<Vec<_>>();
2616 assert_eq!(
2617 remaining_full_snapshot_archives.len(),
2618 maximum_full_snapshot_archives_to_retain.get(),
2619 );
2620 remaining_full_snapshot_archives.sort_unstable();
2621 let latest_full_snapshot_archive_slot =
2622 remaining_full_snapshot_archives.last().unwrap().slot();
2623
2624 let mut remaining_incremental_snapshot_archives =
2629 incremental_snapshot_archives_iter(incremental_snapshot_archives_dir.path())
2630 .collect::<Vec<_>>();
2631 assert_eq!(
2632 remaining_incremental_snapshot_archives.len(),
2633 maximum_incremental_snapshot_archives_to_retain
2634 .get()
2635 .saturating_add(
2636 maximum_full_snapshot_archives_to_retain
2637 .get()
2638 .saturating_sub(1)
2639 )
2640 );
2641 remaining_incremental_snapshot_archives.sort_unstable();
2642 remaining_incremental_snapshot_archives.reverse();
2643
2644 for i in (1..maximum_full_snapshot_archives_to_retain.get()).rev() {
2646 let incremental_snapshot_archive =
2647 remaining_incremental_snapshot_archives.pop().unwrap();
2648
2649 let expected_base_slot =
2650 latest_full_snapshot_archive_slot - (i * full_snapshot_interval) as u64;
2651 assert_eq!(incremental_snapshot_archive.base_slot(), expected_base_slot);
2652 let expected_slot = expected_base_slot
2653 + (full_snapshot_interval - incremental_snapshot_interval) as u64;
2654 assert_eq!(incremental_snapshot_archive.slot(), expected_slot);
2655 }
2656
2657 for incremental_snapshot_archive in &remaining_incremental_snapshot_archives {
2659 assert_eq!(
2660 incremental_snapshot_archive.base_slot(),
2661 latest_full_snapshot_archive_slot
2662 );
2663 }
2664
2665 let expected_remaining_incremental_snapshot_archive_slots =
2667 (latest_full_snapshot_archive_slot..)
2668 .step_by(incremental_snapshot_interval)
2669 .take(num_incremental_snapshots_per_full_snapshot)
2670 .skip(
2671 num_incremental_snapshots_per_full_snapshot
2672 - maximum_incremental_snapshot_archives_to_retain.get(),
2673 )
2674 .collect::<HashSet<_>>();
2675
2676 let actual_remaining_incremental_snapshot_archive_slots =
2677 remaining_incremental_snapshot_archives
2678 .iter()
2679 .map(|snapshot| snapshot.slot())
2680 .collect::<HashSet<_>>();
2681 assert_eq!(
2682 actual_remaining_incremental_snapshot_archive_slots,
2683 expected_remaining_incremental_snapshot_archive_slots
2684 );
2685 }
2686
2687 #[test]
2688 fn test_purge_all_incremental_snapshot_archives_when_no_full_snapshot_archives() {
2689 let full_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2690 let incremental_snapshot_archives_dir = tempfile::TempDir::new().unwrap();
2691
2692 for snapshot_filenames in [
2693 format!("incremental-snapshot-100-120-{}.tar.zst", Hash::default()),
2694 format!("incremental-snapshot-100-140-{}.tar.zst", Hash::default()),
2695 format!("incremental-snapshot-100-160-{}.tar.zst", Hash::default()),
2696 format!("incremental-snapshot-100-180-{}.tar.zst", Hash::default()),
2697 format!("incremental-snapshot-200-220-{}.tar.zst", Hash::default()),
2698 format!("incremental-snapshot-200-240-{}.tar.zst", Hash::default()),
2699 format!("incremental-snapshot-200-260-{}.tar.zst", Hash::default()),
2700 format!("incremental-snapshot-200-280-{}.tar.zst", Hash::default()),
2701 ] {
2702 let snapshot_path = incremental_snapshot_archives_dir
2703 .path()
2704 .join(snapshot_filenames);
2705 fs::File::create(snapshot_path).unwrap();
2706 }
2707
2708 purge_old_snapshot_archives(
2709 full_snapshot_archives_dir.path(),
2710 incremental_snapshot_archives_dir.path(),
2711 NonZeroUsize::new(usize::MAX).unwrap(),
2712 NonZeroUsize::new(usize::MAX).unwrap(),
2713 );
2714
2715 let remaining_incremental_snapshot_archives =
2716 incremental_snapshot_archives_iter(incremental_snapshot_archives_dir.path())
2717 .collect::<Vec<_>>();
2718 assert!(remaining_incremental_snapshot_archives.is_empty());
2719 }
2720
2721 #[test]
2722 fn test_get_snapshot_file_kind() {
2723 assert_eq!(None, get_snapshot_file_kind("file.txt"));
2724 assert_eq!(
2725 Some(SnapshotFileKind::Version),
2726 get_snapshot_file_kind(snapshot_paths::SNAPSHOT_VERSION_FILENAME)
2727 );
2728 assert_eq!(
2729 Some(SnapshotFileKind::BankFields),
2730 get_snapshot_file_kind("1234")
2731 );
2732 assert_eq!(
2733 Some(SnapshotFileKind::Storage),
2734 get_snapshot_file_kind("1000.999")
2735 );
2736 }
2737
2738 #[test_case(0)]
2739 #[test_case(1)]
2740 #[test_case(10)]
2741 fn test_serialize_deserialize_account_storage_entries(num_storages: u64) {
2742 let temp_dir = tempfile::tempdir().unwrap();
2743 let bank_snapshot_dir = temp_dir.path();
2744 let storage_dir = tempfile::tempdir().unwrap();
2745 let snapshot_slot = num_storages + 1 as Slot;
2746
2747 let mut snapshot_storages = Vec::new();
2749 for i in 0..num_storages {
2750 let storage = Arc::new(AccountStorageEntry::new(
2751 storage_dir.path(),
2752 i, i as u32, 1024,
2755 AccountsFileProvider::AppendVec,
2756 ));
2757 snapshot_storages.push(storage);
2758 }
2759
2760 write_obsolete_accounts_to_snapshot(
2762 bank_snapshot_dir,
2763 &snapshot_storages,
2764 snapshot_slot,
2765 &IoSetupState::default(),
2766 )
2767 .unwrap();
2768
2769 let mut deserialized_accounts =
2771 deserialize_obsolete_accounts(bank_snapshot_dir, MAX_OBSOLETE_ACCOUNTS_FILE_SIZE)
2772 .unwrap()
2773 .into_hashmap();
2774
2775 for storage in &snapshot_storages {
2777 let obsolete_accounts = deserialized_accounts.remove(&storage.slot()).unwrap();
2778 assert!(obsolete_accounts.into_tuple().2 == 0);
2779 }
2780 }
2781
2782 #[test]
2783 #[should_panic(expected = "bytes would exceed limit of 100")]
2784 fn test_serialize_obsolete_accounts_too_large_file() {
2785 let temp_dir = tempfile::tempdir().unwrap();
2786 let bank_snapshot_dir = temp_dir.path();
2787 let storage_dir = tempfile::tempdir().unwrap();
2788 let num_storages = 10;
2789 let snapshot_slot = num_storages + 1 as Slot;
2790
2791 let mut snapshot_storages = Vec::new();
2793 for i in 0..num_storages {
2794 let storage = Arc::new(AccountStorageEntry::new(
2795 storage_dir.path(),
2796 i, i as u32, 1024,
2799 AccountsFileProvider::AppendVec,
2800 ));
2801 snapshot_storages.push(storage);
2802 }
2803
2804 let obsolete_accounts =
2806 SerdeObsoleteAccountsMap::new_from_storages(&snapshot_storages, snapshot_slot);
2807
2808 serialize_obsolete_accounts(
2810 bank_snapshot_dir,
2811 &obsolete_accounts,
2812 100,
2813 &IoSetupState::default(),
2814 )
2815 .unwrap();
2816 }
2817
2818 #[test]
2819 #[should_panic(expected = "too large obsolete accounts file to deserialize")]
2820 fn test_deserialize_obsolete_accounts_too_large_file() {
2821 let temp_dir = tempfile::tempdir().unwrap();
2822 let bank_snapshot_dir = temp_dir.path();
2823 let storage_dir = tempfile::tempdir().unwrap();
2824 let num_storages = 10;
2825 let snapshot_slot = num_storages + 1 as Slot;
2826
2827 let mut snapshot_storages = Vec::new();
2829 for i in 0..num_storages {
2830 let storage = Arc::new(AccountStorageEntry::new(
2831 storage_dir.path(),
2832 i, i as u32, 1024,
2835 AccountsFileProvider::AppendVec,
2836 ));
2837 snapshot_storages.push(storage);
2838 }
2839
2840 write_obsolete_accounts_to_snapshot(
2842 bank_snapshot_dir,
2843 &snapshot_storages,
2844 snapshot_slot,
2845 &IoSetupState::default(),
2846 )
2847 .unwrap();
2848
2849 deserialize_obsolete_accounts(bank_snapshot_dir, 100).unwrap();
2852 }
2853
2854 #[test]
2855 fn test_is_bank_snapshot_complete() {
2856 let temp_dir = TempDir::new().unwrap();
2857 let slot = 123;
2858 let bank_snapshot_dir = temp_dir.as_ref().join(slot.to_string());
2859 fs::create_dir(&bank_snapshot_dir).unwrap();
2860
2861 let version_path = bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_VERSION_FILENAME);
2862 let serialized_bank_path = bank_snapshot_dir.join(slot.to_string());
2863 let status_cache_path =
2864 bank_snapshot_dir.join(snapshot_paths::SNAPSHOT_STATUS_CACHE_FILENAME);
2865
2866 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2868
2869 let too_large = format!(
2871 "{:v>width$}",
2872 "hi",
2873 width = (MAX_SNAPSHOT_VERSION_FILE_SIZE + 1) as usize,
2874 );
2875 fs::write(&version_path, too_large).unwrap();
2876 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2877
2878 fs::remove_file(&version_path).unwrap();
2880 let bad_version = String::from("v0.0.0");
2881 fs::write(&version_path, bad_version).unwrap();
2882 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2883
2884 fs::remove_file(&version_path).unwrap();
2886 fs::File::create_new(&version_path).unwrap();
2887 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2888
2889 fs::remove_file(&version_path).unwrap();
2891 fs::write(&version_path, SnapshotVersion::default().as_str()).unwrap();
2892
2893 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2895
2896 fs::File::create_new(&serialized_bank_path).unwrap();
2898 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2899
2900 fs::remove_file(&serialized_bank_path).unwrap();
2902 fs::write(&serialized_bank_path, "serialized bank").unwrap();
2903
2904 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2906
2907 fs::File::create_new(&status_cache_path).unwrap();
2909 assert!(!is_bank_snapshot_complete(&bank_snapshot_dir));
2910
2911 fs::remove_file(&status_cache_path).unwrap();
2913 fs::write(&status_cache_path, "status cache").unwrap();
2914
2915 assert!(is_bank_snapshot_complete(bank_snapshot_dir));
2917 }
2918
2919 #[test]
2920 fn test_prune_stale_storages() {
2921 let account_path = tempfile::TempDir::new().unwrap();
2922 let keep_a = account_path.path().join(AccountsFile::file_name(100, 1));
2924 let keep_b = account_path.path().join(AccountsFile::file_name(200, 2));
2925 let stale = account_path.path().join(AccountsFile::file_name(300, 3));
2927 let untouched = account_path.path().join("something_else.txt");
2929 for path in [&keep_a, &keep_b, &stale, &untouched] {
2930 fs::write(path, b"x").unwrap();
2931 }
2932
2933 let storages_list = StoragesList::from_items(vec![
2934 StorageListItem { slot: 100, id: 1 },
2935 StorageListItem { slot: 200, id: 2 },
2936 ]);
2937 prune_stale_storages(
2938 std::slice::from_ref(&account_path.path().to_path_buf()),
2939 storages_list,
2940 )
2941 .unwrap();
2942
2943 assert!(keep_a.exists(), "expected storage file was deleted");
2944 assert!(keep_b.exists(), "expected storage file was deleted");
2945 assert!(!stale.exists(), "stale storage file was not removed");
2946 assert!(untouched.exists(), "non-storage file was wrongly removed");
2947 }
2948}