use super::archive_certificate::{
ExpectedBinaryReferences, ExpectedGraph, validate_exact_archive_trailers,
};
use super::file_identity::{
FileAccess, RegularFileIdentity, UncommittedArchiveStaging, held_file_identity,
open_regular_file_no_follow, path_object_identity, preserve_file_metadata,
remove_published_link_if_same, require_held_file_identity, require_path_file_identity,
sync_directory_strict,
};
use super::providers::{
ArchiveSegmentsProvider, FilteredTrailers, archive_segments_provider,
certify_reopened_active_archive,
};
use super::reclaim::{ArchiveRewritePolicy, plan_archive_sweep};
use super::sweep_plan::DeferredFileDeletion;
use super::sweep_plan::{ArchiveSweepDisposition, ArchiveSweepOutcome, PlannedArchiveSweep};
use crate::content::provider::SegmentProvider;
use crate::error::{Error, Result};
use crate::segment::identifier::SegmentIdentifier;
use crate::tar_archive::archive::TarArchiveReader;
use crate::writer::compaction::CompactionKind;
use crate::writer::segment_builder::GarbageCollectionGeneration;
use crate::writer::tar_writer::TarArchiveWriter;
use std::path::Path;
mod publish;
mod stage;
mod validate;
pub(crate) use publish::*;
pub(crate) use stage::*;
pub(crate) use validate::*;
#[allow(
clippy::too_many_arguments,
reason = "staging, semantic validation, atomic publication, and source unlinking form one deliberately linear safety sequence, and each parameter names one input that sequence must not re-derive"
)]
pub(super) fn sweep_one_archive<'archives>(
directory: &Path,
reader: &'archives TarArchiveReader,
reclaimable: &std::collections::HashSet<SegmentIdentifier>,
previously_unavailable_graph_targets: &std::collections::HashSet<SegmentIdentifier>,
all_archives: &[&'archives TarArchiveReader],
fallback_provider: &mut Option<ArchiveSegmentsProvider<'archives>>,
source_certificate_provider: Option<&dyn SegmentProvider>,
rewrite_policy: ArchiveRewritePolicy,
) -> Result<ArchiveSweepOutcome> {
let path = directory.join(reader.file_name());
let Some(planned) = plan_archive_sweep(
directory,
reader,
reclaimable,
rewrite_policy,
&std::collections::HashSet::new(),
)?
else {
return Ok(ArchiveSweepOutcome::default());
};
let reopened_source = reopen_actionable_source(
directory,
&path,
reader,
reclaimable,
&planned,
source_certificate_provider,
rewrite_policy,
)?;
let reader = reopened_source
.as_ref()
.map_or(reader, |(reopened, _, _)| reopened);
let Some(index) = reader.index() else {
return Ok(ArchiveSweepOutcome::default());
};
let planned_unavailable = reclaimable_in_archive(index, reclaimable);
let (replacement_name, planned_reclaimable_count) = match planned {
PlannedArchiveSweep::Remove { .. } => {
return remove_swept_archive(
&path,
reader,
reopened_source.as_ref(),
planned_unavailable,
);
}
PlannedArchiveSweep::Rewrite {
replacement_name,
segment_count,
..
} => (replacement_name, segment_count),
PlannedArchiveSweep::DeferredBySavings { .. }
| PlannedArchiveSweep::DeferredAtLastGeneration { .. }
| PlannedArchiveSweep::BlockedByOccupiedGeneration { .. } => {
return Ok(ArchiveSweepOutcome::default());
}
};
let survivors = partition_entries(index, reclaimable, planned_reclaimable_count);
let current_rewrite_targets = planned_unavailable;
let trailers = source_trailers(
reader,
reclaimable,
previously_unavailable_graph_targets,
¤t_rewrite_targets,
);
let scan_provider = scan_provider_for(&trailers, all_archives, fallback_provider)?;
let staging_name = next_archive_staging_name(directory, &replacement_name)?;
let staging_path = directory.join(&staging_name);
let replacement_path = directory.join(&replacement_name);
let mut uncommitted_staging = UncommittedArchiveStaging::new(directory, staging_path.clone());
let (_, source_file, source_identity) = reopened_source
.as_ref()
.expect("an archive rewrite always has an actionable reopened source");
let source_metadata = source_file.metadata()?;
let mut writer =
TarArchiveWriter::new_exclusive_staged(directory, &staging_name, &replacement_name);
let (expected_graph, expected_binary_references) = copy_survivors(
SurvivorSources {
reader,
trailers: &trailers,
previously_unavailable_graph_targets,
current_rewrite_targets: ¤t_rewrite_targets,
scan_provider,
},
&survivors,
&mut writer,
&mut uncommitted_staging,
)?;
writer.close()?;
publish_swept_archive(
SweepPublication {
directory,
source_path: &path,
reader,
source_file,
source_identity: *source_identity,
staging_name: &staging_name,
staging_path: &staging_path,
replacement_name: &replacement_name,
replacement_path: &replacement_path,
survivors: &survivors,
expected_graph: &expected_graph,
expected_binary_references: &expected_binary_references,
source_metadata: &source_metadata,
current_rewrite_targets,
},
&mut uncommitted_staging,
)
}
#[allow(
clippy::too_many_arguments,
reason = "the re-certification repeats every input the original plan was derived from"
)]
pub(crate) fn reopen_actionable_source(
directory: &Path,
path: &Path,
reader: &TarArchiveReader,
reclaimable: &std::collections::HashSet<SegmentIdentifier>,
planned: &PlannedArchiveSweep,
source_certificate_provider: Option<&dyn SegmentProvider>,
rewrite_policy: ArchiveRewritePolicy,
) -> Result<Option<(TarArchiveReader, std::fs::File, RegularFileIdentity)>> {
let reopened_source = if planned.changes_disk() {
let source_file = open_regular_file_no_follow(path, FileAccess::ReadOnly)?;
let source_identity = held_file_identity(&source_file)?;
let reopened = TarArchiveReader::open_file(path, &source_file)?;
if let Some(provider) = source_certificate_provider {
certify_reopened_active_archive(provider, &reopened)?;
}
if plan_archive_sweep(
directory,
&reopened,
reclaimable,
rewrite_policy,
&std::collections::HashSet::new(),
)?
.as_ref()
!= Some(planned)
{
return Err(Error::InvalidFormat {
details: format!(
"the actionable source archive {} changed before its immediate cleanup certificate",
reader.file_name()
),
});
}
Some((reopened, source_file, source_identity))
} else {
None
};
Ok(reopened_source)
}
pub(crate) fn reclaimable_in_archive(
index: &crate::tar_archive::index::SegmentIndex,
reclaimable: &std::collections::HashSet<SegmentIdentifier>,
) -> std::collections::HashSet<SegmentIdentifier> {
index
.entries()
.iter()
.map(|entry| entry.segment_identifier)
.filter(|identifier| reclaimable.contains(identifier))
.collect()
}
pub(crate) fn remove_swept_archive(
path: &Path,
reader: &TarArchiveReader,
reopened_source: Option<&(TarArchiveReader, std::fs::File, RegularFileIdentity)>,
planned_unavailable: std::collections::HashSet<SegmentIdentifier>,
) -> Result<ArchiveSweepOutcome> {
let (_, source_file, source_identity) = reopened_source
.as_ref()
.expect("an archive removal always has an actionable reopened source");
#[cfg(test)]
crate::writer::fault_injection::substitute_path_if_armed(
"sweep.remove-before-source-identity",
path,
)?;
require_held_file_identity(source_file, *source_identity, "certified removal source")?;
require_path_file_identity(path, *source_identity, "certified removal source")?;
#[cfg(test)]
crate::writer::fault_injection::remove_path_if_armed(
"sweep.remove-before-source-unlink-not-found",
path,
)?;
Ok(match std::fs::remove_file(path) {
Ok(()) => ArchiveSweepOutcome {
disposition: ArchiveSweepDisposition::Removed,
deletion_failures: Vec::new(),
newly_unavailable: planned_unavailable,
},
Err(error) => ArchiveSweepOutcome {
disposition: ArchiveSweepDisposition::Unchanged,
deletion_failures: vec![DeferredFileDeletion {
file_name: reader.file_name().to_owned(),
error: error.to_string(),
target_was_already_absent: error.kind() == std::io::ErrorKind::NotFound,
}],
newly_unavailable: std::collections::HashSet::new(),
},
})
}
pub(crate) fn partition_entries(
index: &crate::tar_archive::index::SegmentIndex,
reclaimable: &std::collections::HashSet<SegmentIdentifier>,
planned_reclaimable_count: usize,
) -> Vec<crate::tar_archive::index::SegmentIndexEntry> {
let mut entries: Vec<_> = index.entries().to_vec();
entries.sort_by_key(|entry| entry.position);
let mut survivors = Vec::new();
for entry in entries {
if !reclaimable.contains(&entry.segment_identifier) {
survivors.push(entry);
}
}
debug_assert_eq!(
index.entries().len() - survivors.len(),
planned_reclaimable_count
);
survivors
}
pub(crate) fn source_trailers(
reader: &TarArchiveReader,
reclaimable: &std::collections::HashSet<SegmentIdentifier>,
previously_unavailable_graph_targets: &std::collections::HashSet<SegmentIdentifier>,
current_rewrite_targets: &std::collections::HashSet<SegmentIdentifier>,
) -> FilteredTrailers {
FilteredTrailers::from_archive(
reader,
reclaimable,
previously_unavailable_graph_targets,
current_rewrite_targets,
)
}
pub(crate) fn scan_provider_for<'fallback, 'archives>(
trailers: &FilteredTrailers,
all_archives: &[&'archives TarArchiveReader],
fallback_provider: &'fallback mut Option<ArchiveSegmentsProvider<'archives>>,
) -> Result<Option<&'fallback ArchiveSegmentsProvider<'archives>>> {
if trailers.catalog.is_some() {
return Ok(None);
}
if fallback_provider.is_none() {
*fallback_provider = Some(archive_segments_provider(all_archives)?);
}
Ok(fallback_provider.as_ref())
}
pub(crate) fn is_reclaimable(
reference: GarbageCollectionGeneration,
segment: GarbageCollectionGeneration,
kind: CompactionKind,
retained_generations: i32,
) -> bool {
match kind {
CompactionKind::Full => {
reference
.full_generation
.wrapping_sub(segment.full_generation)
>= retained_generations
|| (reference.generation.wrapping_sub(segment.generation) >= retained_generations
&& !segment.is_compacted)
}
CompactionKind::Tail => {
reference.generation.wrapping_sub(segment.generation) >= retained_generations
&& !(segment.is_compacted && segment.full_generation == reference.full_generation)
}
}
}
#[cfg(test)]
pub(super) fn probe_archive_sweep_phase_boundary(cutpoint: &str) -> Result<()> {
crate::writer::fault_injection::fail_if_armed(cutpoint)?;
crate::writer::fault_injection::crash_if_armed(cutpoint);
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
use crate::segment::identifier::SegmentIdentifier;
use crate::tar_archive::archive::TarArchiveReader;
use crate::writer::compaction::CompactionKind;
use crate::writer::store_writer::reclaim::*;
use crate::writer::store_writer::sweep_plan::*;
use crate::writer::store_writer::test_support::*;
use std::collections::HashSet;
use std::path::Path;
#[test]
fn full_reclaimer_retained_two_honors_exact_and_wrapping_boundaries() {
let reference = generation(10, 8, true);
assert!(is_reclaimable(
reference,
generation(10, 6, true),
CompactionKind::Full,
2
));
assert!(!is_reclaimable(
reference,
generation(10, 7, true),
CompactionKind::Full,
2
));
assert!(is_reclaimable(
reference,
generation(8, 7, false),
CompactionKind::Full,
2
));
assert!(!is_reclaimable(
reference,
generation(8, 7, true),
CompactionKind::Full,
2
));
assert!(!is_reclaimable(
reference,
generation(9, 7, false),
CompactionKind::Full,
2
));
let wrapping_reference = generation(i32::MIN, i32::MIN, false);
assert!(!is_reclaimable(
wrapping_reference,
generation(i32::MAX, i32::MAX, false),
CompactionKind::Full,
2
));
assert!(is_reclaimable(
wrapping_reference,
generation(i32::MAX - 1, i32::MAX - 1, false),
CompactionKind::Full,
2
));
assert!(!is_reclaimable(
generation(i32::MAX, i32::MAX, false),
generation(i32::MIN, i32::MIN, false),
CompactionKind::Full,
2
));
}
#[test]
fn post_compaction_reclaimer_still_retains_exactly_one_generation() {
let reference = generation(5, 5, true);
assert!(is_reclaimable(
reference,
generation(4, 4, true),
CompactionKind::Full,
1
));
assert!(!is_reclaimable(
reference,
generation(5, 5, true),
CompactionKind::Full,
1
));
assert!(is_reclaimable(
reference,
generation(4, 5, false),
CompactionKind::Tail,
1
));
assert!(!is_reclaimable(
reference,
generation(4, 5, true),
CompactionKind::Tail,
1
));
}
#[test]
fn one_retained_generation_reclaims_what_two_spared() {
let tail_compacted_head = generation(1, 0, true);
let untouched_tail = generation(0, 0, false);
assert!(is_reclaimable(
tail_compacted_head,
untouched_tail,
CompactionKind::Full,
1
));
assert!(!is_reclaimable(
tail_compacted_head,
untouched_tail,
CompactionKind::Full,
2
));
assert_eq!(
crate::writer::store_writer::RETAINED_GENERATIONS,
1,
"the run's own retention value is the one this boundary describes"
);
}
#[test]
fn occupied_next_generation_is_never_truncated_or_rewritten() {
let directory = TestDirectory::new("occupied-next-generation");
let root = data_identifier(10);
let old_one = data_identifier(11);
let old_two = data_identifier(12);
write_test_archive(
&directory,
"data00000a.tar",
&[
TestArchiveEntry::new(root, 1, generation(4, 4, false)),
TestArchiveEntry::new(old_one, 1, generation(0, 0, false)),
TestArchiveEntry::new(old_two, 1, generation(0, 0, false)),
],
);
let occupied = b"interrupted-cleanup-evidence-must-survive";
std::fs::write(directory.path.join("data00000b.tar"), occupied)
.expect("write occupied target");
let reader =
TarArchiveReader::open(&directory.path.join("data00000a.tar")).expect("open source");
let cleaned = HashSet::from([old_one, old_two]);
let planned = plan_archive_sweep(
&directory.path,
&reader,
&cleaned,
ArchiveRewritePolicy::default(),
&std::collections::HashSet::new(),
)
.expect("plan")
.expect("archive has reclaimable entries");
assert!(matches!(
planned,
PlannedArchiveSweep::BlockedByOccupiedGeneration {
ref occupied_name,
..
} if occupied_name == "data00000b.tar"
));
let mut fallback = None;
sweep_one_archive(
&directory.path,
&reader,
&cleaned,
&cleaned,
&[&reader],
&mut fallback,
None,
ArchiveRewritePolicy::default(),
)
.expect("blocked sweep is a safe no-op");
assert_eq!(
std::fs::read(directory.path.join("data00000b.tar")).expect("read occupied target"),
occupied
);
assert!(directory.path.join("data00000a.tar").exists());
}
#[test]
fn occupied_higher_generation_blocks_whole_archive_removal() {
let directory = TestDirectory::new("occupied-blocks-whole-removal");
let obsolete = data_identifier(13);
write_test_archive(
&directory,
"data00000a.tar",
&[TestArchiveEntry::new(obsolete, 1, generation(0, 0, false))],
);
let occupied = b"damaged-higher-generation-must-not-become-active";
std::fs::write(directory.path.join("data00000c.tar"), occupied)
.expect("write recovered residue");
let source_path = directory.path.join("data00000a.tar");
let source_before = std::fs::read(&source_path).expect("read source");
let reader = TarArchiveReader::open(&source_path).expect("open source");
let cleaned = HashSet::from([obsolete]);
assert!(matches!(
plan_archive_sweep(
&directory.path,
&reader,
&cleaned,
ArchiveRewritePolicy::default(),
&std::collections::HashSet::new(),
)
.expect("plan")
.expect("eligible archive"),
PlannedArchiveSweep::BlockedByOccupiedGeneration {
occupied_name,
segment_count: 1,
..
} if occupied_name == "data00000c.tar"
));
let mut fallback = None;
sweep_one_archive(
&directory.path,
&reader,
&cleaned,
&cleaned,
&[&reader],
&mut fallback,
None,
ArchiveRewritePolicy::default(),
)
.expect("blocked removal is a no-op");
assert_eq!(
std::fs::read(source_path).expect("source remains"),
source_before
);
assert_eq!(
std::fs::read(directory.path.join("data00000c.tar")).expect("residue remains"),
occupied
);
}
#[test]
fn lower_stale_generation_blocks_whole_active_archive_removal() {
let directory = TestDirectory::new("lower-letter-blocks-whole-removal");
let stale = data_identifier(14);
let obsolete = data_identifier(15);
write_test_archive(
&directory,
"data00000a.tar",
&[TestArchiveEntry::new(stale, 1, generation(0, 0, false))],
);
write_test_archive(
&directory,
"data00000b.tar",
&[TestArchiveEntry::new(obsolete, 1, generation(0, 0, false))],
);
let stale_path = directory.path.join("data00000a.tar");
let active_path = directory.path.join("data00000b.tar");
let stale_before = std::fs::read(&stale_path).expect("read stale generation");
let active_before = std::fs::read(&active_path).expect("read active generation");
let active = TarArchiveReader::open(&active_path).expect("open active generation");
let cleaned = HashSet::from([obsolete]);
assert!(matches!(
plan_archive_sweep(
&directory.path,
&active,
&cleaned,
ArchiveRewritePolicy::default(),
&std::collections::HashSet::new(),
)
.expect("plan")
.expect("eligible archive"),
PlannedArchiveSweep::BlockedByOccupiedGeneration {
occupied_name,
segment_count: 1,
..
} if occupied_name == "data00000a.tar"
));
let mut fallback = None;
sweep_one_archive(
&directory.path,
&active,
&cleaned,
&cleaned,
&[&active],
&mut fallback,
None,
ArchiveRewritePolicy::default(),
)
.expect("blocked removal is a no-op");
assert_eq!(
std::fs::read(active_path).expect("active remains"),
active_before
);
assert_eq!(
std::fs::read(stale_path).expect("stale remains"),
stale_before
);
}
#[test]
fn last_generation_z_is_deferred_without_creating_an_invalid_successor() {
let directory = TestDirectory::new("generation-z");
let root = data_identifier(20);
let old_one = data_identifier(21);
let old_two = data_identifier(22);
write_test_archive(
&directory,
"data00000z.tar",
&[
TestArchiveEntry::new(root, 1, generation(4, 4, false)),
TestArchiveEntry::new(old_one, 1, generation(0, 0, false)),
TestArchiveEntry::new(old_two, 1, generation(0, 0, false)),
],
);
let path = directory.path.join("data00000z.tar");
let before = std::fs::read(&path).expect("read source");
let reader = TarArchiveReader::open(&path).expect("open source");
let cleaned = HashSet::from([old_one, old_two]);
assert!(matches!(
plan_archive_sweep(
&directory.path,
&reader,
&cleaned,
ArchiveRewritePolicy::default(),
&std::collections::HashSet::new(),
)
.expect("plan")
.expect("has eligible entries"),
PlannedArchiveSweep::DeferredAtLastGeneration { .. }
));
let mut fallback = None;
sweep_one_archive(
&directory.path,
&reader,
&cleaned,
&cleaned,
&[&reader],
&mut fallback,
None,
ArchiveRewritePolicy::default(),
)
.expect("z sweep is a no-op");
assert_eq!(std::fs::read(path).expect("read after"), before);
assert_eq!(
std::fs::read_dir(&directory.path)
.expect("list")
.filter_map(std::result::Result::ok)
.filter(|entry| entry.file_name().to_string_lossy().ends_with(".tar"))
.count(),
1
);
}
struct ReplanFixture {
target: SegmentIdentifier,
old_one: SegmentIdentifier,
old_two: SegmentIdentifier,
root: SegmentIdentifier,
}
fn build_replan_fixture(directory: &TestDirectory) -> ReplanFixture {
let target = data_identifier(83);
let retained = data_identifier(84);
let old_one = data_identifier(85);
let old_two = data_identifier(86);
let root = data_identifier(87);
let reference = generation(5, 5, false);
write_test_archive(
directory,
"data00000a.tar",
&[
TestArchiveEntry::new(target, 1, generation(0, 0, false)),
TestArchiveEntry::new(retained, 1, reference),
],
);
write_test_archive(
directory,
"data00001a.tar",
&[
TestArchiveEntry::new(old_one, 1, generation(0, 0, false)),
TestArchiveEntry::new(old_two, 1, generation(0, 0, false)),
TestArchiveEntry::new(root, 1, reference).referencing(&[target]),
],
);
ReplanFixture {
target,
old_one,
old_two,
root,
}
}
fn assert_both_sources_plan_rewrites(
directory: &Path,
sources: [&TarArchiveReader; 2],
reclaimable: &HashSet<SegmentIdentifier>,
) {
for source in sources {
assert!(matches!(
plan_archive_sweep(
directory,
source,
reclaimable,
ArchiveRewritePolicy::default(),
&HashSet::new(),
)
.expect("initial plan")
.expect("archive is initially actionable"),
PlannedArchiveSweep::Rewrite { .. }
));
}
}
#[test]
fn rewrite_replan_noop_reports_no_unavailable_graph_targets() {
let directory = TestDirectory::new("rewrite-replan-noop-graph-target");
let ReplanFixture {
target,
old_one,
old_two,
root,
} = build_replan_fixture(&directory);
let first = TarArchiveReader::open(&directory.path.join("data00000a.tar"))
.expect("open first rewrite source");
let second = TarArchiveReader::open(&directory.path.join("data00001a.tar"))
.expect("open second rewrite source");
let reclaimable = HashSet::from([target, old_one, old_two]);
assert_both_sources_plan_rewrites(&directory.path, [&first, &second], &reclaimable);
let occupied = b"occupied after outer planning";
std::fs::write(directory.path.join("data00000b.tar"), occupied)
.expect("occupy first replacement");
let provider_order = [&first, &second];
let mut fallback = None;
let mut actually_unavailable = HashSet::new();
let first_outcome = sweep_one_archive(
&directory.path,
&first,
&reclaimable,
&actually_unavailable,
&provider_order,
&mut fallback,
None,
ArchiveRewritePolicy::default(),
)
.expect("blocked immediate replan is a no-op");
assert!(first_outcome.deletion_failures.is_empty());
assert!(
first_outcome.newly_unavailable.is_empty(),
"a planned rewrite that never published cannot justify graph filtering"
);
assert!(
directory.path.join("data00000a.tar").exists(),
"the blocked immediate replan must leave its source available"
);
assert_eq!(
std::fs::read(directory.path.join("data00000b.tar")).expect("read occupied target"),
occupied,
"the blocked immediate replan must not replace the new pathname"
);
actually_unavailable.extend(first_outcome.newly_unavailable);
let second_outcome = sweep_one_archive(
&directory.path,
&second,
&reclaimable,
&actually_unavailable,
&provider_order,
&mut fallback,
None,
ArchiveRewritePolicy::default(),
)
.expect("second rewrite publishes");
assert_eq!(
second_outcome.newly_unavailable,
HashSet::from([old_one, old_two])
);
let rewritten = TarArchiveReader::open(&directory.path.join("data00001b.tar"))
.expect("open second replacement");
assert_eq!(
rewritten.segment_graph().expect("valid graph").as_map()[&root],
[target],
"the later rewrite must retain an edge to the still-available first target"
);
}
}