Garbage collection: operation-log retention, pruning with grace period, pack rewriting; oxy gc --now ba160f60219e

//! Garbage collection (spec §21): pack loose objects, then delete objects

//! nothing reachable refers to.

//!

//! Roots: the current view, the views of every operation newer than

//! `core.op-retention`, reference attestations, and the policy. Operation

//! objects are always kept so the log stays walkable, but an operation older

//! than the retention window no longer keeps its states alive, so undoing

//! that far back may fail once its objects are gone.

use super::*;

use crate::store::{PackReport, PruneReport};

use std::collections::HashSet;

#[derive(Debug, Clone, Default, PartialEq, Eq)]

pub struct GcReport {

    pub packed: PackReport,

    pub pruned: PruneReport,

    pub reachable: usize,

}

impl Repo {

    /// Pack loose objects and prune unreachable ones older than the grace

    /// period (`now` ignores the grace period).

    pub fn gc(&self, now: bool) -> Result<GcReport> {

        let packed = self.store.pack_loose()?;

        let keep = self.reachable()?;

        let grace = if now { std::time::Duration::ZERO } else { std::time::Duration::from_secs(self.config.gc_grace_days * 86_400) };

        let pruned = self.store.prune(&keep, grace)?;

        Ok(GcReport { packed, pruned, reachable: keep.len() })

    }

    /// Every object that must survive garbage collection.

    pub fn reachable(&self) -> Result<HashSet<ObjectRef>> {

        let mut keep: HashSet<ObjectRef> = HashSet::new();

        let mut states: Vec<ObjectRef> = Vec::new();

        let mut changes: Vec<ChangeHash> = Vec::new();

        let mut attests: Vec<ObjectRef> = Vec::new();

        let mut policies: Vec<ObjectRef> = Vec::new();

        // Operations: always kept; views within retention are roots.

        let cutoff = Hlc::now().wall_ms.saturating_sub(self.config.op_retention_days * 86_400_000);

        for (i, (r, op)) in self.op_log()?.into_iter().enumerate() {

            keep.insert(r);

            if i == 0 || op.hlc.wall_ms >= cutoff {

                states.extend(op.view.bookmarks.values().copied());

                states.extend(op.view.remotes.values().flat_map(|b| b.values().copied()));

                changes.extend(op.view.workspaces.values().copied());

                policies.extend(op.view.policy);

            }

        }

        // Reference attestations.

        for name in self.view()?.bookmarks.keys() {

            if let Some((r, _)) = self.latest_ref_attest(name)? {

                attests.push(r);

            }

        }

        for st in states {

            if keep.insert(st) {

                changes.extend(self.store.get_state(&st)?.heads);

            }

        }

        for p in policies {

            if keep.insert(p) {

                let policy = crate::policy::Policy::decode(&self.store.get_typed(&p, ObjectType::Policy)?)?;

                for id in &policy.identities {

                    keep.insert(id.hash());

                }

                for sig in self.store.signatures(&p)? {

                    keep.insert(sig.key);

                }

            }

        }

        // Changes: their closure, identities, blobs, chunks, seals.

        let mut stack = changes;

        while let Some(h) = stack.pop() {

            if !keep.insert(h) {

                continue;

            }

            let c = self.store.get_change(&h)?;

            stack.extend(c.deps.iter().copied());

            if !c.author.is_zero() {

                keep.insert(c.author);

            }

            for sig in self.store.signatures(&h)? {

                keep.insert(sig.key);

            }

            for e in &c.edits {

                if let Edit::Replace { old, new, .. } = e {

                    for b in [old, new].into_iter().flatten() {

                        if keep.insert(*b) {

                            if let Ok(bytes) = self.store.get_typed(b, ObjectType::Blob) {

                                keep.extend(Blob::decode(&bytes)?.chunks);

                            }

                        }

                    }

                }

            }

            attests.extend(self.store.seals_for(&h)?);

        }

        for a in attests {

            if keep.insert(a) {

                for sig in self.store.signatures(&a)? {

                    keep.insert(sig.key);

                }

            }

        }

        Ok(keep)

    }

}

    /// `core.op-retention`: days for which operations keep their states reachable (spec §12.1, §21).

    pub op_retention_days: u64,

    /// `core.gc-grace`: days an unreachable object survives, protecting concurrent writers (spec §21).

    pub gc_grace_days: u64,

            op_retention_days: 90,

            gc_grace_days: 14,

            "core.op-retention" => self.op_retention_days = parse_days(value)?,

            "core.gc-grace" => self.gc_grace_days = parse_days(value)?,

