From 5874c5d6c3b18bdd5243afd258b9da036ebda8d7 Mon Sep 17 00:00:00 2001 From: FeathBow Date: Sun, 9 Aug 2026 00:51:33 +0100 Subject: [PATCH] feat(cli): report quota usage deltas around permanent purge batches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wraps the three permanent-cleanup batches — direct `clean --purge`, clean expiry purge, and explicit `trash purge` — with a read-only quota observation: a pre snapshot, the mutation, a post snapshot, and a signed per-scope usage delta. Observation is strictly best-effort and never blocks or alters the mutation: a failed pre canonicalization, provider error, or non-absolute request is reported as unavailable, and the mutation result and exit code are preserved regardless. A delta is reported only when the before/after snapshots are comparable — provider, data source, filesystem, mount identity (device major/minor and source), subject, and observation anchor must all match — so a mount replaced between the two probes yields an explicit incomparable result rather than a fabricated delta. Anchors resolving to the same scope fold to one probe. Signed deltas use i128 and are never clamped; JSON carries the full signed values and human output states the observed change is not attributed to degu. Empty entry plans still run, and observe, the aged-claim-marker housekeeping the human path previously skipped. The observation report is additive: default JSON/human output and exit codes are otherwise unchanged. A subject with privileged mount-namespace control can still stage an ABA mount swap within a single probe to influence a reporting-only delta; that grants no mutation authority and never changes deletion selection or exit code, and remains the existing hostile-root boundary. --- Cargo.toml | 2 +- crates/degu/src/action_result.rs | 109 +- crates/degu/src/commands/clean/execution.rs | 227 +++- crates/degu/src/commands/clean/output/json.rs | 7 +- crates/degu/src/commands/mod.rs | 13 + .../degu/src/commands/quota/output/tests.rs | 4 +- crates/degu/src/commands/trash/purge.rs | 78 +- crates/degu/src/lib.rs | 1 + .../degu/src/lifecycle/purge/housekeeping.rs | 4 +- crates/degu/src/lifecycle/purge/plan.rs | 10 + crates/degu/src/quota/model.rs | 40 +- crates/degu/src/quota/platform.rs | 58 +- crates/degu/src/quota/platform/linux.rs | 53 +- .../degu/src/quota/platform/linux/lustre.rs | 9 + crates/degu/src/quota_observation.rs | 1184 +++++++++++++++++ crates/degu/tests/clean/expiry.rs | 94 ++ .../degu/tests/schema/clean_scan_summary.rs | 35 + crates/degu/tests/schema/operations.rs | 34 + crates/degu/tests/schema/support.rs | 12 +- crates/degu/tests/trash/purge.rs | 102 +- 20 files changed, 1994 insertions(+), 82 deletions(-) create mode 100644 crates/degu/src/quota_observation.rs diff --git a/Cargo.toml b/Cargo.toml index 15fc919..dd9b891 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ comfy-table = "7.2" crossterm = { version = "0.29", default-features = false } terminal_size = "0.4" serde = { version = "1.0", features = ["derive"] } -serde_json = "1.0" +serde_json = { version = "1.0", features = ["arbitrary_precision"] } toml = "1.1" jiff = "0.2" thiserror = "2.0" diff --git a/crates/degu/src/action_result.rs b/crates/degu/src/action_result.rs index 5f6084c..a110647 100644 --- a/crates/degu/src/action_result.rs +++ b/crates/degu/src/action_result.rs @@ -6,7 +6,7 @@ //! converted into a lifecycle capability. use std::collections::HashSet; -use std::path::{Component, Path, PathBuf}; +use std::path::{Path, PathBuf}; use std::sync::Arc; /// Stable identity of one action within the result owner's namespace. @@ -109,24 +109,18 @@ pub(crate) enum CompletionBoundary { Completed, } -/// Canonical absolute path selected solely for read-only post-action observation. +/// Uninterpreted path requested solely for read-only action observation. /// -/// The caller must obtain the path from canonicalization and choose an anchor -/// expected to survive the action (normally a mount point or persistent parent). -/// Lexical validation here prevents accidentally recording a relative or -/// parent-traversing path, but does not assert filesystem identity or authority. +/// It is intentionally an infallible data wrapper: lexical `.`/`..` and symlink +/// traversal retain filesystem meaning until the observation pass canonicalizes +/// inside the non-authoritative probe phase. Relative requests are also captured +/// and later reported as unavailable rather than becoming setup failures. #[derive(Clone, Debug, Eq, Hash, PartialEq)] -pub(crate) struct CanonicalObservationAnchor(PathBuf); +pub(crate) struct ObservationRequestPath(PathBuf); -impl CanonicalObservationAnchor { - pub(crate) fn from_canonicalized(path: PathBuf) -> Result { - let mut components = path.components(); - if !matches!(components.next(), Some(Component::RootDir)) - || !components.all(|component| matches!(component, Component::Normal(_))) - { - return Err(ContractError::InvalidObservationAnchor); - } - Ok(Self(path)) +impl ObservationRequestPath { + pub(crate) fn new(path: PathBuf) -> Self { + Self(path) } /// Read-only quota probe input. Possessing this path grants no permission to @@ -136,21 +130,21 @@ impl CanonicalObservationAnchor { } } -/// One prospective quota scope, addressed by a canonical persistent anchor. -/// The quota-observation pass probes it and decides whether two anchors identify -/// the same scope and subject; this module intentionally does not duplicate the -/// quota identity vocabulary. +/// One prospective quota scope, addressed by an uninterpreted path request. +/// The quota-observation pass canonicalizes and probes it, then decides whether +/// requests identify the same provider scope and subject; this module intentionally +/// does not duplicate the quota identity vocabulary. #[derive(Clone, Debug, Eq, PartialEq)] pub(crate) struct QuotaObservationTarget { - anchor: CanonicalObservationAnchor, + anchor: ObservationRequestPath, } impl QuotaObservationTarget { - pub(crate) fn new(anchor: CanonicalObservationAnchor) -> Self { + pub(crate) fn new(anchor: ObservationRequestPath) -> Self { Self { anchor } } - pub(crate) fn anchor(&self) -> &CanonicalObservationAnchor { + pub(crate) fn anchor(&self) -> &ObservationRequestPath { &self.anchor } } @@ -177,7 +171,7 @@ impl ActionObservationTargets { &self.quota_scopes } - fn anchors(&self) -> HashSet { + fn anchors(&self) -> HashSet { self.quota_scopes .iter() .map(|target| target.anchor.clone()) @@ -200,7 +194,7 @@ pub(crate) enum QuotaObservationState { /// after the observation pass proves they address the same provider scope and subject. #[derive(Clone, Debug, Eq, PartialEq)] pub(crate) struct ResolvedQuotaObservation { - anchors: Vec, + anchors: Vec, state: QuotaObservationState, } @@ -208,7 +202,7 @@ impl ResolvedQuotaObservation { pub(crate) fn new( - anchors: impl IntoIterator, + anchors: impl IntoIterator, state: QuotaObservationState, ) -> Result { let mut seen = HashSet::new(); @@ -228,7 +222,7 @@ impl }) } - pub(crate) fn anchors(&self) -> &[CanonicalObservationAnchor] { + pub(crate) fn anchors(&self) -> &[ObservationRequestPath] { &self.anchors } @@ -482,6 +476,48 @@ impl BatchPostObservationPending { .ok_or(ContractError::ObservationTicketAlreadyTaken) } + /// Completes with the supplied resolution, or seals every exact target as + /// unavailable if an internal observation contract check fails. This keeps + /// reporting defects from discarding an already-produced mutation result. + pub(crate) fn complete_or_all_unavailable( + self, + observations: Result< + ActionObservations, + ContractError, + >, + unavailable: Unavailable, + ) -> CompletedActionBatchResult { + let observations = match observations { + Ok(observations) + if observations.belongs_to(&self.descriptor.correlation) + && observations.covers(&self.descriptor.targets) + && !observations.any_not_attempted() => + { + observations + } + Ok(_) | Err(_) => ActionObservations { + correlation: self.descriptor.correlation.clone(), + quota_scopes: self + .descriptor + .targets + .quota_scopes() + .iter() + .map(|target| ResolvedQuotaObservation { + anchors: vec![target.anchor.clone()], + state: QuotaObservationState::Unavailable(unavailable.clone()), + }) + .collect(), + }, + }; + CompletedActionBatchResult { + descriptor: self.descriptor, + start: StartBoundary::Started, + completion: CompletionBoundary::Completed, + outcome: self.outcome, + observations, + } + } + /// Crosses the completion boundary only with an observation resolution that /// covers the exact targets carried across the start/execution boundaries. pub(crate) fn complete( @@ -558,7 +594,6 @@ impl #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) enum ContractError { InvalidActionId, - InvalidObservationAnchor, EmptyObservationScope, DuplicateObservationAnchor, ObservationTargetMismatch, @@ -575,8 +610,8 @@ mod tests { ActionId::new(value).unwrap() } - fn anchor(path: &str) -> CanonicalObservationAnchor { - CanonicalObservationAnchor::from_canonicalized(PathBuf::from(path)).unwrap() + fn anchor(path: &str) -> ObservationRequestPath { + ObservationRequestPath::new(PathBuf::from(path)) } fn planned(targets: ActionObservationTargets) -> PlannedActionBatch { @@ -597,17 +632,9 @@ mod tests { } #[test] - fn observation_anchor_is_absolute_and_lexically_confined() { - assert_eq!(anchor("/").as_path(), Path::new("/")); - assert_eq!( - anchor("/home/cache-parent").as_path(), - Path::new("/home/cache-parent") - ); - for invalid in ["relative", "../escape", "/home/../escape"] { - assert_eq!( - CanonicalObservationAnchor::from_canonicalized(PathBuf::from(invalid)), - Err(ContractError::InvalidObservationAnchor) - ); + fn observation_request_preserves_alias_components_without_validation() { + for request in ["relative", "../escape", "/home/./cache", "/alias/../real"] { + assert_eq!(anchor(request).as_path(), Path::new(request)); } } diff --git a/crates/degu/src/commands/clean/execution.rs b/crates/degu/src/commands/clean/execution.rs index 29682a2..dba9bc4 100644 --- a/crates/degu/src/commands/clean/execution.rs +++ b/crates/degu/src/commands/clean/execution.rs @@ -1,16 +1,26 @@ use super::output; use super::preparation::PreparedClean; +use crate::action_result::{ActionKind, ActionResultOwner, NotStartedReason, StartedActionOutcome}; use crate::commands::next_action::{ self, CleanPreviewState, CleanResultState, OutputMode, Request, Workflow, }; use crate::commands::prompt::{confirm_permanent_delete, confirm_required}; use crate::lifecycle::{CleanExecution, ExpiryPlan, Lifecycle, MutationSession, PurgeReport}; +use crate::quota_observation::{ + QuotaActionReport, coordinate, not_attempted_action, planned_action, +}; use anyhow::Result; use degu_core::finding::Finding; +use std::path::PathBuf; pub(super) struct ExpiryExecution { pub(super) plan: ExpiryPlan, pub(super) report: Option, + pub(super) observation: QuotaActionReport, +} + +pub(super) struct CleanQuotaObservations { + pub(super) direct_purge: QuotaActionReport, } pub(super) fn run(prepared: PreparedClean) -> Result<()> { @@ -45,11 +55,34 @@ fn boundary_recheck<'a>( fn run_json(prepared: PreparedClean) -> Result<()> { if prepared.settings.dry_run { + let plan = Lifecycle::new(&prepared.ctx).plan_expired()?; + let expiry_observation = not_attempted_action( + ActionResultOwner::CleanCommand, + ActionKind::ExpiryPurge, + "clean:expiry-purge", + plan.trash_roots().map(PathBuf::from), + NotStartedReason::DryRun, + ) + .map_err(|error| anyhow::anyhow!("invalid expiry observation contract: {error:?}"))?; let expiry = ExpiryExecution { - plan: Lifecycle::new(&prepared.ctx).plan_expired()?, + plan, report: None, + observation: expiry_observation, }; - return output::print_json(&prepared, &[], &expiry); + let direct_purge = not_attempted_action( + ActionResultOwner::CleanCommand, + ActionKind::DirectPurge, + "clean:direct-purge", + [], + NotStartedReason::DryRun, + ) + .map_err(|error| anyhow::anyhow!("invalid direct observation contract: {error:?}"))?; + return output::print_json( + &prepared, + &[], + &expiry, + &CleanQuotaObservations { direct_purge }, + ); } let session = prepared.lock()?; output::validate_json_prepared(&prepared)?; @@ -58,11 +91,27 @@ fn run_json(prepared: PreparedClean) -> Result<()> { prepared.revalidate(&session)?; stop_if_stdout_closed()?; let recheck = boundary_recheck(&prepared, &session); - let executed = session.execute_clean(&prepared.plan, prepared.settings.purge, &recheck); + let (executed, direct_purge) = execute_clean(&prepared, &session, &recheck)?; let clean_failed = executed.iter().any(CleanExecution::failed); - let report = (!clean_failed).then(|| session.execute_expiry(&plan)); - let expiry = ExpiryExecution { plan, report }; - let output_result = output::print_json(&prepared, &executed, &expiry); + let expiry = execute_expiry(&session, plan, clean_failed)?; + let direct_purge = match direct_purge { + Some(observation) => observation, + None => not_attempted_action( + ActionResultOwner::CleanCommand, + ActionKind::DirectPurge, + "clean:direct-purge", + [], + NotStartedReason::Empty, + ) + .map_err(|error| anyhow::anyhow!("invalid direct observation contract: {error:?}"))?, + }; + let observations = CleanQuotaObservations { direct_purge }; + crate::quota_observation::print_warnings( + &observations.direct_purge, + prepared.settings.ui.colors, + ); + crate::quota_observation::print_warnings(&expiry.observation, prepared.settings.ui.colors); + let output_result = output::print_json(&prepared, &executed, &expiry, &observations); ensure_clean_success(clean_failed)?; ensure_expiry_success(&expiry)?; output_result @@ -72,7 +121,10 @@ fn run_human(prepared: PreparedClean) -> Result<()> { if prepared.settings.dry_run { return run_human_preview(&prepared); } - if prepared.plan.items().is_empty() && Lifecycle::new(&prepared.ctx).plan_expired()?.is_empty() + if prepared.plan.items().is_empty() + && !Lifecycle::new(&prepared.ctx) + .plan_expired()? + .has_housekeeping_scope() { return output::print_plan(&prepared); } @@ -80,7 +132,7 @@ fn run_human(prepared: PreparedClean) -> Result<()> { let expiry_plan = session.plan_expired()?; output::print_plan(&prepared)?; output::print_mutation_scope(&prepared, &expiry_plan)?; - if prepared.plan.items().is_empty() && expiry_plan.is_empty() { + if prepared.plan.items().is_empty() && !expiry_plan.has_housekeeping_scope() { return Ok(()); } let permanent = permanent_deletion_planned(&prepared, &expiry_plan); @@ -142,19 +194,24 @@ fn execute_human_plan( stop_if_stdout_closed()?; let started = std::time::Instant::now(); let recheck = boundary_recheck(&prepared, &session); - let executed = session.execute_clean(&prepared.plan, prepared.settings.purge, &recheck); + let (executed, direct_purge) = execute_clean(&prepared, &session, &recheck)?; let elapsed = started.elapsed(); let failed = executed.iter().any(CleanExecution::failed); - let expiry = ExpiryExecution { - report: (!failed).then(|| session.execute_expiry(&plan)), - plan, - }; + let expiry = execute_expiry(&session, plan, failed)?; let output_result = output::print_execution(&prepared, &executed, Some(elapsed)) .and_then(|()| { + if let Some(observation) = &direct_purge { + crate::quota_observation::print_human(observation, prepared.settings.ui.colors)?; + } if failed { Ok(()) } else { - output::print_expiry(&expiry, prepared.settings.ui.colors) + output::print_expiry(&expiry, prepared.settings.ui.colors)?; + crate::quota_observation::print_human( + &expiry.observation, + prepared.settings.ui.colors, + )?; + Ok(()) } }) .and_then(|()| print_result_next(&prepared, &executed)); @@ -163,6 +220,121 @@ fn execute_human_plan( output_result } +/// The caller has completed final batch revalidation and the stdout boundary. +/// The per-finding recheck stays inside lifecycle execution immediately before +/// each mutation; moving it before the pre probe would only widen its race. +fn direct_observation_request( + source: &std::path::Path, + resolved: std::result::Result, +) -> PathBuf { + resolved + .map(|root| root.parent().unwrap_or(&root).to_path_buf()) + .unwrap_or_else(|_| source.parent().unwrap_or(source).to_path_buf()) +} + +fn execute_clean( + prepared: &PreparedClean, + session: &MutationSession, + recheck: &dyn Fn(&Finding) -> Result<(), String>, +) -> Result<(Vec, Option)> { + if !prepared.settings.purge || prepared.plan.items().is_empty() { + return Ok(( + session.execute_clean(&prepared.plan, prepared.settings.purge, recheck), + None, + )); + } + let lifecycle = Lifecycle::new(&prepared.ctx); + let anchors = prepared + .plan + .items() + .iter() + .map(|finding| { + // Observation discovery is reporting-only. A resolver failure must + // not upgrade lifecycle's per-item failure into a batch setup + // failure; retain a non-authoritative source-side request. + direct_observation_request(finding.path(), lifecycle.resolve_trash_dir(finding.path())) + }) + .collect::>(); + let action = planned_action( + ActionResultOwner::CleanCommand, + ActionKind::DirectPurge, + "clean:direct-purge", + anchors, + ) + .map_err(|error| anyhow::anyhow!("invalid direct-purge observation contract: {error:?}"))?; + let mut probe = crate::quota::probe; + let (executed, completed) = coordinate(action, &mut probe, || { + let executed = session.execute_clean(&prepared.plan, true, recheck); + let outcome = clean_outcome(&executed); + (executed, outcome) + }); + Ok((executed, Some(QuotaActionReport::Attempted(completed)))) +} + +fn execute_expiry( + session: &MutationSession, + plan: ExpiryPlan, + clean_failed: bool, +) -> Result { + if clean_failed { + let observation = not_attempted_action( + ActionResultOwner::CleanCommand, + ActionKind::ExpiryPurge, + "clean:expiry-purge", + plan.trash_roots().map(PathBuf::from), + NotStartedReason::PrerequisiteFailed, + ) + .map_err(|error| anyhow::anyhow!("invalid expiry observation contract: {error:?}"))?; + return Ok(ExpiryExecution { + report: None, + plan, + observation, + }); + } + if !plan.has_housekeeping_scope() { + let observation = not_attempted_action( + ActionResultOwner::CleanCommand, + ActionKind::ExpiryPurge, + "clean:expiry-purge", + [], + NotStartedReason::Empty, + ) + .map_err(|error| anyhow::anyhow!("invalid expiry observation contract: {error:?}"))?; + return Ok(ExpiryExecution { + report: Some(PurgeReport::default()), + plan, + observation, + }); + } + let action = planned_action( + ActionResultOwner::CleanCommand, + ActionKind::ExpiryPurge, + "clean:expiry-purge", + plan.trash_roots().map(PathBuf::from), + ) + .map_err(|error| anyhow::anyhow!("invalid expiry observation contract: {error:?}"))?; + let mut probe = crate::quota::probe; + let (report, completed) = coordinate(action, &mut probe, || { + let report = session.execute_expiry(&plan); + let outcome = crate::commands::purge_outcome(&report); + (report, outcome) + }); + Ok(ExpiryExecution { + plan, + report: Some(report), + observation: QuotaActionReport::Attempted(completed), + }) +} + +fn clean_outcome(executed: &[CleanExecution]) -> StartedActionOutcome { + let failures = executed.iter().filter(|item| item.failed()).count(); + match failures { + 0 => StartedActionOutcome::Success, + count if count == executed.len() => StartedActionOutcome::Failure, + _ => StartedActionOutcome::Partial, + } +} + fn print_result_next( prepared: &PreparedClean, executed: &[crate::lifecycle::CleanExecution], @@ -195,3 +367,30 @@ fn ensure_expiry_success(expiry: &ExpiryExecution) -> Result<()> { } Ok(()) } + +#[cfg(test)] +mod tests { + use super::direct_observation_request; + use std::path::{Path, PathBuf}; + + #[test] + fn direct_observation_discovery_keeps_failed_and_valid_items_in_the_batch() { + let requests = [ + direct_observation_request( + Path::new("/source/failed/cache"), + Err("resolver failed".to_owned()), + ), + direct_observation_request( + Path::new("/source/valid/cache"), + Ok(PathBuf::from("/persistent/trash")), + ), + ]; + assert_eq!( + requests, + [ + PathBuf::from("/source/failed"), + PathBuf::from("/persistent") + ] + ); + } +} diff --git a/crates/degu/src/commands/clean/output/json.rs b/crates/degu/src/commands/clean/output/json.rs index 20d4342..c3264b2 100644 --- a/crates/degu/src/commands/clean/output/json.rs +++ b/crates/degu/src/commands/clean/output/json.rs @@ -1,4 +1,4 @@ -use super::super::execution::ExpiryExecution; +use super::super::execution::{CleanQuotaObservations, ExpiryExecution}; use super::super::preparation::PreparedClean; use crate::lifecycle::{CleanExecution, ExpiryPlan, Lifecycle, TRASH_RETENTION_DAYS}; use crate::output::stdoutln; @@ -9,6 +9,7 @@ pub(crate) fn print( prepared: &PreparedClean, executed: &[CleanExecution], expiry: &ExpiryExecution, + observations: &CleanQuotaObservations, ) -> Result<()> { let (planned, excluded, omitted) = prepared_findings_json(prepared)?; let executed = executed @@ -23,6 +24,10 @@ pub(crate) fn print( "executed": executed, "opt_in": prepared.scope.include_review(), "expiry": expiry_json(expiry)?, + "quota_observations": { + "direct_purge": crate::quota_observation::json(&observations.direct_purge), + "expiry_purge": crate::quota_observation::json(&expiry.observation), + }, }); stdoutln!("{}", serde_json::to_string_pretty(&report)?) } diff --git a/crates/degu/src/commands/mod.rs b/crates/degu/src/commands/mod.rs index cb58b6b..0f1d7c4 100644 --- a/crates/degu/src/commands/mod.rs +++ b/crates/degu/src/commands/mod.rs @@ -35,3 +35,16 @@ impl CollectionRunOptions { } } } + +pub(crate) fn purge_outcome( + report: &crate::lifecycle::PurgeReport, +) -> crate::action_result::StartedActionOutcome { + use crate::action_result::StartedActionOutcome; + if report.failed.is_empty() { + StartedActionOutcome::Success + } else if report.purged.is_empty() { + StartedActionOutcome::Failure + } else { + StartedActionOutcome::Partial + } +} diff --git a/crates/degu/src/commands/quota/output/tests.rs b/crates/degu/src/commands/quota/output/tests.rs index 14c2fb3..e4d0271 100644 --- a/crates/degu/src/commands/quota/output/tests.rs +++ b/crates/degu/src/commands/quota/output/tests.rs @@ -1,7 +1,7 @@ use super::{format_count, render_dimension, render_human}; use crate::quota::model::{ ActiveQuota, QuotaDimension, QuotaGrace, QuotaGraceState, QuotaLimits, QuotaScope, - QuotaSnapshot, + QuotaScopeIdentity, QuotaSnapshot, }; use crate::runtime::{Glyphs, Ui}; use std::path::PathBuf; @@ -12,6 +12,7 @@ fn quota_human_escapes_scope_and_provider_terminal_controls() { PathBuf::from("/tmp/target\nrow"), PathBuf::from("/mnt/\tdata"), "ext4\x1b[31m".to_owned(), + QuotaScopeIdentity::new(36, 8, 1, PathBuf::from("/dev/root")), ); let report = QuotaSnapshot::active( scope, @@ -43,6 +44,7 @@ fn quota_human_reflows_fields_for_a_narrow_terminal() { PathBuf::from("/home/user/a-very-long-quota-target"), PathBuf::from("/home/user/a-very-long-mount-point"), "ext4".to_owned(), + QuotaScopeIdentity::new(36, 8, 1, PathBuf::from("/dev/root")), ); let report = QuotaSnapshot::active( scope, diff --git a/crates/degu/src/commands/trash/purge.rs b/crates/degu/src/commands/trash/purge.rs index 2ba7d7c..c63db59 100644 --- a/crates/degu/src/commands/trash/purge.rs +++ b/crates/degu/src/commands/trash/purge.rs @@ -2,11 +2,15 @@ use anyhow::Result; use degu_core::ecosystem::DetectCtx; use std::path::Path; +use crate::action_result::{ActionKind, ActionResultOwner, NotStartedReason}; use crate::commands::prompt::confirm_permanent_delete; use crate::lifecycle::{Lifecycle, TrashPurgePlan}; use crate::output::{flush_stdout, stdoutln}; use crate::presentation::semantic::Tone; use crate::presentation::{display_path, escape_terminal_text, semantic}; +use crate::quota_observation::{ + QuotaActionReport, coordinate, not_attempted_action, planned_action, +}; use crate::runtime::Ui; use serde::Serialize; @@ -20,7 +24,7 @@ pub(super) fn run(json: bool, yes: bool, ui: Ui) -> Result<()> { if json { validate_json_plan(&plan)?; } else { - if plan.is_empty() { + if !plan.has_housekeeping_scope() { return stdoutln!("{}", super::output::TRASH_IS_EMPTY); } print_plan(&plan, &ctx.home, ui.colors.stdout)?; @@ -29,12 +33,42 @@ pub(super) fn run(json: bool, yes: bool, ui: Ui) -> Result<()> { if !yes && !confirm_permanent_delete(ui.colors)? { anyhow::bail!("Purge cancelled; no trash entries were deleted."); } + if crate::output::stdout_consumer_gone() { + return Err(crate::output::stdout_closed_error()); + } - let report = session.execute_purge_all(plan); + let (report, observation) = if !plan.has_housekeeping_scope() { + let observation = not_attempted_action( + ActionResultOwner::TrashPurgeCommand, + ActionKind::TrashPurge, + "trash:purge-all", + [], + NotStartedReason::Empty, + ) + .map_err(|error| anyhow::anyhow!("invalid trash observation contract: {error:?}"))?; + (session.execute_purge_all(plan), observation) + } else { + let action = planned_action( + ActionResultOwner::TrashPurgeCommand, + ActionKind::TrashPurge, + "trash:purge-all", + plan.trash_roots().map(std::path::PathBuf::from), + ) + .map_err(|error| anyhow::anyhow!("invalid trash-purge observation contract: {error:?}"))?; + let mut probe = crate::quota::probe; + let (report, completed) = coordinate(action, &mut probe, || { + let report = session.execute_purge_all(plan); + let outcome = crate::commands::purge_outcome(&report); + (report, outcome) + }); + (report, QuotaActionReport::Attempted(completed)) + }; let output_result = if json { - print_json_report(&report) + crate::quota_observation::print_warnings(&observation, ui.colors); + print_json_report(&report, &observation) } else { print_human_report(&report.purged, &report.failed, ui.colors) + .and_then(|()| crate::quota_observation::print_human(&observation, ui.colors)) }; if !report.failed.is_empty() { anyhow::bail!("one or more trash entries failed to purge") @@ -42,8 +76,14 @@ pub(super) fn run(json: bool, yes: bool, ui: Ui) -> Result<()> { output_result } -fn print_json_report(report: &crate::lifecycle::PurgeReport) -> Result<()> { - stdoutln!("{}", serde_json::to_string_pretty(&json_report(report))?) +fn print_json_report( + report: &crate::lifecycle::PurgeReport, + observation: &QuotaActionReport, +) -> Result<()> { + stdoutln!( + "{}", + serde_json::to_string_pretty(&json_report(report, observation))? + ) } fn validate_json_plan(plan: &TrashPurgePlan) -> Result<()> { @@ -61,6 +101,7 @@ fn validate_json_plan(plan: &TrashPurgePlan) -> Result<()> { struct PurgeJsonReport<'a> { purged: &'a [std::path::PathBuf], failed: Vec>, + quota_observations: serde_json::Value, } #[derive(Serialize)] @@ -69,7 +110,10 @@ struct PurgeFailureJson<'a> { reason: &'a str, } -fn json_report(report: &crate::lifecycle::PurgeReport) -> PurgeJsonReport<'_> { +fn json_report<'a>( + report: &'a crate::lifecycle::PurgeReport, + observation: &QuotaActionReport, +) -> PurgeJsonReport<'a> { let failed = report .failed .iter() @@ -78,16 +122,20 @@ fn json_report(report: &crate::lifecycle::PurgeReport) -> PurgeJsonReport<'_> { PurgeJsonReport { purged: &report.purged, failed, + quota_observations: crate::quota_observation::json(observation), } } fn print_plan(plan: &TrashPurgePlan, home: &Path, color_enabled: bool) -> Result<()> { - let noun = if plan.len() == 1 { "entry" } else { "entries" }; let action = semantic::paint( "will be permanently deleted", Tone::Destructive, color_enabled, ); + if plan.is_empty() { + return stdoutln!("Purge plan: expired trash claim markers, if present, {action}."); + } + let noun = if plan.len() == 1 { "entry" } else { "entries" }; stdoutln!("Purge plan: all {} trash {noun} {action}.", plan.len(),)?; for entry in plan.entries() { stdoutln!(" {}", escape_terminal_text(&display_path(entry, home)))?; @@ -154,8 +202,20 @@ mod tests { "changed".to_owned(), )], }; - let json = serde_json::to_value(json_report(&report)).unwrap(); - assert_eq!(keys(&json), ["failed", "purged"]); + let observation = crate::quota_observation::not_attempted_action( + crate::action_result::ActionResultOwner::TrashPurgeCommand, + crate::action_result::ActionKind::TrashPurge, + "trash:test", + [], + crate::action_result::NotStartedReason::Empty, + ) + .unwrap(); + let json = serde_json::to_value(json_report(&report, &observation)).unwrap(); + assert_eq!(keys(&json), ["failed", "purged", "quota_observations"]); + assert_eq!( + json["quota_observations"]["observation_state"], + "not_attempted" + ); assert_eq!(keys(&json["failed"][0]), ["path", "reason"]); } } diff --git a/crates/degu/src/lib.rs b/crates/degu/src/lib.rs index ca4b0ad..f0a3ead 100644 --- a/crates/degu/src/lib.rs +++ b/crates/degu/src/lib.rs @@ -17,6 +17,7 @@ mod lifecycle; mod output; mod presentation; mod quota; +mod quota_observation; mod runtime; mod source_selection; mod value_parser; diff --git a/crates/degu/src/lifecycle/purge/housekeeping.rs b/crates/degu/src/lifecycle/purge/housekeeping.rs index b6977fb..e980d91 100644 --- a/crates/degu/src/lifecycle/purge/housekeeping.rs +++ b/crates/degu/src/lifecycle/purge/housekeeping.rs @@ -5,7 +5,7 @@ use anyhow::{Context, Result}; use degu_core::oplog::ObjectIdentity; use crate::lifecycle::claims::{reservation_marker_metadata, validate_existing_claims_dir}; -use crate::lifecycle::expiry::{TRASH_TTL, fallback_age}; +use crate::lifecycle::expiry::{TRASH_TTL, fallback_mtime_age}; pub(super) fn purge_expired_claims(root: &Path) -> Result<()> { let Some(claims) = validate_existing_claims_dir(root) @@ -24,7 +24,7 @@ pub(super) fn purge_expired_claims(root: &Path) -> Result<()> { }; let path = entry.path(); let expected = ObjectIdentity::from_metadata(&metadata); - if fallback_age(&metadata, now) >= TRASH_TTL { + if fallback_mtime_age(&metadata, now) >= TRASH_TTL { trash.purge_entry_verified(&path, expected)?; } } diff --git a/crates/degu/src/lifecycle/purge/plan.rs b/crates/degu/src/lifecycle/purge/plan.rs index 8b73b45..fa77e38 100644 --- a/crates/degu/src/lifecycle/purge/plan.rs +++ b/crates/degu/src/lifecycle/purge/plan.rs @@ -38,6 +38,11 @@ impl ExpiryPlan { self.entries().next().is_none() } + /// Even an entry-empty batch may purge aged numeric claim markers. + pub(crate) fn has_housekeeping_scope(&self) -> bool { + !self.batches.is_empty() + } + pub(crate) fn len(&self) -> usize { batch_entry_count(&self.batches) } @@ -59,6 +64,11 @@ impl TrashPurgePlan { self.len() == 0 } + /// Even an entry-empty batch may purge aged numeric claim markers. + pub(crate) fn has_housekeeping_scope(&self) -> bool { + !self.batches.is_empty() + } + pub(crate) fn len(&self) -> usize { batch_entry_count(&self.batches) } diff --git a/crates/degu/src/quota/model.rs b/crates/degu/src/quota/model.rs index ad44335..80775e3 100644 --- a/crates/degu/src/quota/model.rs +++ b/crates/degu/src/quota/model.rs @@ -35,20 +35,55 @@ pub(crate) struct ActiveQuota { pub(crate) inodes: QuotaDimension, } +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +pub(crate) struct QuotaScopeIdentity { + mount_id: u64, + device_major: u32, + device_minor: u32, + source: PathBuf, +} + +#[cfg(any(target_os = "linux", test))] +impl QuotaScopeIdentity { + pub(crate) fn new( + mount_id: u64, + device_major: u32, + device_minor: u32, + source: PathBuf, + ) -> Self { + Self { + mount_id, + device_major, + device_minor, + source, + } + } +} + #[derive(Debug, Serialize)] pub(crate) struct QuotaScope { pub(crate) path: PathBuf, pub(crate) mount_point: PathBuf, pub(crate) filesystem: String, + /// Provider-private identity used to reject comparisons across mount + /// replacement. It is evidence for observation only, not public quota data. + #[serde(skip)] + pub(crate) identity: QuotaScopeIdentity, } #[cfg(any(target_os = "linux", test))] impl QuotaScope { - pub(crate) fn new(path: PathBuf, mount_point: PathBuf, filesystem: String) -> Self { + pub(crate) fn new( + path: PathBuf, + mount_point: PathBuf, + filesystem: String, + identity: QuotaScopeIdentity, + ) -> Self { Self { path, mount_point, filesystem, + identity, } } } @@ -152,7 +187,7 @@ fn headroom(limit: Option, used: u64) -> Option { mod tests { use super::{ ActiveQuota, QuotaDimension, QuotaGrace, QuotaGraceState, QuotaLimits, QuotaScope, - QuotaSnapshot, + QuotaScopeIdentity, QuotaSnapshot, }; use std::path::PathBuf; @@ -234,6 +269,7 @@ mod tests { PathBuf::from("/home/me"), PathBuf::from("/home"), "ext4".to_owned(), + QuotaScopeIdentity::new(36, 8, 1, PathBuf::from("/dev/root")), ); let report = QuotaSnapshot::active( scope, diff --git a/crates/degu/src/quota/platform.rs b/crates/degu/src/quota/platform.rs index 0dc5430..5ba32ee 100644 --- a/crates/degu/src/quota/platform.rs +++ b/crates/degu/src/quota/platform.rs @@ -3,28 +3,40 @@ mod linux; #[cfg(target_os = "macos")] mod macos; -#[cfg(target_os = "linux")] -use super::model::QuotaScope; use super::model::QuotaSnapshot; +#[cfg(target_os = "linux")] +use super::model::{QuotaScope, QuotaScopeIdentity}; use crate::presentation::escape_terminal_text as escaped; use std::fmt; use std::path::Path; #[cfg(any(target_os = "linux", target_os = "macos"))] use std::path::PathBuf; -#[derive(Debug)] +#[derive(Clone, Debug, Eq, PartialEq)] #[cfg(any(target_os = "linux", target_os = "macos"))] pub(super) struct MountInfo { pub(super) mount_point: PathBuf, pub(super) filesystem: String, #[cfg(target_os = "linux")] pub(super) source: PathBuf, + #[cfg(target_os = "linux")] + pub(super) mount_id: u64, + #[cfg(target_os = "linux")] + pub(super) device_major: u32, + #[cfg(target_os = "linux")] + pub(super) device_minor: u32, } #[cfg(target_os = "linux")] impl MountInfo { pub(super) fn scope(self, path: &Path) -> QuotaScope { - QuotaScope::new(path.to_owned(), self.mount_point, self.filesystem) + let identity = QuotaScopeIdentity::new( + self.mount_id, + self.device_major, + self.device_minor, + self.source, + ); + QuotaScope::new(path.to_owned(), self.mount_point, self.filesystem, identity) } } @@ -90,6 +102,44 @@ impl fmt::Display for ProbeError { } } +impl ProbeError { + pub(crate) fn category(&self) -> &'static str { + match self { + #[cfg(target_os = "linux")] + Self::NotConfigured { .. } => "not_configured", + Self::Unsupported { .. } => "unsupported", + #[cfg(any(target_os = "linux", test))] + Self::Unavailable { .. } => "unavailable", + #[cfg(target_os = "linux")] + Self::PermissionDenied { .. } => "permission_denied", + #[cfg(target_os = "linux")] + Self::Incomplete { .. } => "incomplete", + #[cfg(any(target_os = "linux", target_os = "macos"))] + Self::Io { .. } => "io", + } + } + + /// Raw diagnostic for structured JSON. Terminal escaping belongs only to + /// human presentation and must never be persisted in machine output. + pub(crate) fn raw_message(&self) -> String { + match self { + #[cfg(target_os = "linux")] + Self::NotConfigured { + filesystem, + mount_point, + } => format!("quota not configured for {filesystem} mounted at {mount_point}"), + #[cfg(any(target_os = "linux", target_os = "macos"))] + Self::Io { path, source } => { + format!("quota probe failed for {path}: {source}") + } + _ => { + let (label, filesystem, mount_point, reason) = failure_fields(self); + format!("{label} for {filesystem} mounted at {mount_point}: {reason}") + } + } + } +} + impl std::error::Error for ProbeError { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { match self { diff --git a/crates/degu/src/quota/platform/linux.rs b/crates/degu/src/quota/platform/linux.rs index 9468d14..2ee015e 100644 --- a/crates/degu/src/quota/platform/linux.rs +++ b/crates/degu/src/quota/platform/linux.rs @@ -12,7 +12,6 @@ const DATA_SOURCE: &str = "linux_quotactl"; const QUOTA_BLOCK_BYTES: u64 = 1024; const SUPPORTED_FILESYSTEM: &str = "ext4"; const SUBCOMMAND_SHIFT: u32 = 8; -const MOUNT_POINT_FIELD: usize = 4; const OCTAL_DIGIT_COUNT: usize = 3; const OCTAL_RADIX: u32 = 8; const REQUIRED_VALID_FIELDS: u32 = libc::QIF_LIMITS | libc::QIF_USAGE | libc::QIF_TIMES; @@ -27,11 +26,22 @@ pub(super) fn probe(path: &Path) -> Result { let mount = inspect_mount(path)?; // SAFETY: geteuid has no preconditions and does not mutate process state. let subject_id = unsafe { libc::geteuid() }; - match mount.filesystem.as_str() { - SUPPORTED_FILESYSTEM => probe_vfs(mount, path, subject_id), - lustre::FILESYSTEM => lustre::probe(mount, path, subject_id), + let snapshot = match mount.filesystem.as_str() { + SUPPORTED_FILESYSTEM => probe_vfs(mount.clone(), path, subject_id), + lustre::FILESYSTEM => lustre::probe(mount.clone(), path, subject_id), _ => Err(unsupported(&mount)), + }?; + // Detect ordinary concurrent replacement. A privileged mount controller can + // still arrange A -> B -> the same A; that hostile-root ABA is outside this + // reporting-only provider's threat boundary (see the Lustre module notes). + let rebound = inspect_mount(path)?; + if rebound != mount { + return Err(incomplete( + &mount, + "mount identity changed during quota probe", + )); } + Ok(snapshot) } fn probe_vfs(mount: MountInfo, path: &Path, subject_id: u32) -> Result { @@ -72,7 +82,15 @@ fn parse_mountinfo(input: &str, path: &Path) -> Option { fn parse_mount_line(line: &str) -> Option { let (mount, filesystem) = line.split_once(" - ")?; - let mount_point = mount.split_whitespace().nth(MOUNT_POINT_FIELD)?; + let mut mount_fields = mount.split_whitespace(); + let mount_id = mount_fields.next()?.parse().ok()?; + let _parent_id = mount_fields.next()?; + let device = mount_fields.next()?; + let (device_major, device_minor) = device.split_once(':')?; + let device_major = device_major.parse().ok()?; + let device_minor = device_minor.parse().ok()?; + let _root = mount_fields.next()?; + let mount_point = mount_fields.next()?; let mut fields = filesystem.split_whitespace(); let filesystem = fields.next()?.to_owned(); let source = fields.next()?; @@ -80,6 +98,9 @@ fn parse_mount_line(line: &str) -> Option { mount_point: decode_path(mount_point), filesystem, source: decode_path(source), + mount_id, + device_major, + device_minor, }) } @@ -232,6 +253,25 @@ mod tests { let mount = parse_mountinfo(input, Path::new("/home/me/My Data/project")).unwrap(); assert_eq!(mount.mount_point, Path::new("/home/me/My Data")); assert_eq!(mount.source, Path::new("/dev/loop0")); + assert_eq!(mount.mount_id, 40); + assert_eq!((mount.device_major, mount.device_minor), (7, 1)); + + let replacement = parse_mountinfo( + "41 36 7:2 / /home/me/My\\040Data rw - ext4 /dev/loop1 rw", + Path::new("/home/me/My Data/project"), + ) + .unwrap(); + assert_ne!(mount, replacement); + } + + #[test] + fn quota_mount_parser_rejects_missing_or_invalid_identity() { + let missing_device = "40 36 / /home rw - ext4 /dev/loop0 rw"; + let invalid_mount_id = "x 36 7:1 / /home rw - ext4 /dev/loop0 rw"; + let invalid_device = "40 36 7:x / /home rw - ext4 /dev/loop0 rw"; + for input in [missing_device, invalid_mount_id, invalid_device] { + assert!(parse_mountinfo(input, Path::new("/home/project")).is_none()); + } } #[test] @@ -279,6 +319,9 @@ mod tests { mount_point: PathBuf::from("/home"), filesystem: "ext4".to_owned(), source: PathBuf::from("/dev/root"), + mount_id: 36, + device_major: 8, + device_minor: 1, }; let not_configured = classify_error(&mount, std::io::Error::from_raw_os_error(libc::ESRCH)); diff --git a/crates/degu/src/quota/platform/linux/lustre.rs b/crates/degu/src/quota/platform/linux/lustre.rs index d2df2b9..76f83c1 100644 --- a/crates/degu/src/quota/platform/linux/lustre.rs +++ b/crates/degu/src/quota/platform/linux/lustre.rs @@ -1042,6 +1042,9 @@ mod tests { mount_point: PathBuf::from(MOUNT), filesystem: "lustre".to_owned(), source: PathBuf::from("10.0.0.1@tcp:/scratch"), + mount_id: 40, + device_major: 0, + device_minor: 42, } } @@ -1191,6 +1194,9 @@ mod tests { mount_point: dir.path().to_owned(), filesystem: "lustre".to_owned(), source: PathBuf::from("10.0.0.1@tcp:/scratch"), + mount_id: 40, + device_major: 0, + device_minor: 42, }; let error = verify_statfs_is_lustre(&mount).unwrap_err(); assert!(matches!(error, ProbeError::Incomplete { .. }), "{error:?}"); @@ -1202,6 +1208,9 @@ mod tests { mount_point: PathBuf::from("scratch"), filesystem: "lustre".to_owned(), source: PathBuf::from("10.0.0.1@tcp:/scratch"), + mount_id: 40, + device_major: 0, + device_minor: 42, }; let error = require_rooted_mount_point(&mount).unwrap_err(); assert!(matches!(error, ProbeError::Incomplete { .. }), "{error:?}"); diff --git a/crates/degu/src/quota_observation.rs b/crates/degu/src/quota_observation.rs new file mode 100644 index 0000000..7a80cd6 --- /dev/null +++ b/crates/degu/src/quota_observation.rs @@ -0,0 +1,1184 @@ +//! Quota observation around one permanent action batch. +//! +//! This module is reporting-only. Canonical anchors are probe inputs and never +//! mutation authority. + +use crate::action_result::{ + ActionId, ActionKind, ActionObservationTargets, ActionObservations, ActionResultOwner, + CompletedActionBatchResult, ContractError, NotStartedReason, ObservationRequestPath, + PlannedActionBatch, QuotaObservationState, QuotaObservationTarget, ResolvedQuotaObservation, + StartedActionOutcome, +}; +use crate::quota::{ProbeError, QuotaSnapshot}; +use serde::Serialize; +use std::path::{Path, PathBuf}; + +pub(crate) type CompletedQuotaAction = + CompletedActionBatchResult; +type NotAttemptedQuotaAction = CompletedActionBatchResult<(), (), ()>; + +#[derive(Debug, Eq, PartialEq)] +pub(crate) enum QuotaActionReport { + Attempted(CompletedQuotaAction), + NotAttempted(NotAttemptedQuotaAction), +} + +#[derive(Debug)] +pub(crate) enum ObservationPlanError { + Contract(ContractError), +} + +impl std::fmt::Display for ObservationPlanError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Contract(error) => write!(formatter, "invalid observation contract: {error:?}"), + } + } +} + +impl std::error::Error for ObservationPlanError {} + +pub(crate) fn planned_action( + owner: ActionResultOwner, + kind: ActionKind, + id: &'static str, + anchors: impl IntoIterator, +) -> Result { + let targets = anchors + .into_iter() + .map(ObservationRequestPath::new) + .map(QuotaObservationTarget::new) + .collect::>(); + Ok(PlannedActionBatch::new( + owner, + kind, + ActionId::new(id).map_err(ObservationPlanError::Contract)?, + ActionObservationTargets::new(targets), + )) +} + +pub(crate) fn not_attempted( + planned: PlannedActionBatch, + reason: NotStartedReason, +) -> QuotaActionReport { + QuotaActionReport::NotAttempted(planned.complete_not_started(reason)) +} + +pub(crate) fn not_attempted_action( + owner: ActionResultOwner, + kind: ActionKind, + id: &'static str, + anchors: impl IntoIterator, + reason: NotStartedReason, +) -> Result { + // A not-attempted action never dereferences an anchor, so a dry-run keeps its + // captured lexical targets without touching or requiring the filesystem. + Ok(not_attempted( + planned_action(owner, kind, id, anchors)?, + reason, + )) +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum ProbePhase { + Before, + After, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +pub(crate) struct UnavailableObservation { + pub(crate) phase: ProbePhase, + pub(crate) category: &'static str, + pub(crate) message: String, +} + +impl UnavailableObservation { + fn from_error(phase: ProbePhase, error: ProbeError) -> Self { + Self { + phase, + category: error.category(), + message: error.raw_message(), + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum IncomparableDimension { + ActiveState, + Provider, + DataSource, + Filesystem, + MountPoint, + ScopeIdentity, + SubjectKind, + SubjectId, + ObservationAnchor, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +pub(crate) struct ObservedSubject { + pub(crate) kind: &'static str, + pub(crate) id: u32, +} + +#[derive(Clone, Debug, Eq, PartialEq, Serialize)] +pub(crate) struct ObservedUsageDelta { + pub(crate) provider: &'static str, + pub(crate) data_source: &'static str, + pub(crate) filesystem: String, + pub(crate) mount_point: PathBuf, + pub(crate) subject: ObservedSubject, + pub(crate) space_used_before_bytes: u64, + pub(crate) space_used_after_bytes: u64, + pub(crate) space_used_delta_bytes: i128, + pub(crate) inodes_used_before: u64, + pub(crate) inodes_used_after: u64, + pub(crate) inodes_used_delta: i128, +} + +pub(crate) fn compare( + anchor: &Path, + before: &QuotaSnapshot, + after: &QuotaSnapshot, +) -> Result { + if before.state != "active" || after.state != "active" { + return Err(IncomparableDimension::ActiveState); + } + if before.provider != after.provider { + return Err(IncomparableDimension::Provider); + } + if before.data_source != after.data_source { + return Err(IncomparableDimension::DataSource); + } + if before.scope.filesystem != after.scope.filesystem { + return Err(IncomparableDimension::Filesystem); + } + if before.scope.mount_point != after.scope.mount_point { + return Err(IncomparableDimension::MountPoint); + } + if before.scope.identity != after.scope.identity { + return Err(IncomparableDimension::ScopeIdentity); + } + if before.subject.kind != after.subject.kind { + return Err(IncomparableDimension::SubjectKind); + } + if before.subject.id != after.subject.id { + return Err(IncomparableDimension::SubjectId); + } + if before.scope.path != anchor || after.scope.path != anchor { + return Err(IncomparableDimension::ObservationAnchor); + } + Ok(ObservedUsageDelta { + provider: before.provider, + data_source: before.data_source, + filesystem: before.scope.filesystem.clone(), + mount_point: before.scope.mount_point.clone(), + subject: ObservedSubject { + kind: before.subject.kind, + id: before.subject.id, + }, + space_used_before_bytes: before.space.used, + space_used_after_bytes: after.space.used, + space_used_delta_bytes: i128::from(after.space.used) - i128::from(before.space.used), + inodes_used_before: before.inodes.used, + inodes_used_after: after.inodes.used, + inodes_used_delta: i128::from(after.inodes.used) - i128::from(before.inodes.used), + }) +} + +#[derive(Clone, Debug, Eq, Hash, PartialEq)] +struct ScopeIdentity { + provider: &'static str, + data_source: &'static str, + filesystem: String, + mount_point: PathBuf, + provider_scope: crate::quota::model::QuotaScopeIdentity, + subject_kind: &'static str, + subject_id: u32, +} + +impl ScopeIdentity { + fn from_snapshot(snapshot: &QuotaSnapshot) -> Self { + Self { + provider: snapshot.provider, + data_source: snapshot.data_source, + filesystem: snapshot.scope.filesystem.clone(), + mount_point: snapshot.scope.mount_point.clone(), + provider_scope: snapshot.scope.identity.clone(), + subject_kind: snapshot.subject.kind, + subject_id: snapshot.subject.id, + } + } +} + +enum PreResolution { + Scope { + identity: ScopeIdentity, + anchors: Vec, + canonical: PathBuf, + before: Box, + }, + Unavailable { + anchor: ObservationRequestPath, + detail: UnavailableObservation, + }, +} + +fn canonicalize_before(anchor: &Path) -> Result { + if !anchor.is_absolute() { + return Err(UnavailableObservation { + phase: ProbePhase::Before, + category: "invalid_request", + message: format!( + "quota observation request is not absolute: {}", + anchor.display() + ), + }); + } + std::fs::canonicalize(anchor).map_err(|source| UnavailableObservation { + phase: ProbePhase::Before, + category: "canonicalize_io", + message: format!("failed to canonicalize {}: {source}", anchor.display()), + }) +} + +/// The only injectable quota-probe seam. Canonicalization is deliberately part +/// of this best-effort phase: neither it nor a provider failure can prevent the +/// execution closure. A successful pre probe binds post probing to that exact +/// canonical path. Pre-discovery must inspect each distinct requested anchor; +/// once identity is known, post probing occurs once per stable identity. +pub(crate) fn coordinate( + planned: PlannedActionBatch, + probe: &mut impl FnMut(&Path) -> Result, + execute: impl FnOnce() -> (R, StartedActionOutcome), +) -> (R, CompletedQuotaAction) { + let anchors = planned + .observation_targets() + .quota_scopes() + .iter() + .map(|target| target.anchor().clone()) + .collect::>(); + let mut pre = Vec::::new(); + for anchor in anchors { + let canonical = match canonicalize_before(anchor.as_path()) { + Ok(canonical) => canonical, + Err(detail) => { + pre.push(PreResolution::Unavailable { anchor, detail }); + continue; + } + }; + match probe(&canonical) { + Ok(before) => { + let identity = ScopeIdentity::from_snapshot(&before); + if let Some(PreResolution::Scope { anchors, .. }) = pre.iter_mut().find(|entry| { + matches!(entry, PreResolution::Scope { identity: existing, .. } if *existing == identity) + }) { + anchors.push(anchor); + } else { + pre.push(PreResolution::Scope { + identity, + anchors: vec![anchor], + canonical, + before: Box::new(before), + }); + } + } + Err(error) => pre.push(PreResolution::Unavailable { + anchor, + detail: UnavailableObservation::from_error(ProbePhase::Before, error), + }), + } + } + + let started = planned.start(); + let (result, outcome) = execute(); + let mut pending = started.finish_execution(outcome); + + let mut resolved = Vec::with_capacity(pre.len()); + for entry in pre { + match entry { + PreResolution::Scope { + anchors, + canonical, + before, + .. + } => { + let state = match probe(&canonical) { + Err(error) => QuotaObservationState::Unavailable( + UnavailableObservation::from_error(ProbePhase::After, error), + ), + Ok(after) => match compare(&canonical, &before, &after) { + Ok(delta) => match serde_json::to_value(&delta) { + Ok(_) => QuotaObservationState::Observed(delta), + Err(error) => { + QuotaObservationState::Unavailable(UnavailableObservation { + phase: ProbePhase::After, + category: "output_unrepresentable", + message: format!( + "quota observation cannot be represented in JSON: {error}" + ), + }) + } + }, + Err(dimension) => QuotaObservationState::Incomparable(dimension), + }, + }; + resolved.push(ResolvedQuotaObservation::new(anchors, state)); + } + PreResolution::Unavailable { anchor, detail } => { + // A pre-unavailable anchor has no before snapshot, so no post + // probe could yield a delta; it stays unavailable as observed. + resolved.push(ResolvedQuotaObservation::new( + [anchor], + QuotaObservationState::Unavailable(detail), + )); + } + } + } + let observations = resolved + .into_iter() + .collect::, _>>() + .and_then(|resolved| { + let ticket = pending.take_observation_ticket()?; + ActionObservations::resolve(ticket, resolved) + }); + let internal_unavailable = UnavailableObservation { + phase: ProbePhase::After, + category: "internal_contract", + message: observations.as_ref().err().map_or_else( + || "internal observation contract failure".to_owned(), + |error| format!("internal observation contract failure: {error:?}"), + ), + }; + let completed = pending.complete_or_all_unavailable(observations, internal_unavailable); + (result, completed) +} + +fn output_unrepresentable_json(message: &str) -> serde_json::Value { + serde_json::json!({ + "state": "unavailable", + "phase": ProbePhase::After, + "error_category": "output_unrepresentable", + "message": message, + }) +} + +pub(crate) fn json(report: &QuotaActionReport) -> serde_json::Value { + match report { + QuotaActionReport::Attempted(action) => serde_json::json!({ + "observation_state": "resolved", + "owner": owner_json(action.owner()), + "kind": kind_label(action.kind()), + "id": action.id().as_str(), + "quota_observations": action.observations().quota_scopes().iter().map(|scope| { + let detail = match scope.state() { + QuotaObservationState::NotAttempted => serde_json::json!({"state": "not_attempted"}), + QuotaObservationState::Unavailable(unavailable) => serde_json::json!({ + "state": "unavailable", + "phase": unavailable.phase, + "error_category": unavailable.category, + "message": unavailable.message, + }), + QuotaObservationState::Incomparable(dimension) => serde_json::json!({ + "state": "incomparable", + "dimension": dimension, + }), + QuotaObservationState::Observed(observed) => { + match serde_json::to_value(observed) { + Ok(mut value) => { + if let Some(object) = value.as_object_mut() { + object.insert( + "state".to_owned(), + serde_json::Value::String("observed".to_owned()), + ); + value + } else { + output_unrepresentable_json("observed quota delta was not an object") + } + } + Err(error) => output_unrepresentable_json(&error.to_string()), + } + } + }; + serde_json::json!({ + "anchors": scope.anchors().iter().map(|anchor| anchor.as_path().to_string_lossy().into_owned()).collect::>(), + "quota_observed_usage_delta": detail, + }) + }).collect::>(), + }), + QuotaActionReport::NotAttempted(action) => serde_json::json!({ + "observation_state": "not_attempted", + "owner": owner_json(action.owner()), + "kind": kind_label(action.kind()), + "id": action.id().as_str(), + "quota_observations": action.observations().quota_scopes().iter().map(|scope| { + debug_assert!(matches!(scope.state(), QuotaObservationState::NotAttempted)); + serde_json::json!({ + "anchors": scope.anchors().iter().map(|anchor| anchor.as_path().to_string_lossy().into_owned()).collect::>(), + "quota_observed_usage_delta": {"state": "not_attempted"}, + }) + }).collect::>(), + }), + } +} + +fn owner_json(owner: &ActionResultOwner) -> serde_json::Value { + match owner { + ActionResultOwner::CleanCommand => serde_json::json!("clean"), + ActionResultOwner::TrashPurgeCommand => serde_json::json!("trash_purge"), + ActionResultOwner::NativeAdapter { adapter_id } => { + serde_json::json!({"native_adapter": adapter_id.as_str()}) + } + } +} + +fn kind_label(kind: ActionKind) -> &'static str { + match kind { + ActionKind::DirectPurge => "direct_purge", + ActionKind::ExpiryPurge => "expiry_purge", + ActionKind::TrashPurge => "trash_purge", + ActionKind::Native => "native", + } +} + +#[derive(Debug, Eq, PartialEq)] +enum HumanObservationLine { + Stdout(String), + Warning(String), +} + +fn human_lines(report: &QuotaActionReport) -> Vec { + let QuotaActionReport::Attempted(action) = report else { + return Vec::new(); + }; + action + .observations() + .quota_scopes() + .iter() + .filter_map(|scope| { + let anchors = scope + .anchors() + .iter() + .map(|anchor| { + crate::presentation::escape_terminal_text( + &anchor.as_path().display().to_string(), + ) + }) + .collect::>() + .join(", "); + match scope.state() { + QuotaObservationState::Observed(delta) => Some(HumanObservationLine::Stdout( + format!( + "Observed quota usage change for {anchors}: {} bytes, {} inodes. Negative means usage decreased during the observation window; this is not attributed exclusively to degu.", + delta.space_used_delta_bytes, delta.inodes_used_delta + ), + )), + QuotaObservationState::Unavailable(detail) => { + Some(HumanObservationLine::Warning(format!( + "quota observation unavailable for {anchors} ({:?}, {}): {}", + detail.phase, + detail.category, + crate::presentation::escape_terminal_text(&detail.message) + ))) + } + QuotaObservationState::Incomparable(dimension) => { + Some(HumanObservationLine::Warning(format!( + "quota observations for {anchors} are incomparable: {} changed", + serde_json::to_value(dimension) + .ok() + .and_then(|value| value.as_str().map(str::to_owned)) + .unwrap_or_else(|| "identity".to_owned()) + ))) + } + QuotaObservationState::NotAttempted => None, + } + }) + .collect() +} + +pub(crate) fn print_warnings(report: &QuotaActionReport, colors: crate::runtime::OutputColors) { + for line in human_lines(report) { + if let HumanObservationLine::Warning(line) = line { + crate::presentation::print_stderr_note( + crate::presentation::Severity::Warning, + &line, + colors, + ); + } + } +} + +pub(crate) fn print_human( + report: &QuotaActionReport, + colors: crate::runtime::OutputColors, +) -> anyhow::Result<()> { + for line in human_lines(report) { + match line { + HumanObservationLine::Stdout(line) => crate::output::stdoutln!("{line}")?, + HumanObservationLine::Warning(line) => crate::presentation::print_stderr_note( + crate::presentation::Severity::Warning, + &line, + colors, + ), + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::action_result::{ + ActionId, ActionKind, ActionObservationTargets, ActionOutcome, ActionResultOwner, + QuotaObservationTarget, + }; + use crate::quota::model::{ + ActiveQuota, QuotaDimension, QuotaGrace, QuotaGraceState, QuotaLimits, QuotaScope, + QuotaScopeIdentity, + }; + use std::collections::VecDeque; + + fn snapshot(path: &str, used: u64, inodes: u64) -> QuotaSnapshot { + QuotaSnapshot::active( + QuotaScope::new( + PathBuf::from(path), + PathBuf::from("/home"), + "ext4".into(), + QuotaScopeIdentity::new(36, 8, 1, PathBuf::from("/dev/root")), + ), + 1000, + ActiveQuota { + provider: "linux_vfs", + data_source: "linux_quotactl", + space: QuotaDimension::new(used, QuotaLimits::new(0, 0), None), + inodes: QuotaDimension::new(inodes, QuotaLimits::new(0, 0), None), + }, + ) + } + + fn planned(paths: &[&str]) -> PlannedActionBatch { + let targets = ActionObservationTargets::new(paths.iter().map(|path| { + QuotaObservationTarget::new(ObservationRequestPath::new(PathBuf::from(path))) + })); + PlannedActionBatch::new( + ActionResultOwner::CleanCommand, + ActionKind::DirectPurge, + ActionId::new("test:purge").unwrap(), + targets, + ) + } + + #[test] + fn signed_edges_do_not_underflow_or_saturate() { + let low = snapshot("/home", 0, 0); + let high = snapshot("/home", u64::MAX, u64::MAX); + let increase = compare(Path::new("/home"), &low, &high).unwrap(); + let decrease = compare(Path::new("/home"), &high, &low).unwrap(); + assert_eq!(increase.space_used_delta_bytes, i128::from(u64::MAX)); + assert_eq!(decrease.space_used_delta_bytes, -i128::from(u64::MAX)); + assert_eq!(decrease.inodes_used_delta, -i128::from(u64::MAX)); + assert_eq!( + compare(Path::new("/home"), &low, &snapshot("/home", 0, 0)) + .unwrap() + .space_used_delta_bytes, + 0 + ); + let mut replies = VecDeque::from([ + Ok(snapshot("/", u64::MAX, u64::MAX)), + Ok(snapshot("/", 0, 0)), + ]); + let (_, completed) = coordinate( + planned(&["/"]), + &mut |_| replies.pop_front().unwrap(), + || ((), StartedActionOutcome::Success), + ); + let encoded = + serde_json::to_string(&json(&QuotaActionReport::Attempted(completed))).unwrap(); + let roundtrip: serde_json::Value = serde_json::from_str(&encoded).unwrap(); + assert_eq!( + roundtrip["quota_observations"][0]["quota_observed_usage_delta"] + ["space_used_delta_bytes"] + .to_string(), + "-18446744073709551615" + ); + } + + #[test] + fn limits_may_change_while_usage_remains_comparable() { + let before = snapshot("/home", 10, 2); + let mut after = snapshot("/home", 8, 1); + after.space.soft_limit = Some(99); + after.space.grace = Some(QuotaGrace { + state: QuotaGraceState::Expired, + expires_at_unix: None, + }); + assert_eq!( + compare(Path::new("/home"), &before, &after) + .unwrap() + .space_used_delta_bytes, + -2 + ); + } + + #[test] + fn mismatch_order_is_fail_closed() { + let before = snapshot("/home", 10, 2); + let mut after = snapshot("/wrong", 8, 1); + after.state = "inactive"; + after.provider = "other"; + assert_eq!( + compare(Path::new("/home"), &before, &after), + Err(IncomparableDimension::ActiveState) + ); + after.state = "active"; + assert_eq!( + compare(Path::new("/home"), &before, &after), + Err(IncomparableDimension::Provider) + ); + after.provider = before.provider; + after.data_source = "other"; + after.scope.filesystem = "xfs".into(); + after.scope.mount_point = "/other".into(); + after.subject.kind = "group"; + after.subject.id = 1001; + assert_eq!( + compare(Path::new("/home"), &before, &after), + Err(IncomparableDimension::DataSource) + ); + after.data_source = before.data_source; + assert_eq!( + compare(Path::new("/home"), &before, &after), + Err(IncomparableDimension::Filesystem) + ); + after.scope.filesystem = before.scope.filesystem.clone(); + assert_eq!( + compare(Path::new("/home"), &before, &after), + Err(IncomparableDimension::MountPoint) + ); + after.scope.mount_point = before.scope.mount_point.clone(); + after.scope.identity = QuotaScopeIdentity::new(37, 8, 2, PathBuf::from("/dev/replacement")); + assert_eq!( + compare(Path::new("/home"), &before, &after), + Err(IncomparableDimension::ScopeIdentity) + ); + after.scope.identity = before.scope.identity.clone(); + assert_eq!( + compare(Path::new("/home"), &before, &after), + Err(IncomparableDimension::SubjectKind) + ); + after.subject.kind = before.subject.kind; + assert_eq!( + compare(Path::new("/home"), &before, &after), + Err(IncomparableDimension::SubjectId) + ); + after.subject.id = before.subject.id; + assert_eq!( + compare(Path::new("/home"), &before, &after), + Err(IncomparableDimension::ObservationAnchor) + ); + } + + #[test] + fn every_identity_dimension_is_checked_independently() { + fn assert_mismatch( + mutate: impl FnOnce(&mut QuotaSnapshot), + expected: IncomparableDimension, + ) { + let before = snapshot("/home", 10, 2); + let mut after = snapshot("/home", 8, 1); + mutate(&mut after); + assert_eq!(compare(Path::new("/home"), &before, &after), Err(expected)); + } + assert_mismatch( + |after| after.state = "inactive", + IncomparableDimension::ActiveState, + ); + assert_mismatch( + |after| after.provider = "other", + IncomparableDimension::Provider, + ); + assert_mismatch( + |after| after.data_source = "other", + IncomparableDimension::DataSource, + ); + assert_mismatch( + |after| after.scope.filesystem = "xfs".into(), + IncomparableDimension::Filesystem, + ); + assert_mismatch( + |after| after.scope.mount_point = "/other".into(), + IncomparableDimension::MountPoint, + ); + assert_mismatch( + |after| { + after.scope.identity = + QuotaScopeIdentity::new(37, 8, 2, PathBuf::from("/dev/replacement")); + }, + IncomparableDimension::ScopeIdentity, + ); + assert_mismatch( + |after| after.subject.kind = "group", + IncomparableDimension::SubjectKind, + ); + assert_mismatch( + |after| after.subject.id = 1001, + IncomparableDimension::SubjectId, + ); + assert_mismatch( + |after| after.scope.path = "/other".into(), + IncomparableDimension::ObservationAnchor, + ); + } + + #[test] + fn coordinator_orders_pre_execute_post_and_keeps_post_on_failure() { + let events = std::cell::RefCell::new(Vec::new()); + let mut replies = VecDeque::from([Ok(snapshot("/", 10, 2)), Ok(snapshot("/", 7, 1))]); + let mut probe = |_: &Path| { + events.borrow_mut().push("probe"); + replies.pop_front().unwrap() + }; + let (value, completed) = coordinate(planned(&["/"]), &mut probe, || { + events.borrow_mut().push("execute"); + (42, StartedActionOutcome::Failure) + }); + assert_eq!(value, 42); + assert_eq!(*events.borrow(), ["probe", "execute", "probe"]); + assert!(matches!( + completed.observations().quota_scopes()[0].state(), + QuotaObservationState::Observed(delta) if delta.space_used_delta_bytes == -3 + )); + } + + #[test] + fn coordinator_observes_a_partially_failed_batch() { + let mut replies = VecDeque::from([Ok(snapshot("/", 10, 2)), Ok(snapshot("/", 6, 1))]); + let mut probe = |_: &Path| replies.pop_front().unwrap(); + let (_, completed) = coordinate(planned(&["/"]), &mut probe, || { + ((), StartedActionOutcome::Partial) + }); + assert_eq!(completed.outcome(), ActionOutcome::Partial); + assert!(matches!( + completed.observations().quota_scopes()[0].state(), + QuotaObservationState::Observed(delta) if delta.space_used_delta_bytes == -4 + )); + } + + #[test] + fn canonicalization_failure_is_before_unavailable_and_never_blocks_mutation() { + let temp = tempfile::tempdir().unwrap(); + let requested = temp.path().join("created-by-action"); + let action = planned_action( + ActionResultOwner::CleanCommand, + ActionKind::DirectPurge, + "test:canonicalize", + [requested.clone()], + ) + .unwrap(); + let mut calls = 0; + let (_, completed) = coordinate( + action, + &mut |path| { + calls += 1; + Ok(snapshot(path.to_str().unwrap(), 7, 1)) + }, + || { + std::fs::create_dir(&requested).unwrap(); + ((), StartedActionOutcome::Success) + }, + ); + assert_eq!( + calls, 0, + "an anchor whose pre canonicalization failed is never probed" + ); + assert!(requested.is_dir()); + assert!(matches!( + completed.observations().quota_scopes()[0].state(), + QuotaObservationState::Unavailable(UnavailableObservation { + phase: ProbePhase::Before, + category: "canonicalize_io", + .. + }) + )); + } + + #[test] + fn invalid_relative_request_cannot_mask_mutation_failure() { + let action = planned_action( + ActionResultOwner::CleanCommand, + ActionKind::DirectPurge, + "test:relative-request", + [PathBuf::from(".")], + ) + .unwrap(); + let mut probes = 0; + let (mutation, completed) = coordinate( + action, + &mut |_| { + probes += 1; + unreachable!("relative requests must never probe") + }, + || ("mutation-failed", StartedActionOutcome::Failure), + ); + assert_eq!(mutation, "mutation-failed"); + assert_eq!(probes, 0); + assert!(matches!( + completed.observations().quota_scopes()[0].state(), + QuotaObservationState::Unavailable(UnavailableObservation { + phase: ProbePhase::Before, + category: "invalid_request", + .. + }) + )); + } + + #[test] + fn lexical_parent_alias_is_canonicalized_with_filesystem_semantics() { + let temp = tempfile::tempdir().unwrap(); + let physical = temp.path().join("physical"); + let alias = temp.path().join("alias"); + let real = temp.path().join("real"); + std::fs::create_dir(&physical).unwrap(); + std::fs::create_dir(&real).unwrap(); + std::os::unix::fs::symlink(&physical, &alias).unwrap(); + let requested = alias.join("..").join("real"); + let action = planned_action( + ActionResultOwner::CleanCommand, + ActionKind::DirectPurge, + "test:parent-alias", + [requested], + ) + .unwrap(); + let canonical = std::fs::canonicalize(&real).unwrap(); + let mut probed = Vec::new(); + let (_, completed) = coordinate( + action, + &mut |path| { + probed.push(path.to_path_buf()); + Ok(snapshot(path.to_str().unwrap(), 10, 1)) + }, + || ((), StartedActionOutcome::Success), + ); + assert_eq!(probed, [canonical.clone(), canonical]); + assert!(matches!( + completed.observations().quota_scopes()[0].state(), + QuotaObservationState::Observed(_) + )); + } + + #[cfg(target_os = "linux")] + #[test] + fn non_utf8_mount_becomes_output_unrepresentable_without_panicking() { + use std::os::unix::ffi::OsStringExt; + let non_utf8_mount = PathBuf::from(std::ffi::OsString::from_vec(vec![b'/', 0xff])); + let mut before = snapshot("/", 10, 2); + before.scope.mount_point = non_utf8_mount.clone(); + let mut after = snapshot("/", 7, 2); + after.scope.mount_point = non_utf8_mount; + let mut replies = VecDeque::from([Ok(before), Ok(after)]); + let (_, completed) = coordinate( + planned(&["/"]), + &mut |_| replies.pop_front().unwrap(), + || ((), StartedActionOutcome::Failure), + ); + let report = QuotaActionReport::Attempted(completed); + assert!(matches!( + report, + QuotaActionReport::Attempted(ref completed) + if matches!(completed.observations().quota_scopes()[0].state(), + QuotaObservationState::Unavailable(UnavailableObservation { + category: "output_unrepresentable", .. + })) + )); + assert_eq!( + json(&report)["quota_observations"][0]["quota_observed_usage_delta"]["state"], + "unavailable" + ); + } + + #[test] + fn post_probe_retains_the_successful_pre_canonical_binding() { + use std::os::unix::fs::symlink; + let temp = tempfile::tempdir().unwrap(); + let first = temp.path().join("first"); + let second = temp.path().join("second"); + let requested = temp.path().join("requested"); + std::fs::create_dir(&first).unwrap(); + std::fs::create_dir(&second).unwrap(); + symlink(&first, &requested).unwrap(); + let action = planned_action( + ActionResultOwner::CleanCommand, + ActionKind::DirectPurge, + "test:canonical-binding", + [requested.clone()], + ) + .unwrap(); + let mut probed = Vec::new(); + let (_, completed) = coordinate( + action, + &mut |path| { + probed.push(path.to_path_buf()); + Ok(snapshot( + path.to_str().unwrap(), + 10 - probed.len() as u64, + 1, + )) + }, + || { + std::fs::rename(&requested, temp.path().join("old-request")).unwrap(); + symlink(&second, &requested).unwrap(); + ((), StartedActionOutcome::Success) + }, + ); + let first = std::fs::canonicalize(first).unwrap(); + assert_eq!(probed, [first.clone(), first]); + assert!(matches!( + completed.observations().quota_scopes()[0].state(), + QuotaObservationState::Observed(_) + )); + } + + #[test] + fn equal_identities_fold_despite_different_pre_usage_and_probe_post_once() { + let mut second_before = snapshot("/dev", 99, 42); + // Identity excludes the requested observation path and usage values. + second_before.scope.mount_point = "/home".into(); + let mut replies = VecDeque::from([ + Ok(snapshot("/", 10, 2)), + Ok(second_before), + Ok(snapshot("/", 7, 1)), + ]); + let mut calls = 0; + let (_, completed) = coordinate( + planned(&["/", "/dev"]), + &mut |_| { + calls += 1; + replies.pop_front().unwrap() + }, + || ((), StartedActionOutcome::Success), + ); + assert_eq!(calls, 3); + assert_eq!(completed.observations().quota_scopes().len(), 1); + assert_eq!( + completed.observations().quota_scopes()[0].anchors().len(), + 2 + ); + } + + #[test] + fn different_filesystems_are_never_folded() { + let a_before = snapshot("/", 10, 2); + let mut b_before = snapshot("/dev", 10, 2); + b_before.scope.mount_point = "/scratch".into(); + b_before.scope.filesystem = "xfs".into(); + let a_after = snapshot("/", 7, 1); + let mut b_after = snapshot("/dev", 7, 1); + b_after.scope.mount_point = "/scratch".into(); + b_after.scope.filesystem = "xfs".into(); + let mut replies = VecDeque::from([Ok(a_before), Ok(b_before), Ok(a_after), Ok(b_after)]); + let (_, completed) = coordinate( + planned(&["/", "/dev"]), + &mut |_| replies.pop_front().unwrap(), + || ((), StartedActionOutcome::Success), + ); + assert_eq!(completed.observations().quota_scopes().len(), 2); + } + + #[test] + fn replacement_mount_identities_are_never_folded() { + let a_before = snapshot("/", 10, 2); + let mut b_before = snapshot("/dev", 10, 2); + b_before.scope.identity = + QuotaScopeIdentity::new(37, 8, 2, PathBuf::from("/dev/replacement")); + let a_after = snapshot("/", 7, 1); + let mut b_after = snapshot("/dev", 7, 1); + b_after.scope.identity = b_before.scope.identity.clone(); + let mut replies = VecDeque::from([Ok(a_before), Ok(b_before), Ok(a_after), Ok(b_after)]); + let (_, completed) = coordinate( + planned(&["/", "/dev"]), + &mut |_| replies.pop_front().unwrap(), + || ((), StartedActionOutcome::Success), + ); + assert_eq!(completed.observations().quota_scopes().len(), 2); + } + + #[test] + fn pre_failure_does_not_block_execution_and_is_explicit() { + let mut executed = false; + let mut replies = VecDeque::from([ + Err(ProbeError::Unavailable { + filesystem: "ext4".into(), + mount_point: "/home".into(), + reason: "before failed".into(), + }), + Ok(snapshot("/", 7, 1)), + ]); + let (_, completed) = coordinate( + planned(&["/"]), + &mut |_| replies.pop_front().unwrap(), + || { + executed = true; + ((), StartedActionOutcome::Success) + }, + ); + assert!(executed); + assert!(matches!( + completed.observations().quota_scopes()[0].state(), + QuotaObservationState::Unavailable(UnavailableObservation { + phase: ProbePhase::Before, + category: "unavailable", + .. + }) + )); + } + + #[test] + fn observed_json_is_signed_and_keeps_the_b0_action_envelope() { + let mut replies = VecDeque::from([Ok(snapshot("/", 10, 2)), Ok(snapshot("/", 7, 1))]); + let (_, completed) = coordinate( + planned(&["/"]), + &mut |_| replies.pop_front().unwrap(), + || ((), StartedActionOutcome::Success), + ); + let value = json(&QuotaActionReport::Attempted(completed)); + assert_eq!(value["owner"], "clean"); + assert_eq!(value["kind"], "direct_purge"); + let delta = &value["quota_observations"][0]["quota_observed_usage_delta"]; + assert_eq!(delta["state"], "observed"); + assert_eq!(delta["space_used_delta_bytes"], -3); + assert_eq!(delta["inodes_used_delta"], -1); + assert_eq!( + delta["subject"], + serde_json::json!({"kind": "user", "id": 1000}) + ); + assert_eq!( + delta + .as_object() + .unwrap() + .keys() + .map(String::as_str) + .collect::>(), + [ + "data_source", + "filesystem", + "inodes_used_after", + "inodes_used_before", + "inodes_used_delta", + "mount_point", + "provider", + "space_used_after_bytes", + "space_used_before_bytes", + "space_used_delta_bytes", + "state", + "subject", + ] + .into_iter() + .collect() + ); + } + + #[test] + fn human_observed_copy_is_noncausal_and_warning_text_is_escaped() { + let mut observed_replies = + VecDeque::from([Ok(snapshot("/", 10, 2)), Ok(snapshot("/", 7, 1))]); + let (_, observed) = coordinate( + planned(&["/"]), + &mut |_| observed_replies.pop_front().unwrap(), + || ((), StartedActionOutcome::Success), + ); + let lines = human_lines(&QuotaActionReport::Attempted(observed)); + let HumanObservationLine::Stdout(line) = &lines[0] else { + panic!("expected observed stdout") + }; + assert!(line.starts_with("Observed quota usage change")); + assert!(line.contains("Negative means usage decreased")); + assert!(line.contains("not attributed exclusively to degu")); + + let mut unavailable_replies = VecDeque::from([ + Err(ProbeError::Unavailable { + filesystem: "ext4\u{1b}[31m".into(), + mount_point: "/home\nother".into(), + reason: "bad\tprobe".into(), + }), + Ok(snapshot("/", 7, 1)), + ]); + let (_, unavailable) = coordinate( + planned(&["/"]), + &mut |_| unavailable_replies.pop_front().unwrap(), + || ((), StartedActionOutcome::Success), + ); + let report = QuotaActionReport::Attempted(unavailable); + let machine = json(&report); + let detail = &machine["quota_observations"][0]["quota_observed_usage_delta"]; + assert!(detail["message"].as_str().unwrap().contains('\u{1b}')); + assert!(detail["message"].as_str().unwrap().contains('\n')); + assert_eq!( + detail + .as_object() + .unwrap() + .keys() + .map(String::as_str) + .collect::>(), + ["error_category", "message", "phase", "state"] + .into_iter() + .collect() + ); + let lines = human_lines(&report); + let HumanObservationLine::Warning(line) = &lines[0] else { + panic!("expected warning") + }; + assert!(!line.contains('\u{1b}')); + assert!(line.contains("ext4\\u{1b}[31m")); + assert!(line.contains("/home\\nother")); + assert!(line.contains("bad\\tprobe")); + } + + #[test] + fn incomparable_json_shape_is_frozen() { + let before = snapshot("/", 10, 2); + let mut after = snapshot("/", 7, 1); + after.subject.id = 2000; + let mut replies = VecDeque::from([Ok(before), Ok(after)]); + let (_, completed) = coordinate( + planned(&["/"]), + &mut |_| replies.pop_front().unwrap(), + || ((), StartedActionOutcome::Success), + ); + let value = json(&QuotaActionReport::Attempted(completed)); + let detail = &value["quota_observations"][0]["quota_observed_usage_delta"]; + assert_eq!( + detail, + &serde_json::json!({ + "state": "incomparable", + "dimension": "subject_id", + }) + ); + } + + #[test] + fn post_failure_is_explicit_and_does_not_replace_execution_result() { + let mut replies = VecDeque::from([ + Ok(snapshot("/", 10, 2)), + Err(ProbeError::Unavailable { + filesystem: "ext4".into(), + mount_point: "/home".into(), + reason: "after failed".into(), + }), + ]); + let (mutation, completed) = coordinate( + planned(&["/"]), + &mut |_| replies.pop_front().unwrap(), + || ("mutation-failed", StartedActionOutcome::Failure), + ); + assert_eq!(mutation, "mutation-failed"); + assert!(matches!( + completed.observations().quota_scopes()[0].state(), + QuotaObservationState::Unavailable(UnavailableObservation { + phase: ProbePhase::After, + .. + }) + )); + } +} diff --git a/crates/degu/tests/clean/expiry.rs b/crates/degu/tests/clean/expiry.rs index 400c76f..d491a21 100644 --- a/crates/degu/tests/clean/expiry.rs +++ b/crates/degu/tests/clean/expiry.rs @@ -1,6 +1,16 @@ use super::support::*; use std::os::unix::fs::PermissionsExt; +fn seed_aged_numeric_marker(state: &tempfile::TempDir) -> std::path::PathBuf { + let claims = private_trash_root(state).join(".claims"); + std::fs::create_dir_all(&claims).unwrap(); + std::fs::set_permissions(&claims, std::fs::Permissions::from_mode(0o700)).unwrap(); + let marker = claims.join("12345"); + let file = std::fs::File::create(&marker).unwrap(); + file.set_modified(expired_time()).unwrap(); + marker +} + fn seed_expired_interrupted_claim(state: &tempfile::TempDir) -> std::path::PathBuf { let trash = private_trash_root(state); let claims = trash.join(".claims"); @@ -23,6 +33,67 @@ fn seed_expired_interrupted_claim(state: &tempfile::TempDir) -> std::path::PathB claim } +#[test] +fn xdg_state_parent_alias_does_not_block_direct_or_expiry_mutation() { + let home = tempfile::tempdir().unwrap(); + let state = tempfile::tempdir().unwrap(); + let alias = state.path().join("alias"); + let real = state.path().join("real"); + std::fs::create_dir(&alias).unwrap(); + std::fs::create_dir(&real).unwrap(); + let request = alias.join("..").join("real"); + let cache = crate::common::platform_cache_dir(home.path(), "pip"); + std::fs::create_dir_all(&cache).unwrap(); + std::fs::write(cache.join("wheel.whl"), b"cache").unwrap(); + crate::common::make_tree_non_shared_writable(home.path()).unwrap(); + + let out = degu() + .env("HOME", home.path()) + .env("XDG_STATE_HOME", &request) + .args(["clean", "--purge", "--yes", "--json"]) + .output() + .unwrap(); + + assert!( + out.status.success(), + "{}", + String::from_utf8_lossy(&out.stderr) + ); + assert!(!cache.exists()); + let report: serde_json::Value = serde_json::from_slice(&out.stdout).unwrap(); + assert_eq!( + report["quota_observations"]["direct_purge"]["observation_state"], + "resolved" + ); + assert_eq!( + report["quota_observations"]["expiry_purge"]["observation_state"], + "resolved" + ); +} + +#[test] +fn clean_empty_entries_still_runs_observed_claim_housekeeping() { + let home = tempfile::tempdir().unwrap(); + let state = tempfile::tempdir().unwrap(); + let marker = seed_aged_numeric_marker(&state); + + let out = run_clean(&home, &state, &["clean", "--yes", "--json"]); + + assert!( + out.status.success(), + "{}", + String::from_utf8_lossy(&out.stderr) + ); + assert!(!marker.exists()); + let report: serde_json::Value = serde_json::from_slice(&out.stdout).unwrap(); + assert!(report["expiry"]["planned"].as_array().unwrap().is_empty()); + assert_eq!(report["expiry"]["attempted"], true); + assert_eq!( + report["quota_observations"]["expiry_purge"]["observation_state"], + "resolved" + ); +} + #[test] fn clean_purges_expired_trash_entries_after_report() { assert_json_expiry_after_report(); @@ -44,6 +115,17 @@ fn assert_json_expiry_after_report() { assert_eq!(report["expiry"]["planned"].as_array().unwrap().len(), 1); assert_eq!(report["expiry"]["purged"].as_array().unwrap().len(), 1); assert!(report["expiry"]["failed"].as_array().unwrap().is_empty()); + assert_eq!( + report["quota_observations"]["expiry_purge"]["kind"], + "expiry_purge" + ); + assert_eq!( + report["quota_observations"]["expiry_purge"]["quota_observations"] + .as_array() + .unwrap() + .len(), + 1 + ); assert!(!expired.exists()); assert_eq!( visible_trash_entries(&state.path().join("degu/trash")).len(), @@ -99,6 +181,10 @@ fn clean_empty_plan_json_still_purges_expired_trash() { assert_eq!(report["expiry"]["attempted"], true); assert_eq!(report["expiry"]["planned"].as_array().unwrap().len(), 1); assert_eq!(report["expiry"]["purged"].as_array().unwrap().len(), 1); + assert_eq!( + report["quota_observations"]["expiry_purge"]["kind"], + "expiry_purge" + ); assert!(!expired.exists()); assert!(visible_trash_entries(&state.path().join("degu/trash")).is_empty()); } @@ -142,6 +228,14 @@ fn clean_dry_run_previews_and_never_purges_expired_trash() { assert_eq!(report["expiry"]["attempted"], false); assert_eq!(report["expiry"]["planned"].as_array().unwrap().len(), 1); assert!(report["expiry"]["purged"].as_array().unwrap().is_empty()); + assert_eq!( + report["quota_observations"]["direct_purge"]["observation_state"], + "not_attempted" + ); + assert_eq!( + report["quota_observations"]["expiry_purge"]["observation_state"], + "not_attempted" + ); assert_eq!(oplog_records(&state).len(), 1); assert_eq!( visible_trash_entries(&state.path().join("degu/trash")).len(), diff --git a/crates/degu/tests/schema/clean_scan_summary.rs b/crates/degu/tests/schema/clean_scan_summary.rs index b40fa30..6019ebd 100644 --- a/crates/degu/tests/schema/clean_scan_summary.rs +++ b/crates/degu/tests/schema/clean_scan_summary.rs @@ -23,6 +23,12 @@ fn clean_json_schema_is_frozen() { assert!(json["expiry"]["retention_days"].is_number()); assert!(json["expiry"]["planned"].is_array()); assert!(json["expiry"]["purged"].is_array()); + assert_keys( + &json["quota_observations"], + &["direct_purge", "expiry_purge"], + ); + assert!(json["quota_observations"]["direct_purge"]["observation_state"].is_string()); + assert!(json["quota_observations"]["expiry_purge"]["observation_state"].is_string()); for failure in json["expiry"]["failed"].as_array().unwrap() { assert_keys(failure, CLEAN_EXPIRY_FAILURE_KEYS); } @@ -38,6 +44,35 @@ fn clean_json_schema_is_frozen() { } } +#[test] +fn clean_purge_quota_observation_is_one_batch_scope_not_per_finding() { + let home = tempfile::tempdir().unwrap(); + let state = tempfile::tempdir().unwrap(); + let _pip_cache = fake_pip_cache(&home); + let _hf_home = fake_huggingface_cache(&home); + let json = json_stdout( + degu() + .env("HOME", home.path()) + .env("XDG_STATE_HOME", state.path()) + .args(["clean", "--purge", "--include-review", "--yes", "--json"]) + .output() + .unwrap(), + ); + + let action = &json["quota_observations"]["direct_purge"]; + assert_keys(action, QUOTA_ACTION_KEYS); + assert_eq!(action["kind"], "direct_purge"); + let executed = json["executed"].as_array().unwrap().len(); + let scopes = assert_non_empty_array(&action["quota_observations"], "direct quota scopes"); + assert!( + executed > 1, + "folding needs multiple findings; got {executed}" + ); + assert_eq!(scopes.len(), 1, "one trash scope, not one per finding"); + assert_keys(&scopes[0], QUOTA_SCOPE_KEYS); + assert!(scopes[0]["quota_observed_usage_delta"]["state"].is_string()); +} + #[test] fn scan_summary_json_schema_is_frozen() { let home = tempfile::tempdir().unwrap(); diff --git a/crates/degu/tests/schema/operations.rs b/crates/degu/tests/schema/operations.rs index 6a34e9f..d1e4e3c 100644 --- a/crates/degu/tests/schema/operations.rs +++ b/crates/degu/tests/schema/operations.rs @@ -40,6 +40,40 @@ fn trash_purge_json_schema_is_frozen() { assert_keys(&json, TRASH_PURGE_REPORT_KEYS); assert_non_empty_array(&json["purged"], "trash purge purged entries"); json["failed"].as_array().unwrap(); + assert_quota_action(&json["quota_observations"]); +} + +#[test] +fn empty_trash_purge_json_explicitly_observes_housekeeping() { + let home = tempfile::tempdir().unwrap(); + let state = tempfile::tempdir().unwrap(); + let json = json_stdout( + degu() + .env("HOME", home.path()) + .env("XDG_STATE_HOME", state.path()) + .args(["trash", "purge", "--yes", "--json"]) + .output() + .unwrap(), + ); + assert_eq!(json["quota_observations"]["observation_state"], "resolved"); + assert_eq!( + json["quota_observations"]["quota_observations"] + .as_array() + .unwrap() + .len(), + 1 + ); +} + +fn assert_quota_action(action: &serde_json::Value) { + assert_keys(action, QUOTA_ACTION_KEYS); + assert!(action["id"].is_string()); + assert!(action["kind"].is_string()); + for scope in assert_non_empty_array(&action["quota_observations"], "quota scopes") { + assert_keys(scope, QUOTA_SCOPE_KEYS); + assert!(scope["anchors"].is_array()); + assert!(scope["quota_observed_usage_delta"]["state"].is_string()); + } } #[test] diff --git a/crates/degu/tests/schema/support.rs b/crates/degu/tests/schema/support.rs index 6ea3954..b349770 100644 --- a/crates/degu/tests/schema/support.rs +++ b/crates/degu/tests/schema/support.rs @@ -52,6 +52,7 @@ pub(super) const CLEAN_REPORT_KEYS: &[&str] = &[ "omitted", "opt_in", "planned", + "quota_observations", ]; pub(super) const CLEAN_EXPIRY_KEYS: &[&str] = &["attempted", "failed", "planned", "purged", "retention_days"]; @@ -102,7 +103,15 @@ pub(super) const TRASH_LIST_ROW_KEYS: &[&str] = &[ "lower_bound", "original", ]; -pub(super) const TRASH_PURGE_REPORT_KEYS: &[&str] = &["failed", "purged"]; +pub(super) const TRASH_PURGE_REPORT_KEYS: &[&str] = &["failed", "purged", "quota_observations"]; +pub(super) const QUOTA_ACTION_KEYS: &[&str] = &[ + "id", + "kind", + "observation_state", + "owner", + "quota_observations", +]; +pub(super) const QUOTA_SCOPE_KEYS: &[&str] = &["anchors", "quota_observed_usage_delta"]; pub(super) const OP_RECORD_KEYS_LEGACY: &[&str] = &[ "action", "bytes_allocated", @@ -129,6 +138,7 @@ pub(super) fn fake_huggingface_cache(home: &tempfile::TempDir) -> PathBuf { let repo = hf_home.join("hub/models--org--name"); std::fs::create_dir_all(repo.join("snapshots/main")).unwrap(); std::fs::write(repo.join("snapshots/main/model.bin"), [0u8; 8192]).unwrap(); + crate::common::make_tree_non_shared_writable(&hf_home).unwrap(); hf_home } diff --git a/crates/degu/tests/trash/purge.rs b/crates/degu/tests/trash/purge.rs index 7819823..8b4ae3a 100644 --- a/crates/degu/tests/trash/purge.rs +++ b/crates/degu/tests/trash/purge.rs @@ -1,7 +1,107 @@ use super::support::*; #[cfg(target_os = "linux")] use std::os::unix::ffi::OsStringExt; -use std::os::unix::fs::symlink; +use std::os::unix::fs::{PermissionsExt, symlink}; + +fn aged_claim_marker(state: &tempfile::TempDir) -> std::path::PathBuf { + let claims = private_trash_root(state).join(".claims"); + std::fs::create_dir_all(&claims).unwrap(); + std::fs::set_permissions(&claims, std::fs::Permissions::from_mode(0o700)).unwrap(); + let marker = claims.join("12345"); + let file = std::fs::File::create(&marker).unwrap(); + file.set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(8 * 86_400)) + .unwrap(); + marker +} + +#[test] +fn xdg_state_parent_alias_does_not_block_trash_purge() { + let home = tempfile::tempdir().unwrap(); + let state = tempfile::tempdir().unwrap(); + let alias = state.path().join("alias"); + let real = state.path().join("real"); + std::fs::create_dir(&alias).unwrap(); + std::fs::create_dir(&real).unwrap(); + let request = alias.join("..").join("real"); + let cache = crate::common::platform_cache_dir(home.path(), "pip"); + std::fs::create_dir_all(&cache).unwrap(); + std::fs::write(cache.join("wheel.whl"), b"cache").unwrap(); + crate::common::make_tree_non_shared_writable(home.path()).unwrap(); + let staged = degu() + .env("HOME", home.path()) + .env("XDG_STATE_HOME", &request) + .args(["clean", "--yes", "--json"]) + .output() + .unwrap(); + assert!( + staged.status.success(), + "{}", + String::from_utf8_lossy(&staged.stderr) + ); + + let out = degu() + .env("HOME", home.path()) + .env("XDG_STATE_HOME", &request) + .args(["trash", "purge", "--yes", "--json"]) + .output() + .unwrap(); + + assert!( + out.status.success(), + "{}", + String::from_utf8_lossy(&out.stderr) + ); + let report: serde_json::Value = serde_json::from_slice(&out.stdout).unwrap(); + assert_eq!(report["purged"].as_array().unwrap().len(), 1); + assert_eq!( + report["quota_observations"]["observation_state"], + "resolved" + ); +} + +#[test] +fn trash_json_empty_entries_still_runs_observed_claim_housekeeping() { + let home = tempfile::tempdir().unwrap(); + let state = tempfile::tempdir().unwrap(); + let marker = aged_claim_marker(&state); + + let out = run(&home, &state, &["trash", "purge", "--yes", "--json"]); + + assert!( + out.status.success(), + "{}", + String::from_utf8_lossy(&out.stderr) + ); + assert!(!marker.exists()); + let report: serde_json::Value = serde_json::from_slice(&out.stdout).unwrap(); + assert!(report["purged"].as_array().unwrap().is_empty()); + assert_eq!( + report["quota_observations"]["observation_state"], + "resolved" + ); +} + +#[test] +fn trash_human_empty_entries_still_runs_claim_housekeeping() { + let home = tempfile::tempdir().unwrap(); + let state = tempfile::tempdir().unwrap(); + let marker = aged_claim_marker(&state); + + let out = run(&home, &state, &["trash", "purge", "--yes"]); + + assert!( + out.status.success(), + "{}", + String::from_utf8_lossy(&out.stderr) + ); + assert!(!marker.exists()); + let stdout = String::from_utf8(out.stdout).unwrap(); + assert!( + stdout.contains("expired trash claim markers, if present, will be permanently deleted") + ); + assert!(stdout.contains("Purged 0 trash entries")); + assert!(!stdout.contains("Trash is empty.")); +} #[test] fn trash_purge_colors_the_permanent_deletion_plan() {