From a9ad2af860df309ff71747ec61b1edafda42facf Mon Sep 17 00:00:00 2001 From: piquel Date: Wed, 4 Nov 2026 18:52:16 +0100 Subject: [PATCH] feat: drop statement payloads past their retention Payloads, such as CI logs, were kept for ever. The `[retention]` table of `policy.toml` keeps those of each kind for a period, counted from when this node stored the statement: `"ci.*" = "90d"`. The most specific rule wins, and the reserved `core.*` kinds, whose payloads core reads, always keep theirs: a rule naming them is refused. `origin gc` drops the payloads past their retention, then compacts the store. Statements stay, and so does a payload another statement still keeps, or that a revision or a tree refers to. Collection waits for the writers holding the store's objects, and reads the index again once they're done, so a writer never refers to a payload deleted under it. --- README.md | 1 + apps/origin-src/src/annotations/mod.rs | 6 +- apps/origin-src/src/annotations/retention.rs | 293 +++++++++++++++++++ apps/origin-src/src/cli.rs | 4 + apps/origin-src/src/conflict.rs | 4 +- apps/origin-src/src/policy.rs | 211 ++++++++++++- apps/origin-src/src/queue.rs | 68 +++++ apps/origin-src/src/targets.rs | 21 ++ apps/origin-src/src/workspace/annotate.rs | 38 ++- apps/origin-src/tests/cli.rs | 22 ++ 10 files changed, 658 insertions(+), 10 deletions(-) create mode 100644 apps/origin-src/src/annotations/retention.rs diff --git a/README.md b/README.md index 73cf8144..20ba3d05 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,7 @@ origin annotate @ review --withdraw -m "Spoke too soon" # take your statement origin annotate --on flaky --neutral -m "Timeout on CI" # a statement about a statement origin statements [] # the statements about a revision, and those about them origin promote [] # carry results over from revisions with the same inputs +origin gc # drop expired payloads (policy.toml) and pack objects origin land [] [--onto ] # land a change and the ones below it origin land --from --onto # land a target's tip on another diff --git a/apps/origin-src/src/annotations/mod.rs b/apps/origin-src/src/annotations/mod.rs index c8074a05..8ef0cb7f 100644 --- a/apps/origin-src/src/annotations/mod.rs +++ b/apps/origin-src/src/annotations/mod.rs @@ -1,5 +1,5 @@ -//! Statements: issuing them, reading them, and promoting them by input -//! digest. +//! Statements: issuing them, reading them, promoting them by input digest, +//! and dropping their payloads once retention ends. //! //! Core interprets only the reserved kinds: [`Kind::ABANDON`], //! [`Kind::CONFLICT`] and [`Kind::DIVERGENT`], and only from the issuers @@ -9,6 +9,7 @@ //! names still stands. mod promotion; +mod retention; use std::collections::HashSet; @@ -21,6 +22,7 @@ use crate::index::Index; use crate::store::{Store, StoreError}; pub use promotion::{Claim, InvalidPromotion}; +pub use retention::Dropped; /// The statements of a store, through its index. #[derive(Debug)] diff --git a/apps/origin-src/src/annotations/retention.rs b/apps/origin-src/src/annotations/retention.rs new file mode 100644 index 00000000..cccfd31c --- /dev/null +++ b/apps/origin-src/src/annotations/retention.rs @@ -0,0 +1,293 @@ +//! Dropping the payloads of statements once their retention ends. + +use std::collections::HashSet; +use std::time::SystemTime; + +use origin_platform::Platform; +use origin_shared::{ObjectId, RecordId}; + +use super::Annotations; +use crate::index::Index; +use crate::policy::Retention; +use crate::store::StoreError; + +/// What garbage collection dropped. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct Dropped { + /// The payloads deleted. + pub payloads: usize, + /// The bytes that freed on disk. + pub bytes: u64, +} + +impl Annotations<'_, P> { + /// Deletes the payloads whose retention ended by `now`, counted from + /// when this node stored their statement. A payload stays while any + /// statement still keeps it, or a revision or a tree refers to the same + /// blob, such as an empty file and an empty payload. Statements stay. + /// It waits for the writers [holding objects](crate::store::Store::hold_objects), + /// which landings on `policy` are too, then reads the rules from + /// `retention` and the index again, so a writer never refers to a + /// payload deleted under it, and a policy landed meanwhile applies. + /// Without rules, it neither waits nor reads anything: a policy landed + /// meanwhile could only have dropped payloads that this run keeps. + /// + /// # Errors + /// + /// Fails if `retention` does, a record cannot be read or a payload + /// deleted. + pub fn drop_expired_payloads>( + &mut self, + retention: impl Fn() -> Result, + now: SystemTime, + ) -> Result { + if retention()?.is_empty() { + return Ok(Dropped::default()); + } + let _collecting = self.store.lock_for_collection()?; + let retention = retention()?; + if retention.is_empty() { + return Ok(Dropped::default()); + } + self.index = Index::open(self.store)?; + let mut expired = HashSet::new(); + let mut live: HashSet = HashSet::new(); + let mut roots = Vec::new(); + for &record in self.index.arrivals() { + match record { + RecordId::Revision(id) => { + let revision = self.store.get(id)?; + roots.push(revision.snapshot); + live.insert(revision.description.into()); + } + RecordId::Acceptance(id) => roots.push(self.store.get(id)?.body().state), + RecordId::Statement(id) => { + let statement = self.store.get(id)?; + if let RecordId::Snapshot(tree) = statement.subject { + roots.push(tree); + } + let Some(payload) = statement.payload else { + continue; + }; + let ended = match retention.period(&statement.kind) { + Some(period) => { + let stored = self.store.stored_at(id.into())?; + now.duration_since(stored).is_ok_and(|age| age >= period) + } + None => false, + }; + if ended { + expired.insert(payload.into()); + } else { + live.insert(payload.into()); + } + } + RecordId::Snapshot(_) => {} + } + } + if expired.is_empty() { + return Ok(Dropped::default()); + } + self.store.reach(roots, &mut live)?; + let mut doomed = Vec::new(); + for &id in expired.difference(&live) { + if self.store.contains(id)? { + doomed.push(id); + } + } + Ok(Dropped { + payloads: doomed.len(), + bytes: self.store.remove(&doomed)?, + }) + } +} + +#[cfg(test)] +mod tests { + use std::thread; + use std::time::Duration; + + use origin_platform::Native; + use origin_shared::{EncodedObject, Verdict}; + + use super::*; + use crate::annotations::tests::{FILES, check, issue, revision, store}; + use crate::store::Store; + + #[test] + fn retention_drops_payloads_and_keeps_statements() { + let (_dir, store) = store(); + let first = revision(&store, 1, &[], &[("empty", "")]); + let with_payload = |kind: &str, payload: &[u8]| { + let payload = EncodedObject::blob(payload); + let mut statement = check(first, kind, Verdict::Pass, "node:ci", None); + statement.payload = Some(payload.id().into()); + let annotations = Annotations::open(&store).unwrap(); + let id = annotations + .issue(&statement, Some(payload.clone())) + .unwrap(); + (id, payload.id()) + }; + let (log_statement, log) = with_payload("ci.test", b"a long log"); + // The same blob as the empty file, which must stay. + let (_, empty) = with_payload("ci.lint", b""); + let (_, review) = with_payload("review", b"looks good"); + let (_, shared) = with_payload("ci.bench", b"shared"); + // Also the payload of a statement whose retention hasn't ended. + let mut kept = check(first, "keep", Verdict::Pass, "node:ci", None); + kept.payload = Some(shared.into()); + issue(&store, &kept); + + let retention: Retention = toml::from_str::( + "[retention]\n\"ci.*\" = \"1d\"\n\"keep\" = \"1000d\"\n", + ) + .unwrap() + .retention; + let mut annotations = Annotations::open(&store).unwrap(); + let now = SystemTime::now(); + assert_eq!( + annotations + .drop_expired_payloads(|| Ok::<_, StoreError>(retention.clone()), now) + .unwrap(), + Dropped::default(), + "not a day old yet" + ); + let later = now + Duration::from_hours(48); + let dropped = annotations + .drop_expired_payloads(|| Ok::<_, StoreError>(retention.clone()), later) + .unwrap(); + assert_eq!(dropped.payloads, 1); + assert!(!store.contains(log).unwrap()); + for stays in [empty, review, shared] { + assert!(store.contains(stays).unwrap(), "{stays}"); + } + assert!(store.get(log_statement).is_ok(), "statements are permanent"); + } + + #[test] + fn without_rules_collection_waits_for_no_writer() { + let (_dir, store) = store(); + let held = store.hold_objects().unwrap(); + let (done, collected) = std::sync::mpsc::channel(); + thread::scope(|scope| { + scope.spawn(|| { + let none = || Ok::<_, StoreError>(Retention::default()); + let dropped = Annotations::open(&store) + .unwrap() + .drop_expired_payloads(none, SystemTime::now()) + .unwrap(); + done.send(dropped).unwrap(); + }); + let finished = collected.recv_timeout(Duration::from_secs(2)); + drop(held); + assert_eq!(finished, Ok(Dropped::default()), "it waited for the writer"); + }); + } + + #[test] + fn collection_waits_for_writers_holding_objects() { + let (_dir, store) = store(); + let first = revision(&store, 1, &[], FILES); + let payload = EncodedObject::blob(b"a log, and a file"); + let mut statement = check(first, "ci.test", Verdict::Pass, "node:ci", None); + statement.payload = Some(payload.id().into()); + Annotations::open(&store) + .unwrap() + .issue(&statement, Some(payload.clone())) + .unwrap(); + let retention = toml::from_str::("[retention]\n\"ci.*\" = \"1s\"\n") + .unwrap() + .retention; + + // A writer finds the expired payload stored as a file it snapshots, + // so it doesn't write it, then writes the revision referring to it. + let held = store.hold_objects().unwrap(); + assert_eq!(store.write_all(std::slice::from_ref(&payload)).unwrap(), 0); + let (done, collected) = std::sync::mpsc::channel(); + thread::scope(|scope| { + scope.spawn(|| { + let later = SystemTime::now() + Duration::from_hours(1); + let dropped = Annotations::open(&store) + .unwrap() + .drop_expired_payloads(|| Ok::<_, StoreError>(retention.clone()), later) + .unwrap(); + done.send(dropped).unwrap(); + }); + thread::sleep(Duration::from_millis(100)); + assert!(collected.try_recv().is_err(), "collection waits"); + revision(&store, 2, &[], &[("log", "a log, and a file")]); + drop(held); + assert_eq!(collected.recv().unwrap(), Dropped::default()); + }); + assert!(store.contains(payload.id()).unwrap()); + } + + #[test] + fn promotions_and_collection_exclude_each_other() { + let (_dir, store) = store(); + let first = revision(&store, 1, &[], FILES); + let payload = EncodedObject::blob(b"a log to promote"); + let inputs = Some((&["src"][..], false)); + let mut result = check(first, "ci.test", Verdict::Pass, "node:ci", inputs); + result.payload = Some(payload.id().into()); + Annotations::open(&store) + .unwrap() + .issue(&result, Some(payload.clone())) + .unwrap(); + let second = RecordId::Revision(revision(&store, 2, &[], FILES)); + + // A promotion refers to the payload without writing it: it waits + // for a collection under way, which then reads it. + let collecting = store.lock_for_collection().unwrap(); + let (done, promoted) = std::sync::mpsc::channel(); + thread::scope(|scope| { + scope.spawn(|| { + let issuer = "node:queue".parse().unwrap(); + let promotions = Annotations::open(&store) + .unwrap() + .promote(second, &issuer) + .unwrap(); + done.send(promotions.len()).unwrap(); + }); + thread::sleep(Duration::from_millis(100)); + assert!(promoted.try_recv().is_err(), "promotion waits"); + drop(collecting); + assert_eq!(promoted.recv().unwrap(), 1); + }); + assert!(store.contains(payload.id()).unwrap()); + } + + #[test] + fn writers_never_trust_a_pack_collection_rewrote() { + let (dir, store) = store(); + let first = revision(&store, 1, &[], FILES); + let payload = EncodedObject::blob(b"expired, then needed again"); + let mut statement = check(first, "ci.test", Verdict::Pass, "node:ci", None); + statement.payload = Some(payload.id().into()); + Annotations::open(&store) + .unwrap() + .issue(&statement, Some(payload.clone())) + .unwrap(); + store.compact().unwrap(); + // Another process, which loaded the pack holding the payload. + let writer = Store::new(dir.path().join(".origin"), Native::default()); + assert!(writer.contains(payload.id()).unwrap()); + + let retention = toml::from_str::("[retention]\n\"ci.*\" = \"1s\"\n") + .unwrap() + .retention; + let later = SystemTime::now() + Duration::from_hours(1); + let dropped = Annotations::open(&store) + .unwrap() + .drop_expired_payloads(|| Ok::<_, StoreError>(retention.clone()), later) + .unwrap(); + assert_eq!(dropped.payloads, 1, "its pack is rewritten without it"); + let _held = writer.hold_objects().unwrap(); + assert_eq!( + writer.write_all(std::slice::from_ref(&payload)).unwrap(), + 1, + "so the writer writes it again" + ); + assert!(store.contains(payload.id()).unwrap()); + } +} diff --git a/apps/origin-src/src/cli.rs b/apps/origin-src/src/cli.rs index b80df439..267a4ece 100644 --- a/apps/origin-src/src/cli.rs +++ b/apps/origin-src/src/cli.rs @@ -128,6 +128,9 @@ enum WorkspaceCommand { #[arg(default_value = "@")] rev: String, }, + /// Drop the payloads past the retention set in `policy.toml`, pack + /// small objects, and remove the leftovers of crashed writes. + Gc, /// Land a change on a target, with the changes below it the target /// doesn't have yet. The queue rebases it onto the target's tip, checks /// the evidence `policy.toml` requires, and if it's there, appends a @@ -402,6 +405,7 @@ impl WorkspaceCommand { } Self::Statements { rev } => workspace.statements(&rev)?, Self::Promote { rev } => workspace.promote(&rev)?, + Self::Gc => workspace.gc()?, Self::Land(args) => match (args.from, args.onto) { _ if args.status => workspace.land_status()?, (Some(from), Some(onto)) => workspace.land_target(&from, &onto)?, diff --git a/apps/origin-src/src/conflict.rs b/apps/origin-src/src/conflict.rs index 4367cad4..f07fbc14 100644 --- a/apps/origin-src/src/conflict.rs +++ b/apps/origin-src/src/conflict.rs @@ -178,8 +178,8 @@ impl Conflicts { } /// The conflicts a `core.conflict` statement lists. Its payload is the - /// list, which a repository always keeps, so one it lacks fails rather - /// than reading as no conflicts. + /// list, which retention keeps, as it does every reserved kind's, so + /// one the repository lacks fails rather than reading as no conflicts. /// /// # Errors /// diff --git a/apps/origin-src/src/policy.rs b/apps/origin-src/src/policy.rs index 61a67b6f..6b849c95 100644 --- a/apps/origin-src/src/policy.rs +++ b/apps/origin-src/src/policy.rs @@ -1,6 +1,7 @@ //! `policy.toml`, in the snapshot of the `policy` target: the nodes and -//! groups of a repository, and who orders each target and what landing on -//! it requires. The policy in force is the one at the target's head. +//! groups of a repository, who orders each target and what landing on it +//! requires, and how long payloads are kept. The policy in force is the one +//! at the target's head. //! //! ```toml //! [nodes] # each node's ed25519 public key, in hex @@ -17,6 +18,9 @@ //! { kind = "review", verdict = "pass", issuer = "group:maintainers" }, //! ] //! +//! [retention] +//! "ci.*" = "90d" +//! //! [[paths]] //! path = "secrets" //! grants = { all = "none", "group:maintainers" = "write" } @@ -30,6 +34,7 @@ use std::collections::BTreeMap; use std::fmt; use std::str::FromStr; +use std::time::Duration; use origin_platform::Platform; use origin_shared::{ @@ -47,6 +52,8 @@ pub struct Policy { nodes: BTreeMap, groups: BTreeMap>, targets: BTreeMap, + /// How long payloads are kept. + pub retention: Retention, paths: Vec, } @@ -257,6 +264,8 @@ struct PolicyFile { #[serde(default)] targets: BTreeMap, #[serde(default)] + retention: Retention, + #[serde(default)] paths: Vec, } @@ -327,6 +336,7 @@ impl TryFrom for Policy { nodes, groups, targets: BTreeMap::new(), + retention: file.retention, paths: Vec::new(), }; for (name, target) in file.targets { @@ -482,6 +492,140 @@ pub enum RuleError { UnknownGroup(PrincipalName), } +/// How long the payloads of each kind of statement are kept, measured from +/// when this node received the statement: `"ci.*" = "90d"` keeps those of +/// every `ci.` kind for 90 days. A rule names a kind, a prefix ending in +/// `.*`, or `*` for every kind, and the most specific rule that matches +/// wins. Kinds no rule matches keep their payloads, and so do the reserved +/// `core.*` kinds, which core reads, whatever `*` says: a `core.conflict`'s +/// lists its conflicts. A rule naming them is refused. +/// +/// Statements themselves are permanent: only their payloads expire. +#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize)] +#[serde(try_from = "BTreeMap")] +pub struct Retention { + rules: Vec<(KindPattern, Duration)>, +} + +impl Retention { + /// How long payloads of `kind` are kept, or `None` for ever. + #[must_use] + pub fn period(&self, kind: &Kind) -> Option { + if kind.is_reserved() { + return None; + } + self.rules + .iter() + .filter_map(|(pattern, period)| Some((pattern.specificity(kind)?, *period))) + .max_by_key(|&(specificity, _)| specificity) + .map(|(_, period)| period) + } + + /// Whether no payload ever expires. + #[must_use] + pub fn is_empty(&self) -> bool { + self.rules.is_empty() + } + + /// Parses a period: a whole number of seconds (`s`), minutes (`m`), + /// hours (`h`) or days (`d`), such as `90d`. + fn parse_period(period: &str) -> Option { + let unit = match period.chars().last()? { + 's' => 1, + 'm' => 60, + 'h' => 60 * 60, + 'd' => 24 * 60 * 60, + _ => return None, + }; + let count: u64 = period[..period.len() - 1].parse().ok()?; + Some(Duration::from_secs(count.checked_mul(unit)?)) + } +} + +impl TryFrom> for Retention { + type Error = RetentionError; + + fn try_from(table: BTreeMap) -> Result { + let rules = table + .into_iter() + .map(|(pattern, period)| { + let duration = Self::parse_period(&period) + .ok_or_else(|| RetentionError::Period(period.clone()))?; + let parsed: KindPattern = pattern.parse()?; + if parsed.is_reserved() { + return Err(RetentionError::Reserved(pattern)); + } + Ok((parsed, duration)) + }) + .collect::>()?; + Ok(Self { rules }) + } +} + +/// The kinds a retention rule applies to. +#[derive(Clone, Debug, PartialEq, Eq)] +enum KindPattern { + /// `*`: every kind. + Any, + /// `ci.*`: the kinds under a prefix, not the prefix itself. + Prefix(Kind), + /// `ci.test`: one kind. + Exact(Kind), +} + +impl KindPattern { + /// Whether the pattern only matches reserved kinds: `core.*` or one + /// of them. + fn is_reserved(&self) -> bool { + match self { + Self::Any => false, + Self::Prefix(prefix) => prefix.as_str() == "core" || prefix.is_reserved(), + Self::Exact(kind) => kind.is_reserved(), + } + } + + /// How specifically the pattern matches `kind`, if it does: longer + /// prefixes beat shorter ones, and the kind itself beats them all. + fn specificity(&self, kind: &Kind) -> Option<(usize, bool)> { + match self { + Self::Any => Some((0, false)), + Self::Prefix(prefix) => kind + .as_str() + .strip_prefix(prefix.as_str()) + .is_some_and(|rest| rest.starts_with('.')) + .then_some((prefix.as_str().len(), false)), + Self::Exact(exact) => (exact == kind).then_some((exact.as_str().len(), true)), + } + } +} + +impl FromStr for KindPattern { + type Err = RetentionError; + + fn from_str(pattern: &str) -> Result { + let invalid = || RetentionError::Pattern(pattern.to_owned()); + match pattern.strip_suffix(".*") { + _ if pattern == "*" => Ok(Self::Any), + Some(prefix) => Kind::new(prefix).map(Self::Prefix).map_err(|_| invalid()), + None => Kind::new(pattern).map(Self::Exact).map_err(|_| invalid()), + } + } +} + +/// An invalid `[retention]` rule. +#[derive(Debug, PartialEq, Eq, thiserror::Error)] +pub enum RetentionError { + /// The key is not a kind, a prefix ending in `.*`, or `*`. + #[error("`{0}` is not a statement kind, a prefix like `ci.*`, or `*`")] + Pattern(String), + /// The value is not a period. + #[error("`{0}` is not a period like `90d`, `12h`, `30m` or `45s`")] + Period(String), + /// The rule names reserved kinds, whose payloads are kept. + #[error("`{0}` names reserved kinds, whose payloads core reads and always keeps")] + Reserved(String), +} + /// A failure to read the policy. #[derive(Debug, thiserror::Error)] pub enum PolicyError { @@ -516,6 +660,45 @@ pub enum PolicyError { mod tests { use super::*; + fn retention(toml: &str) -> Result { + toml::from_str::(toml).map(|policy| policy.retention) + } + + fn kind(kind: &str) -> Kind { + Kind::new(kind).unwrap() + } + + #[test] + fn the_most_specific_rule_wins() { + let retention = retention( + "[retention]\n\"*\" = \"365d\"\n\"ci.*\" = \"90d\"\n\ + \"ci.bench.*\" = \"12h\"\n\"ci.bench\" = \"30m\"\n", + ) + .unwrap(); + let days = |days: u64| Some(Duration::from_hours(days * 24)); + assert_eq!(retention.period(&kind("ci.test")), days(90)); + assert_eq!( + retention.period(&kind("ci.bench.x")), + Some(Duration::from_hours(12)) + ); + assert_eq!( + retention.period(&kind("ci.bench")), + Some(Duration::from_mins(30)), + "the exact kind, not its children's prefix" + ); + assert_eq!(retention.period(&kind("cix")), days(365), "whole segments"); + assert_eq!(retention.period(&kind("review")), days(365)); + assert_eq!(retention.period(&kind("core.conflict")), None); + } + + #[test] + fn reserved_kinds_keep_their_payloads() { + for rule in ["core.*", "core.conflict"] { + let error = retention(&format!("[retention]\n\"{rule}\" = \"1d\"\n")).unwrap_err(); + assert!(error.message().contains("reserved kinds"), "{error}"); + } + } + /// A policy with every table, whose node `hosted` has the key of the /// secret `[42; 32]`. fn full_policy() -> String { @@ -618,4 +801,28 @@ mod tests { assert!(error.contains(expected), "{error}"); } } + + #[test] + fn without_rules_payloads_stay() { + let retention = retention("[groups]\nrunners = []\n").unwrap(); + assert!(retention.is_empty()); + assert_eq!(retention.period(&kind("ci.test")), None); + } + + #[test] + fn rejects_invalid_rules() { + for invalid in [ + "\"ci.*\" = \"90\"", + "\"ci.*\" = \"d\"", + "\"ci.*\" = \"3y\"", + "\"CI\" = \"1d\"", + "\"ci.\" = \"1d\"", + "\"*.ci\" = \"1d\"", + ] { + assert!( + retention(&format!("[retention]\n{invalid}\n")).is_err(), + "{invalid}" + ); + } + } } diff --git a/apps/origin-src/src/queue.rs b/apps/origin-src/src/queue.rs index 0696027f..eeddc1a5 100644 --- a/apps/origin-src/src/queue.rs +++ b/apps/origin-src/src/queue.rs @@ -1408,6 +1408,74 @@ mod tests { )); } + #[test] + fn landings_wait_for_collection() { + let repo = Repo::new("[targets.main]\nauthority = \"node:queue\"\n"); + repo.change_on_main(1, &[("a", "a")]); + // Its promotions refer to payloads they don't write. + let collecting = repo.store.lock_for_collection().unwrap(); + let (done, landed) = std::sync::mpsc::channel(); + thread::scope(|scope| { + scope.spawn(|| done.send(repo.land(1)).unwrap()); + thread::sleep(std::time::Duration::from_millis(100)); + assert!(landed.try_recv().is_err(), "the landing waits"); + drop(collecting); + assert!(matches!( + landed.recv().unwrap()[..], + [Landing::Landed { .. }] + )); + }); + } + + #[test] + fn collection_applies_a_policy_landed_while_it_waits() { + let rules = |retention: &str| { + format!( + "[targets.main]\nauthority = \"node:queue\"\n[retention]\n\"ci.*\" = \"{retention}\"\n" + ) + }; + let repo = Repo::new(&rules("1h")); + let subject = repo.change_on_main(1, &[("a", "a")]); + let payload = EncodedObject::blob(b"a log"); + let statement = Statement { + payload: Some(payload.id().into()), + ..ci_test(subject, "node:ci") + }; + Annotations::open(&repo.store) + .unwrap() + .issue(&statement, Some(payload.clone())) + .unwrap(); + + // Collection starts, two hours on, under a writer holding objects; + // meanwhile a policy lands that keeps the payload longer. + let held = repo.store.hold_objects().unwrap(); + let (done, collected) = std::sync::mpsc::channel(); + thread::scope(|scope| { + scope.spawn(|| { + let later = std::time::SystemTime::now() + std::time::Duration::from_hours(2); + let retention = || Ok::<_, PolicyError>(Policy::load(&repo.store)?.retention); + let dropped = Annotations::open(&repo.store) + .unwrap() + .drop_expired_payloads(retention, later) + .unwrap(); + done.send(dropped).unwrap(); + }); + thread::sleep(std::time::Duration::from_millis(100)); + assert!(collected.try_recv().is_err(), "collection waits"); + let tip = Base::Acceptance(repo.tip(&TargetName::POLICY).acceptance); + let longer = repo.policy_toml(&rules("1000d")); + repo.revision(2, &[], tip, &[("policy.toml", &longer)]); + let landed = repo + .queue() + .land_change(ChangeId::from_bytes([2; 16]), &TargetName::POLICY) + .unwrap(); + assert!(matches!(landed[..], [Landing::Landed { .. }]), "{landed:?}"); + drop(held); + assert_eq!(collected.recv().unwrap().payloads, 0); + }); + assert!(repo.store.contains(payload.id()).unwrap()); + } + #[test] fn policy_landings_keep_a_valid_policy() { let repo = Repo::new("[targets.main]\nauthority = \"node:queue\"\n"); diff --git a/apps/origin-src/src/targets.rs b/apps/origin-src/src/targets.rs index a41eba14..efb01174 100644 --- a/apps/origin-src/src/targets.rs +++ b/apps/origin-src/src/targets.rs @@ -255,6 +255,11 @@ impl Store