/// A count of days, optionally suffixed `d`.

fn parse_days(v: &str) -> Result<u64> {

    v.trim_end_matches('d').parse().map_err(|_| RepoError::Msg(format!("expected a number of days, got {v:?}")))

}

mod gc;

pub use gc::GcReport;

    // Several snapshots leave superseded working-copy versions behind.

    for i in 0..5 {

        write(root, "a.txt", format!("two\n{i}\n").as_bytes());

        repo.snapshot("status").unwrap();

    }

    write(root, "a.txt", b"two\n");

    repo.snapshot("status").unwrap();

    // With the default retention every operation's view is a root: nothing prunes.

    let r = repo.gc(true).unwrap();

    assert_eq!(r.packed.objects, before.len());

    assert_eq!(r.pruned.objects, 0);

    // With zero retention only the current view is a root: superseded

    // working copies go, everything reachable stays.

    repo.config.set("core.op-retention", "0").unwrap();

    repo.save_config().unwrap();

    let r = repo.gc(true).unwrap();

    assert!(r.pruned.objects >= 5, "{r:?}");

    let after = repo.store().list().unwrap();

    assert!(after.len() < before.len());

    assert!(after.contains(&h1) && after.contains(&h2));

    for (_, op) in repo.op_log().unwrap() {

        assert!(!op.command.is_empty(), "operation objects are all kept");

    }

    // Everything still works from the pruned, packed store.

    // Grace period defers pruning of young garbage.

    let r = repo.gc(false).unwrap();

    assert_eq!(r.pruned.objects, 0);

    assert!(r.pruned.deferred > 0);

    let r = repo.gc(true).unwrap();

    assert!(r.pruned.objects > 0);

    assert_eq!(repo.log().unwrap().len(), 4);

/// What `prune` did.

#[derive(Debug, Clone, Default, PartialEq, Eq)]

pub struct PruneReport {

    pub objects: usize,

    pub bytes: u64,

    /// Unreachable objects left alone because they are younger than the grace period.

