server/lib/src/purge.rs
//! **Emptying a repository, on the main and on its twin** — the owner's
//! non-delivery ruling (2026-09-29): a tenant whose order lapsed is read-only at
//! once and, after the grace the ONE box decides (its `[nonpay] delete_after_days`;
//! holger keeps no day count, `metered::purgeable`), its repository is DELETED, here
//! and on the twin.
//!
//! In holger a tenant IS a repository, so the unit emptied is one repository.
//! This module holds the parts that are not about money and therefore compile in
//! every build:
//!
//! * [`walk`] — list the repository through its backend, page by page, and
//! delete every artifact a predicate chooses (`RepositoryBackendTrait::
//! delete_artifact`, content-id verified like the retention executor). A
//! backend that cannot delete (the immutable znippy archives) answers with
//! EACH artifact it could not remove in `failures` — never a false success;
//! * the **gate** ([`close`], [`is_closed`]) — while a walk runs, every write
//! door refuses an upload to that repository by name, so nothing the tenant
//! does lands behind the walk;
//! * the **record** ([`Record`]) — one immutable JSON file per completed purge
//! under `<data_root>/holger-purges/` (an [`crate::replication::ESTATE`]
//! item, so the twin's estate pull carries it);
//! * the twin's **application** of those records ([`apply_records`]): holger's
//! twin mirror never deletes locally, so without it a purge on the main would
//! survive on the twin for ever. An artifact the twin stored AFTER the purge
//! (the tenant came back and uploaded again) is KEPT — the time rule — and a
//! record is applied ONCE, remembered in a twin-local file that is NOT in the
//! estate ([`APPLIED_FILE`]).
//!
//! The verb that walks on the main — its admin door, its roster, and the state
//! the repository must be in before anything is deleted — is the metered
//! build's (`crate::metered`, `EntitlementService.Purge`).
use crate::lock::LockOrRecover as _;
use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use serde::{Deserialize, Serialize};
use traits::{ArtifactEntry, ArtifactId, RepositoryBackendTrait};
/// The records, one file per completed purge, under the data root.
pub const RECORDS_DIR: &str = "holger-purges";
/// **This box's own memory of which records it has applied** — under the data
/// root and deliberately NOT an estate item: a twin that pulled it from its
/// primary would believe it had applied what it never did.
pub const APPLIED_FILE: &str = "holger-purges-applied.json";
/// How many artifacts one `list` asks for.
pub const PAGE: usize = 1000;
// ── the walk ────────────────────────────────────────────────────────────────
/// What a walk did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Walked {
pub deleted: u64,
pub bytes: u64,
/// Artifacts the predicate chose to keep.
pub kept: u64,
/// One line per artifact (or per listing) that could not be removed. Non-empty
/// means the repository's bytes are PARTIALLY still there.
pub failures: Vec<String>,
}
/// The walk's question about one listed artifact.
pub enum Verdict {
Delete,
Keep,
/// Cannot be decided — reported as a failure, the artifact kept.
Undecided(String),
}
fn key(id: &ArtifactId) -> (Option<String>, String, String) {
(id.namespace.clone(), id.name.clone(), id.version.clone())
}
/// `namespace/name@version`, or `name@version`, or the bare name of a raw path.
pub fn label(id: &ArtifactId) -> String {
let base = match &id.namespace {
Some(ns) if !ns.is_empty() => format!("{ns}/{}", id.name),
_ => id.name.clone(),
};
if id.version.is_empty() {
base
} else {
format!("{base}@{}", id.version)
}
}
/// **Delete every artifact of `backend` that `decide` chooses**, paging
/// `list` until nothing is left to decide. Each delete is content-id verified
/// (`fetch`, sha256, `delete_artifact`) exactly as the retention executor does,
/// so bytes that changed under the walk are refused rather than removed blind.
///
/// Terminates on any backend: every listed artifact is either deleted (and so
/// leaves the listing) or remembered as kept/failed (and so skipped, with the
/// next page asked for that many more).
pub fn walk(
backend: &dyn RepositoryBackendTrait,
decide: &dyn Fn(&ArtifactEntry) -> Verdict,
) -> Walked {
let mut out = Walked::default();
if backend.caps().group {
// A group is a view over other repositories and owns no bytes of its own.
return out;
}
if !backend.caps().listing {
out.failures.push(format!(
"backend `{}` cannot enumerate what it holds, so nothing here can say its artifacts are gone",
backend.name()
));
return out;
}
let mut passed: BTreeSet<(Option<String>, String, String)> = BTreeSet::new();
loop {
let limit = PAGE + passed.len();
let page = match backend.list(None, limit) {
Ok(p) => p,
Err(e) => {
out.failures
.push(format!("listing `{}`: {e:#}", backend.name()));
return out;
}
};
let complete = page.len() < limit;
let fresh: Vec<&ArtifactEntry> = page
.iter()
.filter(|e| !passed.contains(&key(&e.id)))
.collect();
if fresh.is_empty() {
return out;
}
for e in fresh {
match decide(e) {
Verdict::Keep => {
out.kept += 1;
passed.insert(key(&e.id));
}
Verdict::Undecided(why) => {
out.failures.push(format!("{}: {why}", label(&e.id)));
passed.insert(key(&e.id));
}
Verdict::Delete => match backend.fetch(&e.id) {
// Gone between the listing and the fetch: nothing to remove.
Ok(None) => {
passed.insert(key(&e.id));
}
Err(err) => {
out.failures
.push(format!("{}: read failed: {err:#}", label(&e.id)));
passed.insert(key(&e.id));
}
Ok(Some(bytes)) => {
let content_id = nornir_hash::sha256_hex(&bytes);
match backend.delete_artifact(&e.id, &content_id) {
Ok(()) => {
out.deleted += 1;
out.bytes += bytes.len() as u64;
// A backend whose delete said Ok but still lists it
// must not loop the walk for ever.
passed.insert(key(&e.id));
}
Err(err) => {
out.failures.push(format!("{}: {err:#}", label(&e.id)));
passed.insert(key(&e.id));
}
}
}
},
}
}
if complete {
return out;
}
}
}
// ── the gate ────────────────────────────────────────────────────────────────
static GATE: Mutex<BTreeSet<(PathBuf, String)>> = Mutex::new(BTreeSet::new());
/// **The upload gate of one repository while it is walked.** Held for the walk;
/// dropping it opens the repository's writes again (whatever else still refuses
/// them keeps doing so). Keyed by a directory that names the server (its store
/// root), so two servers in one process never share a gate.
pub struct Gate {
key: (PathBuf, String),
}
/// Close `repository`'s writes on the server named by `scope`. `None` = a walk
/// already holds it.
pub fn close(scope: &Path, repository: &str) -> Option<Gate> {
let key = (scope.to_path_buf(), repository.to_string());
let mut g = GATE.lock_or_recover();
if !g.insert(key.clone()) {
return None;
}
Some(Gate { key })
}
/// Is `repository` being walked right now on the server named by `scope`?
pub fn is_closed(scope: &Path, repository: &str) -> bool {
GATE.lock_or_recover()
.contains(&(scope.to_path_buf(), repository.to_string()))
}
impl Drop for Gate {
fn drop(&mut self) {
GATE.lock_or_recover().remove(&self.key);
}
}
/// The refusal an upload meets while its repository is walked.
pub fn gated_refusal(repository: &str) -> String {
format!(
"repository `{repository}` is being purged: every upload is refused while its artifacts are deleted"
)
}
// ── the record ──────────────────────────────────────────────────────────────
/// **One completed purge**, written once and never changed.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Record {
pub repository: String,
/// When the walk FINISHED, by this server's clock. Every artifact stored at
/// or before it was the purged tenant's; one stored after it is new.
pub at_unix_ms: u64,
pub reason: String,
pub artifacts: u64,
pub bytes: u64,
}
fn file_safe(repository: &str) -> String {
repository.replace(['/', '\\'], "_").replace("..", "_")
}
/// The records directory under a data root.
pub fn records_dir(data_root: &Path) -> PathBuf {
data_root.join(RECORDS_DIR)
}
/// **Write the record** `<data_root>/holger-purges/<repository>-<at_unix_ms>.json`
/// — create-new (an existing record is never overwritten), temporary file, fsync,
/// rename — and remember it as applied HERE: this box already did what it says.
pub fn write_record(data_root: &Path, r: &Record) -> Result<PathBuf, String> {
let dir = records_dir(data_root);
std::fs::create_dir_all(&dir).map_err(|e| format!("{}: {e}", dir.display()))?;
let name = format!("{}-{}.json", file_safe(&r.repository), r.at_unix_ms);
let at = dir.join(&name);
if at.exists() {
return Err(format!(
"{}: a purge record is never overwritten",
at.display()
));
}
let text = serde_json::to_string_pretty(r).map_err(|e| format!("serialize the record: {e}"))?;
let tmp = dir.join(format!(".{name}.tmp"));
{
use std::io::Write as _;
let mut w = std::fs::File::create(&tmp).map_err(|e| format!("{}: {e}", tmp.display()))?;
w.write_all(text.as_bytes())
.and_then(|_| w.sync_all())
.map_err(|e| format!("{}: {e}", tmp.display()))?;
}
std::fs::rename(&tmp, &at).map_err(|e| format!("{}: {e}", at.display()))?;
let mut applied = applied(data_root);
applied.insert(name);
save_applied(data_root, &applied)?;
Ok(at)
}
/// Every record under the data root, by file name. A file that does not parse is
/// named in the second list rather than skipped in silence.
pub fn records(data_root: &Path) -> (Vec<(String, Record)>, Vec<String>) {
let (mut ok, mut bad) = (Vec::new(), Vec::new());
let Ok(dir) = std::fs::read_dir(records_dir(data_root)) else {
return (ok, bad);
};
for entry in dir.flatten() {
let name = entry.file_name().to_string_lossy().to_string();
if name.starts_with('.') || !name.ends_with(".json") {
continue;
}
match std::fs::read_to_string(entry.path())
.map_err(|e| e.to_string())
.and_then(|t| serde_json::from_str::<Record>(&t).map_err(|e| e.to_string()))
{
Ok(r) => ok.push((name, r)),
Err(e) => bad.push(format!("{RECORDS_DIR}/{name}: {e}")),
}
}
ok.sort_by(|a, b| a.0.cmp(&b.0));
(ok, bad)
}
/// The records this box has applied (by file name).
pub fn applied(data_root: &Path) -> BTreeSet<String> {
std::fs::read_to_string(data_root.join(APPLIED_FILE))
.ok()
.and_then(|t| serde_json::from_str(&t).ok())
.unwrap_or_default()
}
fn save_applied(data_root: &Path, set: &BTreeSet<String>) -> Result<(), String> {
let at = data_root.join(APPLIED_FILE);
let tmp = data_root.join(format!(".{APPLIED_FILE}.tmp"));
let text = serde_json::to_string_pretty(set).map_err(|e| e.to_string())?;
std::fs::write(&tmp, text).map_err(|e| format!("{}: {e}", tmp.display()))?;
std::fs::rename(&tmp, &at).map_err(|e| format!("{}: {e}", at.display()))
}
/// What one pass of [`apply_records`] did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Applied {
/// `(record, repository, artifacts deleted here, kept as newer)` per record
/// applied in this pass.
pub applied: Vec<(String, String, u64, u64)>,
/// Records NOT marked applied, and why — asked again next pass.
pub failures: Vec<String>,
}
impl Applied {
/// The repositories emptied in this pass — whose mirror change map must be
/// forgotten, so a coordinate the main holds again is fetched again.
pub fn repositories(&self) -> BTreeSet<String> {
self.applied.iter().map(|(_, r, _, _)| r.clone()).collect()
}
}
/// **Apply every purge record this box has not applied** (the twin's follow,
/// and a recharge): delete the local artifacts of the record's repository that
/// were stored AT OR BEFORE the record's `at_unix_ms`; keep every artifact
/// stored after it (the tenant came back and uploaded again — that is new
/// data, and a record carried by a recharge must never take it). An artifact
/// whose stored time this backend does not know is a failure, kept, and the
/// record is asked again next pass rather than marked applied over it.
///
/// A record naming a repository this box does not declare is applied as
/// nothing (there is nothing of it here) and marked, so it is not asked again.
pub fn apply_records(
data_root: &Path,
repos: &[(String, Arc<dyn RepositoryBackendTrait>)],
) -> Applied {
let mut out = Applied::default();
let (records, bad) = records(data_root);
out.failures.extend(bad);
let mut done = applied(data_root);
let before = done.len();
for (name, r) in records {
if done.contains(&name) {
continue;
}
let Some((_, backend)) = repos.iter().find(|(n, _)| *n == r.repository) else {
done.insert(name.clone());
out.applied.push((name, r.repository.clone(), 0, 0));
continue;
};
let at = r.at_unix_ms as i64;
let w = walk(
backend.as_ref(),
&|e: &ArtifactEntry| match e.published_unix_ms {
Some(t) if t <= at => Verdict::Delete,
Some(_) => Verdict::Keep,
None => Verdict::Undecided(format!(
"this backend does not say when it was stored, so it cannot be told apart from an \
upload after the purge of {}",
r.at_unix_ms
)),
},
);
if w.failures.is_empty() {
done.insert(name.clone());
out.applied
.push((name, r.repository.clone(), w.deleted, w.kept));
} else {
out.failures.push(format!(
"{RECORDS_DIR}/{name}: {} deleted, {} could not be: {}",
w.deleted,
w.failures.len(),
w.failures.join("; ")
));
}
}
if done.len() != before {
if let Err(e) = save_applied(data_root, &done) {
out.failures
.push(format!("remember the applied records: {e}"));
}
}
out
}
#[cfg(test)]
mod tests {
use super::*;
fn backend(dir: &Path) -> Arc<dyn RepositoryBackendTrait> {
Arc::new(crate::raw::RawBackend::new(
"crates".into(),
dir.to_path_buf(),
))
}
fn id(name: &str) -> ArtifactId {
ArtifactId {
namespace: None,
name: name.into(),
version: String::new(),
}
}
/// Every artifact goes, over MORE than one page, and the bytes are counted.
#[test]
fn the_walk_deletes_every_artifact_over_several_pages() {
let d = tempfile::tempdir().unwrap();
let b = backend(&d.path().join("crates"));
let n = PAGE + 7;
for i in 0..n {
b.put(&id(&format!("a/{i}.bin")), &[1u8; 3]).unwrap();
}
let w = walk(b.as_ref(), &|_| Verdict::Delete);
assert_eq!(
(w.deleted, w.bytes, w.failures.len()),
(n as u64, 3 * n as u64, 0),
"{w:?}"
);
assert!(b.list(None, 0).unwrap().is_empty(), "nothing left");
let again = walk(b.as_ref(), &|_| Verdict::Delete);
assert_eq!(
again,
Walked::default(),
"a second walk of an empty repository is 0, not an error"
);
}
/// A backend that cannot delete answers with EACH artifact in failures.
#[test]
fn a_backend_that_cannot_delete_names_every_artifact() {
struct Immutable(Arc<dyn RepositoryBackendTrait>);
impl RepositoryBackendTrait for Immutable {
fn caps(&self) -> traits::Caps {
traits::Caps {
listing: true,
..traits::Caps::NONE
}
}
fn name(&self) -> &str {
"immutable"
}
fn format(&self) -> traits::ArtifactFormat {
self.0.format()
}
fn is_writable(&self) -> bool {
false
}
fn fetch(&self, id: &ArtifactId) -> anyhow::Result<Option<traits::Bytes>> {
self.0.fetch(id)
}
fn put(&self, _: &ArtifactId, _: &[u8]) -> anyhow::Result<()> {
anyhow::bail!("immutable")
}
fn list(&self, f: Option<&str>, limit: usize) -> anyhow::Result<Vec<ArtifactEntry>> {
self.0.list(f, limit)
}
fn handle_http2_request(
&self,
_: &str,
_: &str,
_: &[u8],
) -> anyhow::Result<(u16, Vec<(String, String)>, Vec<u8>)> {
anyhow::bail!("no door")
}
}
let d = tempfile::tempdir().unwrap();
let inner = backend(&d.path().join("crates"));
for i in 0..5 {
inner.put(&id(&format!("{i}.bin")), b"x").unwrap();
}
let w = walk(&Immutable(inner.clone()), &|_| Verdict::Delete);
assert_eq!(w.deleted, 0);
assert_eq!(w.failures.len(), 5, "{w:?}");
assert!(
w.failures
.iter()
.all(|f| f.contains("does not support Delete")),
"{w:?}"
);
}
#[test]
fn the_gate_is_held_once_and_opens_when_dropped() {
let scope = Path::new("/nonexistent/purge-gate-test");
assert!(!is_closed(scope, "crates"));
let g = close(scope, "crates").expect("closed");
assert!(is_closed(scope, "crates"));
assert!(close(scope, "crates").is_none(), "one walk at a time");
assert!(!is_closed(scope, "other"), "per repository");
drop(g);
assert!(!is_closed(scope, "crates"));
}
/// The twin's rule: stored at or before the purge ⇒ deleted; after ⇒ kept;
/// once applied, a re-pulled record is not applied again.
#[test]
fn a_record_deletes_what_was_stored_before_it_and_keeps_what_came_after_once() {
let d = tempfile::tempdir().unwrap();
let b = backend(&d.path().join("crates"));
b.put(&id("old.bin"), b"old").unwrap();
let listed = b.list(None, 0).unwrap();
let old_at = listed[0].published_unix_ms.expect("raw knows its mtime") as u64;
let rec = Record {
repository: "crates".into(),
at_unix_ms: old_at,
reason: "t".into(),
artifacts: 1,
bytes: 3,
};
// Written on "the main" (another root) and carried here like the estate does.
let main = tempfile::tempdir().unwrap();
let path = write_record(main.path(), &rec).unwrap();
std::fs::create_dir_all(records_dir(d.path())).unwrap();
std::fs::copy(&path, records_dir(d.path()).join(path.file_name().unwrap())).unwrap();
// A NEW upload after the purge: its time is later than the record's.
std::thread::sleep(std::time::Duration::from_millis(20));
b.put(&id("new.bin"), b"new").unwrap();
let repos = vec![("crates".to_string(), b.clone())];
let a = apply_records(d.path(), &repos);
assert!(a.failures.is_empty(), "{a:?}");
assert_eq!(a.applied.len(), 1);
assert_eq!(
(a.applied[0].2, a.applied[0].3),
(1, 1),
"old deleted, new kept"
);
assert!(b.fetch(&id("old.bin")).unwrap().is_none());
assert!(b.fetch(&id("new.bin")).unwrap().is_some());
// Re-applied: nothing — even an old artifact that arrives again is not
// taken by a record already applied.
b.put(&id("old.bin"), b"old").unwrap();
let again = apply_records(d.path(), &repos);
assert!(
again.applied.is_empty() && again.failures.is_empty(),
"{again:?}"
);
assert!(b.fetch(&id("old.bin")).unwrap().is_some());
}
#[test]
fn a_record_is_never_overwritten() {
let d = tempfile::tempdir().unwrap();
let r = Record {
repository: "crates".into(),
at_unix_ms: 5,
reason: "t".into(),
artifacts: 0,
bytes: 0,
};
write_record(d.path(), &r).unwrap();
assert!(write_record(d.path(), &r).is_err());
assert!(
applied(d.path()).contains("crates-5.json"),
"the writer applied it already"
);
}
}