{ if body.target != target.id() { return Err(StoreError::WrongTarget(target.clone())); } + // A new policy changes which payloads collection keeps, so it never + // lands while collection, which reads the policy first, runs. + let _held = (*target == TargetName::POLICY) + .then(|| self.hold_objects()) + .transpose()?; let encoded = acceptance.encode(); // The acceptance is durable before the log refers to it. self.write_all(slice::from_ref(&encoded))?; @@ -400,6 +405,22 @@ mod tests { assert_eq!(store.targets().unwrap(), [main]); } + #[test] + fn policy_heads_wait_for_collection() { + let (_dir, store) = store(); + let collecting = store.lock_for_collection().unwrap(); + // Other targets don't. + land(&store, &TargetName::MAIN, 1); + let (done, landed) = std::sync::mpsc::channel(); + thread::scope(|scope| { + scope.spawn(|| done.send(land(&store, &TargetName::POLICY, 1)).unwrap()); + thread::sleep(std::time::Duration::from_millis(100)); + assert!(landed.try_recv().is_err(), "the policy waits"); + drop(collecting); + assert_eq!(landed.recv().unwrap().len(), 1); + }); + } + #[test] fn appends_must_follow_the_latest_acceptance() { let (_dir, store) = store(); diff --git a/apps/origin-src/src/workspace/annotate.rs b/apps/origin-src/src/workspace/annotate.rs index 164a4972..0082e5ed 100644 --- a/apps/origin-src/src/workspace/annotate.rs +++ b/apps/origin-src/src/workspace/annotate.rs @@ -1,10 +1,11 @@ -//! The statement commands: `annotate`, `statements` and `promote`, +//! The statement commands: `annotate`, `statements`, `promote` and `gc`, //! and flagging divergence when a revision is saved. use std::collections::HashSet; use std::fs::File; use std::io; use std::path::{Path, PathBuf}; +use std::time::SystemTime; use origin_shared::{ BlobId, EncodedObject, Inputs, Kind, Principal, RecordId, Revision, RevisionId, Statement, @@ -14,6 +15,7 @@ use origin_shared::{ use super::{WorkingCopy, Workspace, WorkspaceError}; use crate::annotations::Annotations; use crate::index::{FindError, Index}; +use crate::policy::Policy; use crate::store::StoreError; use crate::tree::Snapshots; @@ -300,11 +302,12 @@ impl Workspace { Ok(()) } - /// A payload's first line if it's text, its size otherwise, or that - /// this repository doesn't have it. + /// A payload's first line if it's text, its size otherwise, or that this + /// repository doesn't have it: retention may have dropped it, or it + /// never came here. fn payload_summary(&self, payload: BlobId) -> Result { let Some(content) = self.store.try_blob(payload.into())? else { - return Ok("not in this repository".to_owned()); + return Ok("unavailable here".to_owned()); }; Ok( match str::from_utf8(&content).map(|text| text.lines().next()) { @@ -347,4 +350,31 @@ impl Workspace { } Ok(()) } + + /// `origin gc`: drops the payloads past the retention `policy.toml` + /// sets, packs the small loose objects, and removes the temporary files + /// crashed writes left behind. Statements are never dropped. + /// + /// # Errors + /// + /// Fails if the policy is invalid, or the store cannot be read, + /// compacted or cleaned. + pub fn gc(&self) -> Result<(), WorkspaceError> { + // The policy in force once collection may start, not before. + let retention = || Ok::<_, WorkspaceError>(Policy::load(&self.store)?.retention); + let dropped = + Annotations::open(&self.store)?.drop_expired_payloads(retention, SystemTime::now())?; + println!( + "Dropped {} payloads past their retention, freeing {} bytes", + dropped.payloads, dropped.bytes + ); + match self.store.gc()? { + Some(compaction) => println!( + "Packed {} loose objects, merging {} packs", + compaction.packed, compaction.merged + ), + None => println!("Another process is compacting the store"), + } + Ok(()) + } } diff --git a/apps/origin-src/tests/cli.rs b/apps/origin-src/tests/cli.rs index 34fe0e8f..17bb0a99 100644 --- a/apps/origin-src/tests/cli.rs +++ b/apps/origin-src/tests/cli.rs @@ -759,6 +759,24 @@ fn statements_about_statements_show_under_them() { ); } +#[test] +fn a_payload_the_repository_lacks_is_unavailable() { + let repo = Repo::new(); + repo.ok(&["annotate", "@", "review", "--pass", "-m", "Looks good"]); + let hex = EncodedObject::blob(b"Looks good").id().to_hex(); + fs::remove_file(repo.path(&format!( + ".origin/objects/loose/{}/{}", + &hex[..2], + &hex[2..] + ))) + .unwrap(); + let listed = repo.ok(&["statements"]); + assert!( + listed.contains(" payload: unavailable here\n"), + "{listed}" + ); +} + #[test] fn annotating_a_statement_needs_one_and_no_inputs() { let repo = Repo::new(); @@ -921,6 +939,10 @@ fn statements_are_promoted_where_their_inputs_are_unchanged() { .contains("is a reserved kind") ); assert!(repo.fails(&["annotate", "@", "ci.lint"]).contains("--pass")); + assert!( + repo.ok(&["gc"]) + .starts_with("Dropped 0 payloads past their retention") + ); } #[test]