    pub deferred: usize,

}

        self.write_pack(&loose.iter().map(|(r, _)| *r).collect::<Vec<_>>(), &loose.iter().map(|(_, p)| p.clone()).collect::<Vec<_>>(), &[])

    }

    /// Delete every object not in `keep` (spec §21). Loose objects and packs

    /// younger than `grace` are left alone, since a concurrent writer may not

    /// have recorded its references yet. Packs holding unreachable objects

    /// are rewritten without them. Signature sidecars and seal-index entries

    /// of deleted objects go too.

    pub fn prune(&self, keep: &std::collections::HashSet<ObjectRef>, grace: std::time::Duration) -> Result<PruneReport> {

        let now = std::time::SystemTime::now();

        let old_enough = |m: &fs::Metadata| m.modified().map(|t| now.duration_since(t).unwrap_or_default() >= grace).unwrap_or(true);

        let mut report = PruneReport::default();

        // Loose objects.

        for (r, path) in self.loose_objects()? {

            if keep.contains(&r) {

                continue;

            }

            let meta = fs::metadata(&path)?;

            if !old_enough(&meta) {

                report.deferred += 1;

                continue;

            }

            report.bytes += meta.len();

            report.objects += 1;

            fs::remove_file(&path)?;

            let _ = fs::remove_dir(path.parent().unwrap());

            self.remove_sidecars(&r)?;

        }

        // Packs: rewrite those that hold unreachable objects.

        let mut keep_refs: Vec<ObjectRef> = Vec::new();

        let mut old_paths: Vec<PathBuf> = Vec::new();

        let mut rewrite = false;

        for p in self.packs.borrow().iter() {

            let meta = fs::metadata(&p.path)?;

            let unreachable = p.refs.iter().filter(|r| !keep.contains(r)).count();

            if unreachable == 0 {

                continue;

            }

            if !old_enough(&meta) {

                report.deferred += unreachable;

                continue;

            }

            rewrite = true;

            old_paths.push(p.path.clone());

            for (r, e) in p.refs.iter().zip(p.entries.iter()) {

                if keep.contains(r) {

                    keep_refs.push(*r);

                } else {

                    report.objects += 1;

                    report.bytes += e.stored_len as u64;

                }

            }

        }

        if rewrite {

            let dropped: Vec<ObjectRef> = {

                let packs = self.packs.borrow();

                packs.iter().filter(|p| old_paths.contains(&p.path)).flat_map(|p| p.refs.iter().copied()).filter(|r| !keep.contains(r)).collect()

            };

            self.write_pack(&keep_refs, &[], &old_paths)?;

            for r in dropped {

                self.remove_sidecars(&r)?;

            }

        }

        Ok(report)

    }

    fn remove_sidecars(&self, r: &ObjectRef) -> Result<()> {

        for path in [self.sig_path(r), self.seal_index_path(r)] {

            match fs::remove_file(&path) {

                Ok(()) => {

                    let _ = fs::remove_dir(path.parent().unwrap());

                }

                Err(e) if e.kind() == io::ErrorKind::NotFound => {}

                Err(e) => return Err(e.into()),

            }

        }

        Ok(())

    }

    /// Write `refs` into a new pack, then delete `loose_paths` and `old_packs`.

    fn write_pack(&self, refs: &[ObjectRef], loose_paths: &[PathBuf], old_packs: &[PathBuf]) -> Result<PackReport> {

        if refs.is_empty() {

            for p in old_packs {

                fs::remove_file(p)?;

                fs::remove_file(p.with_extension("idx"))?;

            }

            self.reload_packs()?;

        let mut refs: Vec<ObjectRef> = refs.to_vec();

        refs.sort();

        refs.dedup();

        let mut index: Vec<(ObjectRef, PackEntry)> = Vec::with_capacity(refs.len());

            for r in &refs {

        for p in old_packs {

            fs::remove_file(p)?;

            fs::remove_file(p.with_extension("idx"))?;

        }

        for path in loose_paths {

    fn prune_removes_unreachable_loose_and_packed_objects() {

        use std::collections::HashSet;

        let dir = tempfile::tempdir().unwrap();

        let s = Store::open(dir.path()).unwrap();

        let keep_a = s.put(ObjectType::Chunk, b"keep a").unwrap();

        let drop_a = s.put(ObjectType::Chunk, b"drop a").unwrap();

        s.pack_loose().unwrap();

        let keep_b = s.put(ObjectType::Chunk, b"keep b").unwrap();

        let drop_b = s.put(ObjectType::Chunk, b"drop b").unwrap();

        let sk = crate::identity::SecretKey { ed_seed: [1; 32], ml_seed: [2; 32] };

        let id = sk.identity("a", "b", 0);

        let key = s.put_identity(&id).unwrap();

        s.put_signature(&drop_b, &sk.sign(crate::identity::Domain::Change, &key, &drop_b)).unwrap();

        s.index_seal(&drop_a, &ObjectRef([3; 32])).unwrap();

        let keep: HashSet<ObjectRef> = [keep_a, keep_b, key].into_iter().collect();

        // Everything is too young under a grace period.

        let r = s.prune(&keep, std::time::Duration::from_secs(3600)).unwrap();

        assert_eq!((r.objects, r.deferred), (0, 2));

        let r = s.prune(&keep, std::time::Duration::ZERO).unwrap();

        assert_eq!(r.objects, 2, "{r:?}");

        assert!(s.has(&keep_a) && s.has(&keep_b) && s.has(&key));

        assert!(!s.has(&drop_a) && !s.has(&drop_b));

        assert!(s.signatures(&drop_b).unwrap().is_empty());

        assert!(s.seals_for(&drop_a).unwrap().is_empty());

        assert_eq!(s.pack_count(), 1);

        assert_eq!(s.get(&keep_a).unwrap().1, b"keep a");

        assert_eq!(s.list().unwrap().len(), 3);

    }

    #[test]

    /// Pack loose objects and prune unreachable ones older than the grace period

    Gc {

        /// Ignore the grace period (core.gc-grace) and prune immediately

        #[arg(long)]

        now: bool,

    },

    /// Set a configuration key (author.*, core.sign, core.op-retention, core.gc-grace, workspace.bookmark, remote.<name>.url|token|ca)

        Cmd::Gc { now } => workspace::gc(now),

pub fn gc(now: bool) -> Result<i32> {

    let r = repo.gc(now)?;

    if r.packed.objects > 0 {

            "Packed {}: {} bytes -> {} bytes ({:.0}%).",

            plural(r.packed.objects, "loose object"),

            r.packed.raw_bytes,

            r.packed.packed_bytes,

            100.0 * r.packed.packed_bytes as f64 / r.packed.raw_bytes.max(1) as f64

    println!("{} reachable; pruned {} ({} bytes){}; {} present.",

        plural(r.reachable, "object"),

        plural(r.pruned.objects, "unreachable object"),

        r.pruned.bytes,

        if r.pruned.deferred > 0 { format!("; {} deferred by the grace period (use --now)", plural(r.pruned.deferred, "object")) } else { String::new() },

        plural(repo.store().pack_count(), "pack")

    );