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-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); + } +} 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)); +} 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.