From e836bb9787582c3a6868274ad01394be3e73b7ba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Fri, 17 Jul 2026 20:10:50 -0700 Subject: [PATCH 1/3] feat(policy): evaluate retention plans Add an ecosystem-neutral retention planner. A policy holds ordered keep and expire rule groups; the engine evaluates one project's candidates and returns a deterministic decision per artifact without reading storage. A keep rule wins over an expire rule, matching Google Artifact Registry cleanup precedence, and each decision names the rule that decided it. Candidates carry a caller-assigned version rank, so the engine orders keep-latest and output without naming a package format. Decisions sort by rank, artifact, then digest, so repeating an evaluation yields byte-identical output. A criterion bench plans a generated million-record repository to hold evaluation linearithmic. Refs #481 --- Cargo.lock | 2 + crates/peryx-policy/Cargo.toml | 8 + crates/peryx-policy/benches/retention.rs | 86 +++++ crates/peryx-policy/src/lib.rs | 7 + crates/peryx-policy/src/retention.rs | 308 ++++++++++++++++ crates/peryx-policy/tests/retention.rs | 434 +++++++++++++++++++++++ 6 files changed, 845 insertions(+) create mode 100644 crates/peryx-policy/benches/retention.rs create mode 100644 crates/peryx-policy/src/retention.rs create mode 100644 crates/peryx-policy/tests/retention.rs diff --git a/Cargo.lock b/Cargo.lock index 8e0f99e2..0d43254b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3945,7 +3945,9 @@ dependencies = [ name = "peryx-policy" version = "0.0.1" dependencies = [ + "codspeed-criterion-compat", "serde", + "serde_json", ] [[package]] diff --git a/crates/peryx-policy/Cargo.toml b/crates/peryx-policy/Cargo.toml index 96220f1d..bc9e897e 100644 --- a/crates/peryx-policy/Cargo.toml +++ b/crates/peryx-policy/Cargo.toml @@ -10,5 +10,13 @@ description = "Ecosystem-neutral index policy engine for peryx: name allow/deny, [dependencies] serde.workspace = true +[dev-dependencies] +criterion.workspace = true +serde_json.workspace = true + +[[bench]] +name = "retention" +harness = false + [lints] workspace = true diff --git a/crates/peryx-policy/benches/retention.rs b/crates/peryx-policy/benches/retention.rs new file mode 100644 index 00000000..77bcee34 --- /dev/null +++ b/crates/peryx-policy/benches/retention.rs @@ -0,0 +1,86 @@ +#![allow( + clippy::significant_drop_tightening, + reason = "criterion_group! expands to a temporary flagged by this nursery lint" +)] + +//! How retention evaluation scales over a large index. +//! +//! The design claim is that a plan costs no more than sorting each project's files: the planner groups +//! one project at a time, sorts its candidates, and classifies each in one pass, so a million-record +//! index stays linearithmic rather than quadratic. This leg feeds the planner a generated repository of +//! one million artifact records — fifty thousand projects of twenty versions each — and measures one +//! full evaluation. +//! +//! The CI performance runner builds this leg; it establishes a baseline rather than gating a threshold. + +use criterion::{Criterion, Throughput, black_box, criterion_group, criterion_main}; +use peryx_policy::{ + RetentionCandidate, RetentionClass, RetentionConfig, RetentionOutcome, RetentionPolicy, RetentionSelector, + RetentionVisibility, +}; + +const RECORDS: u64 = 1_000_000; +const VERSIONS_PER_PROJECT: u64 = 20; + +fn candidate(project: u64, version: u64) -> RetentionCandidate { + RetentionCandidate { + project: format!("project-{project}"), + version: Some(format!("{version}.0")), + artifact: format!("project-{project}-{version}.0.whl"), + digest: format!("sha256:{project}-{version}"), + class: if version.is_multiple_of(7) { + RetentionClass::Trash + } else { + RetentionClass::Hosted + }, + visibility: RetentionVisibility::Active, + source: None, + bytes: 4096, + upload_time_unix: Some(version.cast_signed()), + // Ascending rank against descending version, so the planner always sorts. + rank: VERSIONS_PER_PROJECT - version - 1, + orphan: false, + } +} + +fn repository() -> Vec> { + (0..RECORDS / VERSIONS_PER_PROJECT) + .map(|project| { + (0..VERSIONS_PER_PROJECT) + .map(|version| candidate(project, version)) + .collect() + }) + .collect() +} + +fn policy() -> RetentionPolicy { + RetentionPolicy::compile(&RetentionConfig { + keep: vec![RetentionSelector::KeepLatest { count: 5 }], + expire: vec![ + RetentionSelector::Trash, + RetentionSelector::Age { older_than_seconds: 10 }, + ], + }) +} + +fn bench_plan(criterion: &mut Criterion) { + let repository = repository(); + let policy = policy(); + let mut group = criterion.benchmark_group("retention"); + group.throughput(Throughput::Elements(RECORDS)); + group.bench_function("plan_one_million_records", |bencher| { + bencher.iter(|| { + let mut removed = 0_u64; + for project in &repository { + for decision in policy.plan_project(Some(1_000), black_box(project.clone())) { + removed += u64::from(decision.outcome == RetentionOutcome::Remove); + } + } + black_box(removed) + }); + }); + group.finish(); +} + +criterion_group!(benches, bench_plan); +criterion_main!(benches); diff --git a/crates/peryx-policy/src/lib.rs b/crates/peryx-policy/src/lib.rs index 97ceb579..6a015a2c 100644 --- a/crates/peryx-policy/src/lib.rs +++ b/crates/peryx-policy/src/lib.rs @@ -12,6 +12,13 @@ use std::sync::Arc; use serde::{Deserialize, Serialize}; +mod retention; + +pub use retention::{ + RetentionCandidate, RetentionClass, RetentionConfig, RetentionDecision, RetentionFrontier, RetentionOutcome, + RetentionPolicy, RetentionSelector, RetentionSummary, RetentionVisibility, +}; + /// The ecosystem-neutral policy keys. A driver parses its own format-specific keys separately and /// compiles them into [`ArtifactRule`]s. #[derive(Debug, Default, Clone, PartialEq, Eq, Deserialize)] diff --git a/crates/peryx-policy/src/retention.rs b/crates/peryx-policy/src/retention.rs new file mode 100644 index 00000000..b695a064 --- /dev/null +++ b/crates/peryx-policy/src/retention.rs @@ -0,0 +1,308 @@ +//! Ecosystem-neutral retention-plan evaluation. +//! +//! A retention policy names two ordered rule groups: `keep` rules protect artifacts, `expire` rules +//! mark them for removal. The engine evaluates one repository's [`RetentionCandidate`] stream against +//! the compiled policy and returns a deterministic [`RetentionDecision`] per artifact without touching +//! storage or blobs. An ecosystem crate adapts its own records (a `PyPI` upload, an `OCI` tag) into +//! neutral candidates and hands them here one project at a time, so a large repository never +//! materializes as one in-memory plan. +//! +//! A keep rule always wins over an expire rule (the precedence +//! [Google Artifact Registry cleanup policies](https://cloud.google.com/artifact-registry/docs/repositories/cleanup-policy) +//! define), and the decision names the rule that decided it. Version ordering is the caller's: +//! candidates carry a [`rank`](RetentionCandidate::rank) an ecosystem assigns (`PyPI` through +//! PEP 440), so this crate names no package format. + +use std::collections::BTreeSet; + +use serde::{Deserialize, Serialize}; + +/// The storage lifecycle a candidate belongs to, mirroring the storage accounting classes. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum RetentionClass { + Hosted, + Cached, + Generated, + Trash, +} + +/// A candidate's logical visibility, recorded on the decision so an operator sees what a removal hides. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum RetentionVisibility { + Active, + Yanked, + Hidden, +} + +/// One retention rule. +/// +/// A rule appears in a policy's `keep` or `expire` group and matches a candidate by exactly one +/// dimension. The engine never assigns a rule its group's meaning: the same rule protects in `keep` +/// and removes in `expire`. +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(tag = "selector", rename_all = "kebab-case")] +pub enum RetentionSelector { + /// Match a candidate whose publish time is at least `older_than_seconds` before the evaluation + /// clock. A candidate with no publish time, or an evaluation with no clock, never matches, so the + /// engine ages nothing it cannot date. + Age { older_than_seconds: i64 }, + /// Match a candidate routed from the named source. + Source { name: String }, + /// Match a candidate whose project name begins with `prefix`. + ProjectPrefix { prefix: String }, + /// Match a candidate among the newest `count` versions of its project, by the caller's rank. + KeepLatest { count: u64 }, + /// Match a cached candidate. + Cached, + /// Match a soft-deleted (trash) candidate. + Trash, + /// Match a candidate whose content no live reference reaches. + Orphan, +} + +impl RetentionSelector { + /// The stable rule name a decision records. + #[must_use] + pub const fn name(&self) -> &'static str { + match self { + Self::Age { .. } => "age", + Self::Source { .. } => "source", + Self::ProjectPrefix { .. } => "project-prefix", + Self::KeepLatest { .. } => "keep-latest", + Self::Cached => "cached", + Self::Trash => "trash", + Self::Orphan => "orphan", + } + } + + fn matches(&self, candidate: &RetentionCandidate, now: Option) -> bool { + match self { + Self::Age { older_than_seconds } => { + matches!((now, candidate.upload_time_unix), (Some(now), Some(uploaded)) if now - uploaded >= *older_than_seconds) + } + Self::Source { name } => candidate.source.as_deref() == Some(name.as_str()), + Self::ProjectPrefix { prefix } => candidate.project.starts_with(prefix.as_str()), + Self::KeepLatest { count } => candidate.rank < *count, + Self::Cached => candidate.class == RetentionClass::Cached, + Self::Trash => candidate.class == RetentionClass::Trash, + Self::Orphan => candidate.orphan, + } + } +} + +/// One artifact an ecosystem adapts from its own records for evaluation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RetentionCandidate { + pub project: String, + pub version: Option, + /// The artifact identity within its version: a `PyPI` filename or an `OCI` tag. + pub artifact: String, + pub digest: String, + pub class: RetentionClass, + pub visibility: RetentionVisibility, + /// The routed source, when the record names one; matched by [`RetentionSelector::Source`]. + pub source: Option, + /// The candidate's estimated physical bytes. + pub bytes: u64, + pub upload_time_unix: Option, + /// The candidate's version position within its project, newest first (`0` is newest). The + /// ecosystem assigns it; [`RetentionSelector::KeepLatest`] and the output order read it. + pub rank: u64, + pub orphan: bool, +} + +/// The operator's retention rules, as loaded from configuration. +#[derive(Debug, Clone, Default, PartialEq, Eq, Deserialize)] +#[serde(default)] +pub struct RetentionConfig { + pub keep: Vec, + pub expire: Vec, +} + +/// The compiled retention policy. Compiling records a content [`version`](RetentionPolicy::version) so +/// two runs of the same rules produce the same identity on their plans. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RetentionPolicy { + keep: Vec, + expire: Vec, + version: u64, +} + +impl RetentionPolicy { + /// Compile the operator's rules once. + #[must_use] + pub fn compile(config: &RetentionConfig) -> Self { + Self { + version: policy_version(config), + keep: config.keep.clone(), + expire: config.expire.clone(), + } + } + + /// The policy's content identity: equal rules compile to an equal version, distinct rules to a + /// distinct one, through a stable FNV-1a hash of the rules' canonical form. + #[must_use] + pub const fn version(&self) -> u64 { + self.version + } + + /// Whether the policy has no rules, so evaluation would retain everything. + #[must_use] + pub const fn is_empty(&self) -> bool { + self.keep.is_empty() && self.expire.is_empty() + } + + /// Evaluate one project's candidates against the policy. + /// + /// The returned decisions are ordered deterministically (newest version first by rank, then + /// artifact, then digest), so repeating a call over the same candidates yields byte-identical + /// output. Each removal decision records the surviving versions of the same project as its + /// retained alternatives. `now` is the evaluation clock an age rule ages against. + #[must_use] + pub fn plan_project(&self, now: Option, mut candidates: Vec) -> Vec { + candidates.sort_by(|left, right| { + (left.rank, &left.artifact, &left.digest).cmp(&(right.rank, &right.artifact, &right.digest)) + }); + let mut decisions: Vec = candidates + .into_iter() + .map(|candidate| { + let (outcome, rule) = self.classify(&candidate, now); + RetentionDecision { + project: candidate.project, + version: candidate.version, + artifact: candidate.artifact, + digest: candidate.digest, + class: candidate.class, + visibility: candidate.visibility, + source: candidate.source, + bytes: candidate.bytes, + outcome, + rule, + retained_alternatives: Vec::new(), + } + }) + .collect(); + let retained: BTreeSet = decisions + .iter() + .filter(|decision| decision.outcome == RetentionOutcome::Retain) + .filter_map(|decision| decision.version.clone()) + .collect(); + for decision in &mut decisions { + if decision.outcome == RetentionOutcome::Remove { + decision.retained_alternatives = retained + .iter() + .filter(|version| Some(version.as_str()) != decision.version.as_deref()) + .cloned() + .collect(); + } + } + decisions + } + + fn classify(&self, candidate: &RetentionCandidate, now: Option) -> (RetentionOutcome, Option<&'static str>) { + if let Some(rule) = self.keep.iter().find(|rule| rule.matches(candidate, now)) { + return (RetentionOutcome::Retain, Some(rule.name())); + } + if let Some(rule) = self.expire.iter().find(|rule| rule.matches(candidate, now)) { + return (RetentionOutcome::Remove, Some(rule.name())); + } + (RetentionOutcome::Retain, None) + } +} + +/// Whether a candidate is retained or eligible for removal. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum RetentionOutcome { + Retain, + Remove, +} + +/// One artifact's evaluated outcome. A removal decision estimates the reclaimable `bytes` and lists +/// the project versions a caller would still serve after it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub struct RetentionDecision { + pub project: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub version: Option, + pub artifact: String, + pub digest: String, + pub class: RetentionClass, + pub visibility: RetentionVisibility, + #[serde(skip_serializing_if = "Option::is_none")] + pub source: Option, + pub bytes: u64, + pub outcome: RetentionOutcome, + #[serde(skip_serializing_if = "Option::is_none")] + pub rule: Option<&'static str>, + #[serde(skip_serializing_if = "Vec::is_empty")] + pub retained_alternatives: Vec, +} + +/// The metadata snapshot a plan evaluated, so stale input is rejectable later. It mirrors the storage +/// policy-input generation an adapter reads. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)] +pub struct RetentionFrontier { + pub repository: u64, + pub catalog: u64, + pub policy: u64, +} + +/// A plan's identity header: the policy that produced it and the metadata snapshot it read. An adapter +/// emits it once alongside the streamed decisions. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct RetentionSummary { + pub policy_version: u64, + pub frontier: RetentionFrontier, +} + +fn policy_version(config: &RetentionConfig) -> u64 { + let mut canonical = String::new(); + encode_group(&mut canonical, "keep", &config.keep); + encode_group(&mut canonical, "expire", &config.expire); + fnv1a(canonical.as_bytes()) +} + +fn encode_group(out: &mut String, label: &str, selectors: &[RetentionSelector]) { + out.push_str(label); + for selector in selectors { + out.push('|'); + encode_selector(out, selector); + } + out.push('\n'); +} + +fn encode_selector(out: &mut String, selector: &RetentionSelector) { + match selector { + RetentionSelector::Age { older_than_seconds } => { + out.push_str("age:"); + out.push_str(&older_than_seconds.to_string()); + } + RetentionSelector::Source { name } => { + out.push_str("source:"); + out.push_str(name); + } + RetentionSelector::ProjectPrefix { prefix } => { + out.push_str("project-prefix:"); + out.push_str(prefix); + } + RetentionSelector::KeepLatest { count } => { + out.push_str("keep-latest:"); + out.push_str(&count.to_string()); + } + RetentionSelector::Cached => out.push_str("cached"), + RetentionSelector::Trash => out.push_str("trash"), + RetentionSelector::Orphan => out.push_str("orphan"), + } +} + +fn fnv1a(bytes: &[u8]) -> u64 { + let mut hash = 0xcbf2_9ce4_8422_2325_u64; + for &byte in bytes { + hash ^= u64::from(byte); + hash = hash.wrapping_mul(0x0000_0100_0000_01b3); + } + hash +} diff --git a/crates/peryx-policy/tests/retention.rs b/crates/peryx-policy/tests/retention.rs new file mode 100644 index 00000000..ff17e8f6 --- /dev/null +++ b/crates/peryx-policy/tests/retention.rs @@ -0,0 +1,434 @@ +use peryx_policy::{ + RetentionCandidate, RetentionClass, RetentionConfig, RetentionFrontier, RetentionOutcome, RetentionPolicy, + RetentionSelector, RetentionSummary, RetentionVisibility, +}; + +fn candidate(project: &str, version: &str, rank: u64) -> RetentionCandidate { + RetentionCandidate { + project: project.to_owned(), + version: Some(version.to_owned()), + artifact: format!("{project}-{version}.whl"), + digest: format!("sha256:{project}{version}"), + class: RetentionClass::Hosted, + visibility: RetentionVisibility::Active, + source: None, + bytes: 10, + upload_time_unix: None, + rank, + orphan: false, + } +} + +fn keeping(selector: RetentionSelector) -> RetentionPolicy { + RetentionPolicy::compile(&RetentionConfig { + keep: vec![selector], + expire: Vec::new(), + }) +} + +fn expiring(selector: RetentionSelector) -> RetentionPolicy { + RetentionPolicy::compile(&RetentionConfig { + keep: Vec::new(), + expire: vec![selector], + }) +} + +fn outcomes(decisions: &[peryx_policy::RetentionDecision]) -> Vec<(&str, RetentionOutcome, Option<&str>)> { + decisions + .iter() + .map(|decision| (decision.artifact.as_str(), decision.outcome, decision.rule)) + .collect() +} + +#[test] +fn an_empty_policy_retains_every_candidate_with_no_rule() { + let policy = RetentionPolicy::compile(&RetentionConfig::default()); + assert!(policy.is_empty()); + + let decisions = policy.plan_project(None, vec![candidate("flask", "1.0", 0)]); + + assert_eq!(decisions.len(), 1); + assert_eq!(decisions[0].outcome, RetentionOutcome::Retain); + assert_eq!(decisions[0].rule, None); + assert!(decisions[0].retained_alternatives.is_empty()); +} + +#[test] +fn a_populated_policy_is_not_empty() { + assert!(!expiring(RetentionSelector::Cached).is_empty()); +} + +#[test] +fn each_selector_reports_its_stable_rule_name() { + for (selector, name) in [ + (RetentionSelector::Age { older_than_seconds: 1 }, "age"), + ( + RetentionSelector::Source { + name: "pypi".to_owned(), + }, + "source", + ), + ( + RetentionSelector::ProjectPrefix { + prefix: "acme-".to_owned(), + }, + "project-prefix", + ), + (RetentionSelector::KeepLatest { count: 2 }, "keep-latest"), + (RetentionSelector::Cached, "cached"), + (RetentionSelector::Trash, "trash"), + (RetentionSelector::Orphan, "orphan"), + ] { + assert_eq!(selector.name(), name); + } +} + +#[test] +fn keep_latest_protects_the_newest_versions_and_expires_the_rest() { + let policy = RetentionPolicy::compile(&RetentionConfig { + keep: vec![RetentionSelector::KeepLatest { count: 2 }], + expire: vec![RetentionSelector::ProjectPrefix { prefix: String::new() }], + }); + + let decisions = policy.plan_project( + None, + vec![ + candidate("flask", "1.0", 2), + candidate("flask", "3.0", 0), + candidate("flask", "2.0", 1), + ], + ); + + assert_eq!( + outcomes(&decisions), + vec![ + ("flask-3.0.whl", RetentionOutcome::Retain, Some("keep-latest")), + ("flask-2.0.whl", RetentionOutcome::Retain, Some("keep-latest")), + ("flask-1.0.whl", RetentionOutcome::Remove, Some("project-prefix")), + ] + ); +} + +#[test] +fn a_keep_rule_wins_over_a_matching_expire_rule() { + let policy = RetentionPolicy::compile(&RetentionConfig { + keep: vec![RetentionSelector::KeepLatest { count: 1 }], + expire: vec![RetentionSelector::Cached], + }); + let mut cached = candidate("flask", "1.0", 0); + cached.class = RetentionClass::Cached; + + let decisions = policy.plan_project(None, vec![cached]); + + assert_eq!(decisions[0].outcome, RetentionOutcome::Retain); + assert_eq!(decisions[0].rule, Some("keep-latest")); +} + +#[test] +fn a_removal_lists_the_surviving_versions_of_its_project_as_alternatives() { + let policy = RetentionPolicy::compile(&RetentionConfig { + keep: vec![RetentionSelector::KeepLatest { count: 2 }], + expire: vec![RetentionSelector::ProjectPrefix { prefix: String::new() }], + }); + + let decisions = policy.plan_project( + None, + vec![ + candidate("flask", "1.0", 2), + candidate("flask", "3.0", 0), + candidate("flask", "2.0", 1), + ], + ); + + let removed = decisions + .iter() + .find(|decision| decision.outcome == RetentionOutcome::Remove) + .unwrap(); + assert_eq!(removed.retained_alternatives, vec!["2.0".to_owned(), "3.0".to_owned()]); +} + +#[test] +fn an_age_rule_expires_only_candidates_older_than_its_bound() { + let policy = expiring(RetentionSelector::Age { + older_than_seconds: 100, + }); + let mut old = candidate("flask", "1.0", 0); + old.upload_time_unix = Some(0); + let mut fresh = candidate("flask", "2.0", 1); + fresh.upload_time_unix = Some(950); + + let decisions = policy.plan_project(Some(1_000), vec![old, fresh]); + + assert_eq!( + outcomes(&decisions), + vec![ + ("flask-1.0.whl", RetentionOutcome::Remove, Some("age")), + ("flask-2.0.whl", RetentionOutcome::Retain, None), + ] + ); +} + +#[test] +fn an_age_rule_ages_nothing_without_a_clock_or_a_publish_time() { + let policy = expiring(RetentionSelector::Age { older_than_seconds: 1 }); + let mut dated = candidate("flask", "1.0", 0); + dated.upload_time_unix = Some(0); + + let without_clock = policy.plan_project(None, vec![dated]); + assert_eq!(without_clock[0].outcome, RetentionOutcome::Retain); + + let undated = candidate("flask", "2.0", 0); + let without_time = policy.plan_project(Some(10_000), vec![undated]); + assert_eq!(without_time[0].outcome, RetentionOutcome::Retain); +} + +#[test] +fn a_source_rule_matches_the_named_routed_source() { + let policy = expiring(RetentionSelector::Source { + name: "upstream".to_owned(), + }); + let mut routed = candidate("flask", "1.0", 0); + routed.source = Some("upstream".to_owned()); + let mut other = candidate("flask", "2.0", 1); + other.source = Some("mirror".to_owned()); + + let decisions = policy.plan_project(None, vec![routed, other]); + + assert_eq!( + outcomes(&decisions), + vec![ + ("flask-1.0.whl", RetentionOutcome::Remove, Some("source")), + ("flask-2.0.whl", RetentionOutcome::Retain, None), + ] + ); +} + +#[test] +fn a_project_prefix_rule_matches_by_name() { + let policy = expiring(RetentionSelector::ProjectPrefix { + prefix: "acme-".to_owned(), + }); + + let decisions = policy.plan_project( + None, + vec![candidate("acme-tool", "1.0", 0), candidate("flask", "1.0", 0)], + ); + + assert_eq!( + outcomes(&decisions), + vec![ + ("acme-tool-1.0.whl", RetentionOutcome::Remove, Some("project-prefix")), + ("flask-1.0.whl", RetentionOutcome::Retain, None), + ] + ); +} + +#[test] +fn a_trash_rule_matches_soft_deleted_candidates() { + let policy = expiring(RetentionSelector::Trash); + let mut trashed = candidate("flask", "1.0", 0); + trashed.class = RetentionClass::Trash; + + assert_eq!( + policy.plan_project(None, vec![trashed])[0].outcome, + RetentionOutcome::Remove + ); + assert_eq!( + policy.plan_project(None, vec![candidate("flask", "1.0", 0)])[0].outcome, + RetentionOutcome::Retain + ); +} + +#[test] +fn an_orphan_rule_matches_unreferenced_candidates() { + let policy = expiring(RetentionSelector::Orphan); + let mut orphan = candidate("flask", "1.0", 0); + orphan.orphan = true; + + assert_eq!( + policy.plan_project(None, vec![orphan])[0].outcome, + RetentionOutcome::Remove + ); + assert_eq!( + policy.plan_project(None, vec![candidate("flask", "1.0", 0)])[0].outcome, + RetentionOutcome::Retain + ); +} + +#[test] +fn a_cached_keep_rule_protects_cached_candidates() { + let policy = keeping(RetentionSelector::Cached); + let mut cached = candidate("flask", "1.0", 0); + cached.class = RetentionClass::Cached; + + let decisions = policy.plan_project(None, vec![cached]); + + assert_eq!(decisions[0].outcome, RetentionOutcome::Retain); + assert_eq!(decisions[0].rule, Some("cached")); + assert_eq!(serde_json::to_value(&decisions[0]).unwrap()["class"], "cached"); +} + +#[test] +fn decisions_order_by_rank_then_artifact_then_digest() { + let policy = RetentionPolicy::compile(&RetentionConfig::default()); + let mut tie_a = candidate("flask", "1.0", 0); + tie_a.artifact = "flask-1.0-py3.whl".to_owned(); + tie_a.digest = "sha256:aaa".to_owned(); + let mut tie_b = candidate("flask", "1.0", 0); + tie_b.artifact = "flask-1.0-py3.whl".to_owned(); + tie_b.digest = "sha256:bbb".to_owned(); + + let decisions = policy.plan_project(None, vec![tie_b, candidate("flask", "2.0", 1), tie_a]); + + assert_eq!( + decisions + .iter() + .map(|decision| decision.digest.clone()) + .collect::>(), + vec!["sha256:aaa", "sha256:bbb", "sha256:flask2.0"] + ); +} + +#[test] +fn repeating_a_plan_produces_byte_identical_output() { + let policy = expiring(RetentionSelector::Trash); + let mut trashed = candidate("flask", "1.0", 1); + trashed.class = RetentionClass::Trash; + let build = || policy.plan_project(None, vec![candidate("flask", "2.0", 0), trashed.clone()]); + + assert_eq!( + serde_json::to_string(&build()).unwrap(), + serde_json::to_string(&build()).unwrap() + ); +} + +#[test] +fn a_removal_decision_serializes_every_recorded_field() { + let policy = expiring(RetentionSelector::Trash); + let mut trashed = candidate("flask", "1.0", 1); + trashed.class = RetentionClass::Trash; + trashed.visibility = RetentionVisibility::Yanked; + trashed.source = Some("upstream".to_owned()); + + let decisions = policy.plan_project(None, vec![candidate("flask", "2.0", 0), trashed]); + let removed = decisions + .iter() + .find(|decision| decision.outcome == RetentionOutcome::Remove) + .unwrap(); + + let json = serde_json::to_value(removed).unwrap(); + assert_eq!(json["outcome"], "remove"); + assert_eq!(json["rule"], "trash"); + assert_eq!(json["class"], "trash"); + assert_eq!(json["visibility"], "yanked"); + assert_eq!(json["source"], "upstream"); + assert_eq!(json["bytes"], 10); + assert_eq!(json["retained_alternatives"], serde_json::json!(["2.0"])); +} + +#[test] +fn a_hidden_generated_candidate_serializes_its_class_and_visibility() { + let policy = RetentionPolicy::compile(&RetentionConfig::default()); + let mut generated = candidate("flask", "1.0", 0); + generated.class = RetentionClass::Generated; + generated.visibility = RetentionVisibility::Hidden; + + let json = serde_json::to_value(&policy.plan_project(None, vec![generated])[0]).unwrap(); + + assert_eq!(json["class"], "generated"); + assert_eq!(json["visibility"], "hidden"); + assert_eq!(json.get("rule"), None); + assert_eq!(json.get("retained_alternatives"), None); +} + +#[test] +fn equal_rules_compile_to_one_version_and_distinct_rules_diverge() { + let all = RetentionConfig { + keep: vec![ + RetentionSelector::Age { older_than_seconds: 30 }, + RetentionSelector::Source { + name: "pypi".to_owned(), + }, + RetentionSelector::ProjectPrefix { + prefix: "acme-".to_owned(), + }, + RetentionSelector::KeepLatest { count: 5 }, + RetentionSelector::Cached, + RetentionSelector::Trash, + RetentionSelector::Orphan, + ], + expire: vec![RetentionSelector::Orphan], + }; + + assert_eq!( + RetentionPolicy::compile(&all).version(), + RetentionPolicy::compile(&all).version() + ); + assert_ne!( + RetentionPolicy::compile(&all).version(), + RetentionPolicy::compile(&RetentionConfig::default()).version() + ); + assert_ne!( + keeping(RetentionSelector::Orphan).version(), + expiring(RetentionSelector::Orphan).version() + ); +} + +#[test] +fn a_config_deserializes_every_selector_from_json() { + let config: RetentionConfig = serde_json::from_str( + r#"{ + "keep": [ + {"selector": "age", "older_than_seconds": 86400}, + {"selector": "source", "name": "pypi"}, + {"selector": "project-prefix", "prefix": "acme-"}, + {"selector": "keep-latest", "count": 5}, + {"selector": "cached"} + ], + "expire": [ + {"selector": "trash"}, + {"selector": "orphan"} + ] + }"#, + ) + .unwrap(); + + assert_eq!( + config, + RetentionConfig { + keep: vec![ + RetentionSelector::Age { + older_than_seconds: 86_400 + }, + RetentionSelector::Source { + name: "pypi".to_owned() + }, + RetentionSelector::ProjectPrefix { + prefix: "acme-".to_owned() + }, + RetentionSelector::KeepLatest { count: 5 }, + RetentionSelector::Cached, + ], + expire: vec![RetentionSelector::Trash, RetentionSelector::Orphan], + } + ); +} + +#[test] +fn a_summary_serializes_the_policy_version_and_metadata_frontier() { + let summary = RetentionSummary { + policy_version: keeping(RetentionSelector::Cached).version(), + frontier: RetentionFrontier { + repository: 7, + catalog: 3, + policy: 2, + }, + }; + + let json = serde_json::to_value(summary).unwrap(); + assert_eq!( + json["frontier"], + serde_json::json!({"repository": 7, "catalog": 3, "policy": 2}) + ); + assert_eq!(json["policy_version"], serde_json::json!(summary.policy_version)); +} From 33b52b05e697e1809832745cd639362bcfb370f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Fri, 17 Jul 2026 20:10:58 -0700 Subject: [PATCH 2/3] feat(pypi): plan hosted upload retention Adapt an index's hosted upload records into neutral retention candidates and stream their decisions. Uploads scan in key order, so a project's files group without a second pass; the adapter ranks each project's releases newest-first under PEP 440 and collapses equal spellings to one rank. The scan reads only indexed metadata, so an interrupted evaluation writes nothing. The result carries the policy version and the metadata frontier a later apply step needs to reject stale input. Refs #481 --- crates/peryx-ecosystem-pypi/src/lib.rs | 4 + crates/peryx-ecosystem-pypi/src/policy.rs | 2 +- crates/peryx-ecosystem-pypi/src/retention.rs | 374 +++++++++++++++++++ 3 files changed, 379 insertions(+), 1 deletion(-) create mode 100644 crates/peryx-ecosystem-pypi/src/retention.rs diff --git a/crates/peryx-ecosystem-pypi/src/lib.rs b/crates/peryx-ecosystem-pypi/src/lib.rs index a4bc9f42..705c3f3d 100644 --- a/crates/peryx-ecosystem-pypi/src/lib.rs +++ b/crates/peryx-ecosystem-pypi/src/lib.rs @@ -40,6 +40,8 @@ mod quota; #[cfg(feature = "serving")] mod requirement; #[cfg(feature = "serving")] +pub mod retention; +#[cfg(feature = "serving")] pub mod search_pypi; mod serial; #[cfg(feature = "serving")] @@ -58,6 +60,8 @@ mod version; #[cfg(feature = "serving")] pub use quota::quota_reservation; #[cfg(feature = "serving")] +pub use retention::evaluate_retention; +#[cfg(feature = "serving")] pub use search_pypi::PypiIndexer; #[cfg(feature = "serving")] pub use serving::PypiServing; diff --git a/crates/peryx-ecosystem-pypi/src/policy.rs b/crates/peryx-ecosystem-pypi/src/policy.rs index a905fce5..2825e5f3 100644 --- a/crates/peryx-ecosystem-pypi/src/policy.rs +++ b/crates/peryx-ecosystem-pypi/src/policy.rs @@ -610,7 +610,7 @@ fn facts_from_filename(filename: &str, size: Option) -> ArtifactFacts { /// Parse a Simple-API `upload-time` (RFC 3339, per PEP 700) into a Unix timestamp. A value without an /// offset, or otherwise unparseable, yields `None`, which the release-delay rule treats as a missing /// upload time. -fn parse_upload_time(value: &str) -> Option { +pub(crate) fn parse_upload_time(value: &str) -> Option { time::OffsetDateTime::parse(value, &time::format_description::well_known::Rfc3339) .ok() .map(time::OffsetDateTime::unix_timestamp) diff --git a/crates/peryx-ecosystem-pypi/src/retention.rs b/crates/peryx-ecosystem-pypi/src/retention.rs new file mode 100644 index 00000000..a972b084 --- /dev/null +++ b/crates/peryx-ecosystem-pypi/src/retention.rs @@ -0,0 +1,374 @@ +//! The `PyPI` half of retention-plan evaluation: adapt one index's hosted upload records into the +//! neutral [`RetentionCandidate`]s the [`peryx_policy`] engine plans over. +//! +//! Uploads scan in key order (`{index}/{normalized}/{filename}`), so a project's files arrive +//! contiguously. This groups them, ranks their versions newest-first under +//! [PEP 440](https://peps.python.org/pep-0440/), and streams the resulting decisions one project at a +//! time, so a large index never materializes as one in-memory plan. The scan reads only indexed +//! metadata, so an interrupted evaluation writes nothing. + +use std::cmp::Ordering; +use std::collections::HashMap; + +use peryx_policy::{ + RetentionCandidate, RetentionClass, RetentionDecision, RetentionFrontier, RetentionPolicy, RetentionSummary, + RetentionVisibility, +}; +use peryx_storage::meta::MetaStore; + +use crate::policy::parse_upload_time; +use crate::store::PypiStore as _; +use crate::upload::Uploaded; +use crate::version::{VersionKey, version_key}; +use crate::{Yanked, error_message}; + +/// Evaluate one index's hosted uploads against `policy`. +/// +/// Each artifact's decision passes to `emit` in deterministic order (newest version first). Returns the +/// plan's identity: the policy version and the metadata frontier the scan read. +/// +/// # Errors +/// Returns a message when the store cannot be read or an upload record does not decode. +pub fn evaluate_retention( + meta: &MetaStore, + index: &str, + policy: &RetentionPolicy, + now: Option, + mut emit: F, +) -> Result +where + F: FnMut(RetentionDecision), +{ + let frontier = read_frontier(meta, index)?; + let prefix = format!("{index}/"); + let mut current: Option = None; + let mut group: Vec = Vec::new(); + meta.scan_upload_records(|key, bytes| { + let Some((project, _filename)) = key.strip_prefix(&prefix).and_then(|rest| rest.split_once('/')) else { + return Ok(()); + }; + if current.as_deref() != Some(project) { + if let Some(name) = current.take() { + plan_group(&name, &group, policy, now, &mut emit); + } + current = Some(project.to_owned()); + group.clear(); + } + let uploaded: Uploaded = + serde_json::from_slice(bytes).map_err(|err| format!("corrupt upload record {key}: {err}"))?; + group.push(uploaded); + Ok::<(), String>(()) + }) + .map_err(error_message)?; + if let Some(name) = current { + plan_group(&name, &group, policy, now, &mut emit); + } + Ok(RetentionSummary { + policy_version: policy.version(), + frontier, + }) +} + +fn plan_group(project: &str, group: &[Uploaded], policy: &RetentionPolicy, now: Option, emit: &mut F) +where + F: FnMut(RetentionDecision), +{ + for decision in policy.plan_project(now, candidates(project, group)) { + emit(decision); + } +} + +fn candidates(project: &str, group: &[Uploaded]) -> Vec { + let ranks = version_ranks(group); + group + .iter() + .map(|uploaded| { + let file = &uploaded.file; + RetentionCandidate { + project: project.to_owned(), + version: Some(uploaded.version.clone()), + artifact: file.filename.clone(), + digest: file.hashes.get("sha256").cloned().unwrap_or_default(), + class: if uploaded.trashed.is_some() { + RetentionClass::Trash + } else { + RetentionClass::Hosted + }, + visibility: match file.yanked { + Yanked::No => RetentionVisibility::Active, + Yanked::Yes | Yanked::Reason(_) => RetentionVisibility::Yanked, + }, + source: None, + bytes: file.size.unwrap_or(0), + upload_time_unix: file.upload_time.as_deref().and_then(parse_upload_time), + rank: ranks[&version_key(&uploaded.version)], + orphan: false, + } + }) + .collect() +} + +/// Rank each distinct release newest-first. Two spellings of one release (`1.0`, `1.0.0`) collapse to +/// one rank; an unparseable legacy version ranks after every valid one, by string order. +fn version_ranks(group: &[Uploaded]) -> HashMap { + let mut distinct: Vec = group.iter().map(|uploaded| version_key(&uploaded.version)).collect(); + distinct.sort_by(version_key_desc); + distinct.dedup(); + distinct + .into_iter() + .enumerate() + .map(|(rank, key)| (key, rank as u64)) + .collect() +} + +fn version_key_desc(left: &VersionKey, right: &VersionKey) -> Ordering { + match (left, right) { + (VersionKey::Parsed(left), VersionKey::Parsed(right)) => right.cmp(left), + (VersionKey::Raw(left), VersionKey::Raw(right)) => left.cmp(right), + // A parsed release outranks any legacy spelling; both mixed orders resolve here, so neither + // depends on which direction the sort happens to compare them. + _ => parse_class(left).cmp(&parse_class(right)), + } +} + +const fn parse_class(key: &VersionKey) -> u8 { + match key { + VersionKey::Parsed(_) => 0, + VersionKey::Raw(_) => 1, + } +} + +fn read_frontier(meta: &MetaStore, index: &str) -> Result { + let generation = meta.policy_input_generation(index).map_err(error_message)?; + Ok(RetentionFrontier { + repository: meta.current_serial().map_err(error_message)?, + catalog: generation.catalog, + policy: generation.policy, + }) +} + +#[cfg(test)] +mod tests { + use std::cmp::Ordering; + use std::collections::BTreeMap; + + use peryx_policy::{ + RetentionClass, RetentionConfig, RetentionDecision, RetentionFrontier, RetentionOutcome, RetentionPolicy, + RetentionSelector, RetentionVisibility, + }; + use peryx_storage::meta::MetaStore; + + use super::evaluate_retention; + use crate::store::PypiStore as _; + use crate::upload::{TrashInfo, Uploaded}; + use crate::version::version_key; + use crate::{CoreMetadata, File, Provenance, Yanked}; + + fn store() -> (tempfile::TempDir, MetaStore) { + let dir = tempfile::tempdir().unwrap(); + let meta = MetaStore::open(dir.path().join("peryx.redb")).unwrap(); + (dir, meta) + } + + fn seed(meta: &MetaStore, index: &str, project: &str, version: &str, yanked: Yanked, trashed: Option) { + let filename = format!("{project}-{version}.whl"); + let uploaded = Uploaded { + version: version.to_owned(), + file: File { + filename: filename.clone(), + url: format!("https://files/{filename}"), + hashes: BTreeMap::from([("sha256".to_owned(), format!("sha-{version}"))]), + requires_python: None, + size: Some(1024), + upload_time: Some("2020-01-01T00:00:00Z".to_owned()), + yanked, + core_metadata: CoreMetadata::Absent, + dist_info_metadata: CoreMetadata::Absent, + gpg_sig: None, + provenance: Provenance::Absent, + }, + trashed, + }; + meta.put_upload(index, project, &filename, &serde_json::to_vec(&uploaded).unwrap()) + .unwrap(); + } + + fn plan(meta: &MetaStore, index: &str, policy: &RetentionPolicy) -> (Vec, RetentionFrontier) { + let mut decisions = Vec::new(); + let summary = evaluate_retention(meta, index, policy, None, |decision| decisions.push(decision)).unwrap(); + assert_eq!(summary.policy_version, policy.version()); + (decisions, summary.frontier) + } + + fn expire_all_but_latest(count: u64) -> RetentionPolicy { + RetentionPolicy::compile(&RetentionConfig { + keep: vec![RetentionSelector::KeepLatest { count }], + expire: vec![RetentionSelector::ProjectPrefix { prefix: String::new() }], + }) + } + + #[test] + fn test_evaluate_retention_orders_versions_by_pep440_and_keeps_the_newest() { + let (_dir, meta) = store(); + for version in ["2.0", "1.0", "1.0rc1", "2.0+local", "not-a-version", "also-bad"] { + seed(&meta, "pypi", "demo", version, Yanked::No, None); + } + + let (decisions, _) = plan(&meta, "pypi", &expire_all_but_latest(2)); + + let ordered: Vec<(&str, RetentionOutcome)> = decisions + .iter() + .map(|decision| (decision.version.as_deref().unwrap(), decision.outcome)) + .collect(); + assert_eq!( + ordered, + vec![ + ("2.0+local", RetentionOutcome::Retain), + ("2.0", RetentionOutcome::Retain), + ("1.0", RetentionOutcome::Remove), + ("1.0rc1", RetentionOutcome::Remove), + ("also-bad", RetentionOutcome::Remove), + ("not-a-version", RetentionOutcome::Remove), + ] + ); + } + + #[test] + fn test_evaluate_retention_lists_surviving_versions_as_alternatives() { + let (_dir, meta) = store(); + seed(&meta, "pypi", "demo", "2.0", Yanked::No, None); + seed(&meta, "pypi", "demo", "1.0", Yanked::No, None); + + let (decisions, _) = plan(&meta, "pypi", &expire_all_but_latest(1)); + + let removed = decisions + .iter() + .find(|decision| decision.outcome == RetentionOutcome::Remove) + .unwrap(); + assert_eq!(removed.version.as_deref(), Some("1.0")); + assert_eq!(removed.retained_alternatives, vec!["2.0".to_owned()]); + } + + #[test] + fn test_evaluate_retention_marks_a_trashed_record_and_records_its_class() { + let (_dir, meta) = store(); + seed( + &meta, + "pypi", + "demo", + "1.0", + Yanked::No, + Some(TrashInfo { + deleted_at_unix: 0, + actor: None, + reason: None, + }), + ); + + let policy = RetentionPolicy::compile(&RetentionConfig { + keep: Vec::new(), + expire: vec![RetentionSelector::Trash], + }); + let (decisions, _) = plan(&meta, "pypi", &policy); + + assert_eq!(decisions[0].outcome, RetentionOutcome::Remove); + assert_eq!(decisions[0].rule, Some("trash")); + assert_eq!(decisions[0].class, RetentionClass::Trash); + assert_eq!(decisions[0].bytes, 1024); + } + + #[test] + fn test_evaluate_retention_records_yanked_visibility() { + let (_dir, meta) = store(); + seed(&meta, "pypi", "demo", "1.0", Yanked::Reason("bad".to_owned()), None); + + let (decisions, _) = plan(&meta, "pypi", &RetentionPolicy::compile(&RetentionConfig::default())); + + assert_eq!(decisions[0].visibility, RetentionVisibility::Yanked); + assert_eq!(decisions[0].class, RetentionClass::Hosted); + } + + #[test] + fn test_evaluate_retention_streams_each_project_independently() { + let (_dir, meta) = store(); + seed(&meta, "pypi", "alpha", "2.0", Yanked::No, None); + seed(&meta, "pypi", "alpha", "1.0", Yanked::No, None); + seed(&meta, "pypi", "beta", "1.0", Yanked::No, None); + + let (decisions, _) = plan(&meta, "pypi", &expire_all_but_latest(1)); + + let removed: Vec<&str> = decisions + .iter() + .filter(|decision| decision.outcome == RetentionOutcome::Remove) + .map(|decision| decision.project.as_str()) + .collect(); + assert_eq!(removed, vec!["alpha"]); + } + + #[test] + fn test_evaluate_retention_skips_records_from_other_indexes() { + let (_dir, meta) = store(); + seed(&meta, "pypi", "demo", "1.0", Yanked::No, None); + seed(&meta, "other", "demo", "9.0", Yanked::No, None); + + let (decisions, _) = plan(&meta, "pypi", &expire_all_but_latest(1)); + + assert_eq!(decisions.len(), 1); + assert_eq!(decisions[0].version.as_deref(), Some("1.0")); + } + + #[test] + fn test_evaluate_retention_rejects_a_corrupt_upload_record() { + let (_dir, meta) = store(); + meta.put_upload("pypi", "demo", "demo-1.0.whl", b"not json").unwrap(); + + let result = evaluate_retention(&meta, "pypi", &expire_all_but_latest(1), None, |_| ()); + + assert!(result.unwrap_err().contains("corrupt upload record")); + } + + #[test] + fn test_evaluate_retention_plans_nothing_for_an_empty_index() { + let (_dir, meta) = store(); + + let (decisions, frontier) = plan(&meta, "pypi", &expire_all_but_latest(1)); + + assert!(decisions.is_empty()); + assert_eq!(frontier, RetentionFrontier::default()); + } + + #[test] + fn test_evaluate_retention_reports_the_metadata_frontier() { + let (_dir, meta) = store(); + meta.advance_policy_generation("pypi").unwrap(); + seed(&meta, "pypi", "demo", "1.0", Yanked::No, None); + + let (_, frontier) = plan(&meta, "pypi", &expire_all_but_latest(1)); + + assert_eq!(frontier.policy, 1); + } + + #[test] + fn test_evaluate_retention_is_byte_identical_across_runs() { + let (_dir, meta) = store(); + seed(&meta, "pypi", "demo", "2.0", Yanked::No, None); + seed(&meta, "pypi", "demo", "1.0", Yanked::No, None); + let policy = expire_all_but_latest(1); + let render = || serde_json::to_string(&plan(&meta, "pypi", &policy).0).unwrap(); + + assert_eq!(render(), render()); + } + + #[test] + fn test_version_key_desc_ranks_releases_before_legacy_spellings() { + let release = version_key("2.0"); + let older = version_key("1.0"); + let legacy = version_key("not-a-version"); + let other_legacy = version_key("also-bad"); + + assert_eq!(super::version_key_desc(&release, &older), Ordering::Less); + assert_eq!(super::version_key_desc(&release, &legacy), Ordering::Less); + assert_eq!(super::version_key_desc(&legacy, &release), Ordering::Greater); + assert_eq!(super::version_key_desc(&other_legacy, &legacy), Ordering::Less); + } +} From 2ba75cd9e19a23e5775bb8203a0f9efcb2be1645 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Fri, 17 Jul 2026 20:10:59 -0700 Subject: [PATCH 3/3] docs(core): document retention plans Describe the rule syntax, keep-over-expire precedence, PEP 440 version ordering, byte estimates, snapshot and policy identity, the output schema, and the side-effect-free contract. Refs #481 --- site/content/core/retention.md | 90 ++++++++++++++++++++++++++++++++++ 1 file changed, 90 insertions(+) create mode 100644 site/content/core/retention.md diff --git a/site/content/core/retention.md b/site/content/core/retention.md new file mode 100644 index 00000000..213bdad7 --- /dev/null +++ b/site/content/core/retention.md @@ -0,0 +1,90 @@ ++++ +title = "Retention plans" +description = "Evaluate an index's retention rules into a deterministic, side-effect-free removal plan." +weight = 10 ++++ + +A retention plan names which artifacts an index keeps and which become eligible for removal. Evaluation reads one +metadata snapshot, applies the configured rules, and returns an ordered decision per artifact. It changes no metadata +and touches no blob. This release ships the planner, which computes and reports decisions. Applying a plan, exposing it +over HTTP or the CLI, and reclaiming bytes come later and consume this planner without rebuilding it. + +The subject is an index's hosted upload records. Cached upstream pages are evicted through cache maintenance, and +unreferenced blobs are reclaimed through blob collection; neither is a retention decision. + +## Rules + +A policy holds two ordered rule groups. `keep` rules protect an artifact; `expire` rules mark it for removal. A rule +matches one dimension: + +| Rule | Matches | +| ---------------- | -------------------------------------------------------------- | +| `age` | An artifact published at least `older_than_seconds` before now | +| `source` | An artifact routed from the named source | +| `project-prefix` | An artifact whose project name begins with `prefix` | +| `keep-latest` | An artifact among the newest `count` versions of its project | +| `cached` | A cached artifact | +| `trash` | A soft-deleted artifact | +| `orphan` | An artifact no live reference reaches | + +The same rule protects in `keep` and removes in `expire`; the group gives it meaning. An `age` rule matches nothing when +the artifact carries no publish time or the evaluation supplies no clock, so the planner ages only what it can date. + +Rules load from configuration as a tagged list: + +```toml +keep = [ + { selector = "keep-latest", count = 10 }, + { selector = "age", older_than_seconds = 2592000 }, +] +expire = [ + { selector = "trash" }, + { selector = "project-prefix", prefix = "scratch-" }, +] +``` + +## Precedence + +A `keep` rule always wins over an `expire` rule, the precedence +[Google Artifact Registry cleanup policies](https://cloud.google.com/artifact-registry/docs/repositories/cleanup-policy) +define. The planner evaluates each artifact in order: the first matching `keep` rule retains it; otherwise the first +matching `expire` rule removes it; otherwise it is retained with no rule. Each decision names the rule that decided it, +so an operator reads why an artifact survived a policy that could have removed it. + +## Version ordering + +Versions rank newest first within a project. Python versions order under [PEP 440](https://peps.python.org/pep-0440/), +so `2.0` outranks `2.0rc1` and `2.0+local` outranks `2.0`. Two spellings of one release (`1.0` and `1.0.0`) collapse to +one rank, so `keep-latest` counts releases, not filenames. A version that is not valid PEP 440 ranks after every valid +one, ordered by its string, so a legacy spelling still gets a stable, documented position rather than an arbitrary one. + +`keep-latest` reads this rank: `count = 10` protects the ten newest releases and their files. + +## Output + +Evaluation streams one decision per artifact, ordered newest release first, then by filename, then by digest. The order +is total, so repeating an evaluation over the same snapshot and policy produces byte-identical output. + +Each decision records the artifact's project, version, filename, digest, storage class, and logical visibility (active, +yanked, or hidden). A removal decision adds: + +- `outcome`: `remove`, against `retain` for a kept artifact. +- `rule`: the rule that decided it. +- `bytes`: the artifact's estimated physical size, the capacity a removal would reclaim. +- `retained_alternatives`: the project's surviving versions, so a reader sees what a removal leaves in place. + +## Snapshot and policy identity + +A plan carries the identity of both inputs it read, so a later apply step can reject a plan built against stale state: + +- `policy_version`: a stable content hash of the compiled rules. Equal rules produce an equal version; any rule change + produces a different one. +- `frontier`: the metadata generation the scan read, combining the repository serial, the catalog generation, and the + policy generation. It mirrors the store's policy-input generation. + +## Side-effect-free contract + +Evaluation opens read transactions only. It reads indexed metadata and digest references and never enumerates backend +blobs. It groups one project at a time and streams that project's decisions before reading the next, so a large index +never holds as one in-memory plan. A dropped connection, a cancelled request, or a crash stops the scan mid-pass, and +the store keeps the state it already held, because the scan wrote none.