1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
//! Verifying every table file against its checksums, read from storage.
use std::num::NonZeroUsize;
use std::path::Path;
use std::sync::Arc;
use tessarikv_constants::DEFAULT_BLOCK_BYTES;
use tokio::task::{JoinError, JoinSet};
use super::Db;
use super::block_cache::BlockCache;
use super::flush_job::table_path;
use super::table_blocks::open_table;
use super::tables::format_error;
use crate::storage::Storage;
use crate::{Error, Result};
impl Db {
/// Reads every table file of the database from storage and verifies it whole: its footer,
/// index, filter and properties, its range deletes and every data block, each against the
/// checksum it was written with (and its authentication tag, when encrypted). The block cache
/// is neither consulted nor filled, so a block whose good copy is cached is still read from
/// the device. Files are verified concurrently, each on a blocking thread. A file a
/// compaction removes during the sweep is passed over: its entries are in newer files, which
/// were verified as they were written. In memory there are no table files and it returns at
/// once.
///
/// # Errors
///
/// [`Error::Corruption`] naming the first damaged file found, the storage's error for a
/// failed read, and [`Error::ReadFailed`] when a reading thread stopped without an answer.
pub async fn verify_checksums(&self) -> Result<()> {
let Some(installer) = &self.installer else {
return Ok(());
};
let ids: Vec<u64> = self
.view
.load()
.all_files()
.iter()
.map(|file| file.meta.id)
.collect();
let at_once = std::thread::available_parallelism().map_or(1, NonZeroUsize::get);
let mut running = JoinSet::new();
for id in ids {
if running.len() >= at_once {
joined(running.join_next().await)?;
}
let storage = Arc::clone(&installer.storage);
let path = table_path(&installer.dir, id);
// Holds nothing: every block is read from storage.
let cache = Arc::new(
BlockCache::new(0, DEFAULT_BLOCK_BYTES)
.with_sealing(installer.cache.shared_sealing()),
);
running.spawn_blocking(move || verify_table(storage.as_ref(), &path, id, &cache));
}
while let Some(done) = running.join_next().await {
joined(Some(done))?;
}
Ok(())
}
}
/// The answer of a finished verification; a thread that died gave none.
fn joined(done: Option<std::result::Result<Result<()>, JoinError>>) -> Result<()> {
match done {
Some(Ok(verified)) => verified,
Some(Err(_)) => Err(Error::ReadFailed),
None => Ok(()),
}
}
/// Opens table file `id` at `path` through `cache`, which keeps nothing, and reads it whole.
/// Runs on a blocking thread.
fn verify_table(
storage: &dyn Storage,
path: &Path,
id: u64,
cache: &Arc<BlockCache>,
) -> Result<()> {
let (table, blocks) = match open_table(storage, path, id, cache, None) {
Ok(opened) => opened,
// Removed by a compaction since the sweep began.
Err(Error::NotFound { .. }) => return Ok(()),
Err(error) => return Err(error),
};
table
.range_deletes()
.map_err(|error| format_error(&error, Some(path)))?;
let mut walk = table.iter();
walk.seek_to_first();
blocks.complete_blocking(&mut walk)?;
while walk.valid() {
walk.next();
blocks.complete_blocking(&mut walk)?;
}
walk.status()
.map_err(|error| format_error(&error, Some(path)))
}