use crate::{
MAX_SEQNO, SeqNo, SharedSequenceNumberGenerator,
comparator::SharedComparator,
fs::{Fs, SyncMode},
memtable::Memtable,
tree::sealed::SealedMemtables,
version::{Version, VersionId, edit_log, persist_version},
};
fn remove_if_present(fs: &dyn Fs, path: &Path) -> crate::Result<()> {
match fs.remove_file(path) {
Ok(()) => Ok(()),
Err(e) if e.kind() == crate::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e.into()),
}
}
fn sync_new_table_directories(
prior: &Version,
next: &Version,
tree_path: &Path,
manifest_fs: &dyn Fs,
sync_mode: SyncMode,
) -> crate::Result<()> {
let known: crate::HashSet<crate::TableId> = prior.iter_tables().map(crate::Table::id).collect();
let mut folders: Vec<(&Arc<dyn Fs>, &Path, bool)> = Vec::new();
for table in next.iter_tables() {
if known.contains(&table.id()) {
continue;
}
let Some(folder) = table.path.parent() else {
continue;
};
let needs_sync = !table.entry_durable();
if let Some(seen) = folders
.iter_mut()
.find(|(fs, at, _)| Arc::ptr_eq(fs, &table.fs) && *at == folder)
{
seen.2 |= needs_sync;
} else {
folders.push((&table.fs, folder, needs_sync));
}
}
for (fs, folder, needs_sync) in folders {
let flushes_device =
sync_mode == SyncMode::Barrier && !same_device(&**fs, folder, manifest_fs, tree_path);
if flushes_device {
crate::file::fsync_directory(folder, &**fs, SyncMode::Full)?;
} else if needs_sync {
crate::file::fsync_directory(folder, &**fs, sync_mode)?;
}
}
Ok(())
}
fn same_device(fs: &dyn Fs, path: &Path, other_fs: &dyn Fs, other: &Path) -> bool {
matches!(
(fs.volume_id(path), other_fs.volume_id(other)),
(Some(a), Some(b)) if a == b
)
}
fn manifest_sync_mode(
prior: &Version,
next: &Version,
tree_path: &Path,
manifest_fs: &dyn Fs,
sync_mode: SyncMode,
) -> SyncMode {
if sync_mode != SyncMode::Barrier {
return sync_mode;
}
let elsewhere = |fs: &dyn Fs, path: &Path| {
path.parent()
.is_none_or(|folder| !same_device(fs, folder, manifest_fs, tree_path))
};
let kept: crate::HashMap<crate::TableId, Option<&crate::UserKey>> = next
.iter_tables()
.map(|table| (table.id(), table.restrict_lower_bound()))
.collect();
let mutates_table_elsewhere = prior.iter_tables().any(|table| {
kept.get(&table.id())
.is_none_or(|&bound| bound != table.restrict_lower_bound())
&& elsewhere(&*table.fs, &table.path)
});
let mutates_blob_file_elsewhere = || {
prior.blob_files.iter().any(|blob_file| {
next.blob_files
.get(blob_file.id())
.is_none_or(|kept| kept.live_data_start() != blob_file.live_data_start())
&& elsewhere(&*blob_file.0.fs, &blob_file.0.path)
})
};
if mutates_table_elsewhere || mutates_blob_file_elsewhere() {
SyncMode::Full
} else {
sync_mode
}
}
use alloc::sync::Arc;
#[cfg(not(feature = "std"))]
use alloc::vec::Vec;
#[cfg(feature = "std")]
use arc_swap::ArcSwap;
use crate::path::Path;
#[derive(Clone)]
pub struct SuperVersion {
#[doc(hidden)]
pub active_memtable: Arc<Memtable>,
pub(crate) sealed_memtables: Arc<SealedMemtables>,
pub(crate) version: Version,
pub(crate) seqno: SeqNo,
}
pub enum SnapshotRef {
#[cfg(feature = "std")]
Latest(arc_swap::Guard<Arc<SuperVersion>>),
Owned(SuperVersion),
}
impl core::ops::Deref for SnapshotRef {
type Target = SuperVersion;
fn deref(&self) -> &SuperVersion {
match self {
#[cfg(feature = "std")]
Self::Latest(guard) => guard,
Self::Owned(version) => version,
}
}
}
#[derive(Clone, Copy, Debug)]
pub enum RetentionEffect {
Keep,
GcBelow(SeqNo),
DropsData,
}
impl RetentionEffect {
pub(crate) fn of_run(
filter_transformed: bool,
collected_below_watermark: bool,
watermark: SeqNo,
) -> Self {
if filter_transformed {
Self::DropsData
} else if collected_below_watermark {
Self::GcBelow(watermark)
} else {
Self::Keep
}
}
}
pub struct SuperVersions {
current: SuperVersion,
comparator_name: Arc<str>,
sync_mode: SyncMode,
snapshot_id: VersionId,
log_rotate_bytes: u64,
log_bytes: Option<u64>,
rotation_pending: bool,
rotations: u64,
kept_outputs: Vec<(Option<u64>, crate::compaction::output_ledger::KeptOutputs)>,
edit_scratch: Vec<u8>,
#[cfg(feature = "std")]
latest: Arc<ArcSwap<SuperVersion>>,
filter_budget: Option<Arc<crate::filter_budget::FilterBudget>>,
released: Arc<crate::blob_tree::released::ReleasedObjects>,
}
impl SuperVersions {
pub fn new(
version: Version,
comparator: &SharedComparator,
sync_mode: SyncMode,
snapshot_id: VersionId,
log_rotate_bytes: u64,
) -> Self {
let comparator_name: Arc<str> = comparator.name().into();
let initial = SuperVersion {
active_memtable: Arc::new(Memtable::new(0, comparator.clone())),
sealed_memtables: Arc::default(),
seqno: version.retention_floor(),
version,
};
Self {
#[cfg(feature = "std")]
latest: Arc::new(ArcSwap::from_pointee(initial.clone())),
current: initial,
comparator_name,
sync_mode,
snapshot_id,
log_rotate_bytes,
log_bytes: None,
rotation_pending: false,
rotations: 0,
kept_outputs: Vec::new(),
edit_scratch: Vec::new(),
filter_budget: None,
released: Arc::default(),
}
}
pub(crate) fn released(&self) -> &Arc<crate::blob_tree::released::ReleasedObjects> {
&self.released
}
pub(crate) fn keep_outputs(&mut self, kept: crate::compaction::output_ledger::KeptOutputs) {
let until = self.rotation_pending.then_some(self.rotations);
self.kept_outputs.push((until, kept));
}
pub(crate) fn holds_kept_outputs(&self) -> bool {
!self.kept_outputs.is_empty()
}
pub(crate) fn take_superseded_outputs(
&mut self,
) -> Vec<crate::compaction::output_ledger::KeptOutputs> {
let rotations = self.rotations;
let (superseded, held) = core::mem::take(&mut self.kept_outputs)
.into_iter()
.partition::<Vec<_>, _>(|(until, _)| until.is_none_or(|kept_at| kept_at < rotations));
self.kept_outputs = held;
superseded.into_iter().map(|(_, kept)| kept).collect()
}
#[must_use]
pub(crate) fn with_filter_budget(
mut self,
budget: Arc<crate::filter_budget::FilterBudget>,
) -> Self {
budget.publish(&self.current.version);
self.filter_budget = Some(budget);
self
}
#[expect(
clippy::too_many_arguments,
reason = "version upgrade threads tree_path, mutator closure, two seqno gens, fs, \
runtime, encryption, retention effect: every parameter is load-bearing \
per the manifest-persist contract"
)]
pub(crate) fn upgrade_version<F: FnOnce(&SuperVersion) -> crate::Result<SuperVersion>>(
&mut self,
tree_path: &Path,
f: F,
seqno: &SharedSequenceNumberGenerator,
visible_seqno: &SharedSequenceNumberGenerator,
fs: &dyn Fs,
runtime: Arc<crate::runtime_config::RuntimeConfig>,
encryption: Option<Arc<dyn crate::encryption::EncryptionProvider>>,
retention: RetentionEffect,
) -> crate::Result<()> {
self.upgrade_version_with_seqno(
tree_path,
f,
seqno.next(),
visible_seqno,
fs,
runtime,
encryption,
retention,
)
}
#[expect(
clippy::too_many_arguments,
reason = "version upgrade with pre-allocated seqno: tree_path, mutator, seqno, \
visible_seqno, fs, runtime, encryption, retention effect; same \
load-bearing surface as the auto-allocating sibling above"
)]
pub(crate) fn upgrade_version_with_seqno<
F: FnOnce(&SuperVersion) -> crate::Result<SuperVersion>,
>(
&mut self,
tree_path: &Path,
f: F,
seqno: SeqNo,
visible_seqno: &SharedSequenceNumberGenerator,
fs: &dyn Fs,
runtime: Arc<crate::runtime_config::RuntimeConfig>,
encryption: Option<Arc<dyn crate::encryption::EncryptionProvider>>,
retention: RetentionEffect,
) -> crate::Result<()> {
let prior = self.latest_version();
let seqno = seqno.max(prior.seqno);
let mut next_version = f(&prior)?;
debug_assert!(
next_version.version.id() > prior.version.id(),
"an install must advance the version id ({} -> {})",
prior.version.id(),
next_version.version.id(),
);
debug_assert!(
next_version
.version
.referenced_dicts()
.iter()
.all(|id| next_version.version.dicts().contains(id)),
"every dictionary a file references is registered ({:?} of {:?})",
next_version.version.referenced_dicts(),
next_version.version.dicts(),
);
next_version.seqno = seqno;
log::trace!("Next version seqno={}", next_version.seqno);
let floor = match retention {
RetentionEffect::Keep | RetentionEffect::GcBelow(0) => None,
RetentionEffect::GcBelow(watermark) => Some((watermark - 1).min(seqno)),
RetentionEffect::DropsData => Some(seqno),
};
if let Some(floor) = floor {
next_version.version.set_retention_floor(floor);
}
sync_new_table_directories(
&prior.version,
&next_version.version,
tree_path,
fs,
self.sync_mode,
)?;
let manifest_mode = manifest_sync_mode(
&prior.version,
&next_version.version,
tree_path,
fs,
self.sync_mode,
);
self.persist_change(
tree_path,
&prior.version,
&next_version.version,
fs,
runtime,
encryption,
manifest_mode,
)?;
self.append_version(next_version);
let next_visible = seqno.saturating_add(1).min(MAX_SEQNO);
visible_seqno.fetch_max(next_visible);
Ok(())
}
#[expect(
clippy::too_many_arguments,
reason = "the transition, where it goes, how it is encoded and how durably: \
every parameter is the install's own"
)]
fn persist_change(
&mut self,
tree_path: &Path,
prior: &Version,
next: &Version,
fs: &dyn Fs,
runtime: Arc<crate::runtime_config::RuntimeConfig>,
encryption: Option<Arc<dyn crate::encryption::EncryptionProvider>>,
sync_mode: SyncMode,
) -> crate::Result<()> {
let log_path = edit_log::log_path(tree_path, self.snapshot_id);
let log_size = if let Some(n) = self.log_bytes {
n
} else {
let n = edit_log::log_size(fs, &log_path)?;
self.log_bytes = Some(n);
n
};
if log_size < self.log_rotate_bytes
&& !self.rotation_pending
&& prior.repair_report() == next.repair_report()
{
let edit = next.diff(prior)?;
match edit_log::append_edit(fs, &log_path, &edit, &mut self.edit_scratch, sync_mode) {
Ok(Some(appended)) => {
self.log_bytes = Some(log_size + appended);
return Ok(());
}
Ok(None) => {
self.edit_scratch.clear();
self.edit_scratch
.shrink_to(crate::version::framing::MAX_FRAME_PAYLOAD as usize);
}
Err(e) => {
self.log_bytes = None;
self.rotation_pending = true;
return Err(e);
}
}
}
let old_snapshot = self.snapshot_id;
if let Err(e) = persist_version(
tree_path,
next,
&self.comparator_name,
fs,
runtime,
encryption,
sync_mode,
) {
self.rotation_pending = true;
return Err(e);
}
self.rotation_pending = false;
self.rotations += 1;
self.snapshot_id = next.id();
self.log_bytes = Some(0);
if let Err(e) = remove_if_present(fs, &log_path) {
log::warn!(
"rotation: failed to remove old edit log {}: {e}",
log_path.display()
);
}
if old_snapshot != self.snapshot_id {
let old_path = tree_path.join(format!("v{old_snapshot}"));
if let Err(e) = remove_if_present(fs, &old_path) {
log::warn!(
"rotation: failed to remove old snapshot {}: {e}",
old_path.display()
);
}
}
Ok(())
}
pub fn append_version(&mut self, version: SuperVersion) {
if let Some(budget) = &self.filter_budget {
budget.publish(&version.version);
}
#[cfg(feature = "std")]
self.latest.store(Arc::new(version.clone()));
self.current = version;
}
pub fn replace_latest_version(&mut self, version: SuperVersion) {
self.append_version(version);
}
#[cfg(feature = "std")]
#[must_use]
pub(crate) fn latest_handle(&self) -> Arc<ArcSwap<SuperVersion>> {
Arc::clone(&self.latest)
}
pub fn latest_version(&self) -> SuperVersion {
self.latest_version_ref().clone()
}
pub fn latest_version_ref(&self) -> &SuperVersion {
&self.current
}
#[must_use]
pub fn registered_dicts(&self) -> Vec<crate::file::DictId> {
self.current.version.dicts().to_vec()
}
pub fn get_version_for_snapshot(&self, seqno: SeqNo) -> crate::Result<SuperVersion> {
self.current.check_serves(seqno)?;
Ok(self.current.clone())
}
}
impl SuperVersion {
#[inline]
pub(crate) fn check_serves(&self, seqno: SeqNo) -> crate::Result<()> {
let floor = self.version.retention_floor();
if seqno > floor || seqno == 0 {
Ok(())
} else {
Err(below_retention(seqno, floor))
}
}
pub(crate) fn unflushed_bytes(&self) -> u64 {
self.active_memtable.size()
+ self
.sealed_memtables
.iter()
.map(|memtable| memtable.size())
.sum::<u64>()
}
pub(crate) fn memtables_reference_blob_file(
&self,
blob_file_id: crate::vlog::BlobFileId,
) -> bool {
self.active_memtable.references_blob_file(blob_file_id)
|| self
.sealed_memtables
.iter()
.any(|memtable| memtable.references_blob_file(blob_file_id))
}
pub(crate) fn memtable_blob_files(&self) -> crate::HashSet<crate::vlog::BlobFileId> {
let mut files = crate::HashSet::default();
self.active_memtable.collect_blob_files(&mut files);
for memtable in self.sealed_memtables.iter() {
memtable.collect_blob_files(&mut files);
}
files
}
}
#[cold]
#[inline(never)]
fn below_retention(requested: SeqNo, floor: SeqNo) -> crate::Error {
crate::Error::SnapshotBelowRetention {
requested,
oldest_retained: floor,
}
}
#[cfg(test)]
mod tests;