[02/03] service-backup: publish immutable BackupSets manifest-last and retain complete sets

crates/service-backup/src/domain.rs

@@ -13,6 +13,8 @@ mod destination;
 mod destination_error;
 mod policy;
 mod policy_error;
+mod retention_plan;
+mod set_key_layout;
 #[cfg(feature = "http-client")]
 mod snapshot_source;
 #[cfg(feature = "http-client")]
@@ -28,6 +30,8 @@ pub use destination::{BackupDestination, SchemeInfo, SUPPORTED_SCHEMES};
 pub use destination_error::DestinationError;
 pub use policy::{BackupPolicy, RetentionPolicy, ScheduledBackupPolicy};
 pub use policy_error::PolicyError;
+pub use retention_plan::{plan_retention, RetentionPlan, SetInventory, SetState};
+pub use set_key_layout::{SetKeyError, SetKeyLayout};
 #[cfg(feature = "http-client")]
 pub use snapshot_source::SnapshotSource;
 #[cfg(feature = "http-client")]

crates/service-backup/src/domain/retention_plan.rs

@@ -0,0 +1,97 @@
+//! Pure retention planning for backup sets in storage sinks.
+
+use std::time::{Duration, SystemTime};
+
+/// Completion state of a backup set in inventory.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SetState {
+    /// The set's manifest object exists.
+    Complete { created_at: SystemTime },
+    /// Objects exist under the set prefix, but no manifest exists.
+    Partial,
+}
+
+/// Description of one backup set prefix discovered in a sink.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SetInventory {
+    pub set_id: String,
+    pub state: SetState,
+    pub newest_object_at: SystemTime,
+}
+
+/// Decision on which backup sets to retain, prune, or delete as orphans.
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+pub struct RetentionPlan {
+    pub keep: Vec<String>,
+    pub delete_complete: Vec<String>,
+    pub delete_orphans: Vec<String>,
+}
+
+/// Plan retention across discovered sets deterministically.
+///
+/// Rules:
+/// - The `keep_latest` newest `Complete` sets are kept, with `keep_latest` raised
+///   to at least 1 so the plan never deletes the last complete set.
+/// - Older `Complete` sets go to `delete_complete`.
+/// - With zero `Complete` sets, nothing in `delete_complete` is produced.
+/// - A `Partial` set goes to `delete_orphans` only when `now - newest_object_at >= orphan_grace`,
+///   otherwise it is neither kept nor deleted (left out of all three lists).
+/// - An empty inventory produces an empty plan.
+/// - Ties on `created_at` break by `set_id`.
+/// - All output lists (`keep`, `delete_complete`, `delete_orphans`) are sorted by `set_id`.
+pub fn plan_retention(
+    inventory: &[SetInventory],
+    keep_latest: usize,
+    now: SystemTime,
+    orphan_grace: Duration,
+) -> RetentionPlan {
+    let mut complete_sets: Vec<(&str, SystemTime)> = Vec::new();
+    let mut delete_orphans: Vec<String> = Vec::new();
+
+    for item in inventory {
+        match &item.state {
+            SetState::Complete { created_at } => {
+                complete_sets.push((&item.set_id, *created_at));
+            }
+            SetState::Partial => {
+                if let Ok(age) = now.duration_since(item.newest_object_at) {
+                    if age >= orphan_grace {
+                        delete_orphans.push(item.set_id.clone());
+                    }
+                }
+            }
+        }
+    }
+
+    // Sort ascending by (created_at, set_id); newest sets are at the end.
+    complete_sets.sort_by(|a, b| a.1.cmp(&b.1).then_with(|| a.0.cmp(b.0)));
+
+    let effective_keep = keep_latest.max(1);
+    let mut keep = Vec::new();
+    let mut delete_complete = Vec::new();
+
+    if !complete_sets.is_empty() {
+        let keep_count = effective_keep.min(complete_sets.len());
+        let split_idx = complete_sets.len() - keep_count;
+        for (id, _) in &complete_sets[split_idx..] {
+            keep.push((*id).to_string());
+        }
+        for (id, _) in &complete_sets[..split_idx] {
+            delete_complete.push((*id).to_string());
+        }
+    }
+
+    // Never delete a partial set that has the same id as a complete set.
+    delete_orphans.retain(|orphan| !keep.contains(orphan) && !delete_complete.contains(orphan));
+
+    keep.sort();
+    delete_complete.sort();
+    delete_orphans.sort();
+    delete_orphans.dedup();
+
+    RetentionPlan {
+        keep,
+        delete_complete,
+        delete_orphans,
+    }
+}

crates/service-backup/src/domain/set_key_layout.rs

@@ -0,0 +1,192 @@
+//! Immutable, collision-safe layout of backup set keys in object storage sinks.
+
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+pub enum SetKeyError {
+    #[error("prefix has leading slash: {0:?}")]
+    LeadingSlashInPrefix(String),
+    #[error("prefix has trailing slash: {0:?}")]
+    TrailingSlashInPrefix(String),
+    #[error("prefix contains empty segment: {0:?}")]
+    EmptyPrefixSegment(String),
+    #[error("prefix segment is dot or dot-dot: {0:?}")]
+    DotSegmentInPrefix(String),
+    #[error("prefix segment contains invalid character: {0:?}")]
+    InvalidPrefixCharacter(String),
+
+    #[error("set id is empty")]
+    EmptySetId,
+    #[error("set id exceeds 128 bytes: {0}")]
+    SetIdTooLong(usize),
+    #[error("set id contains slash: {0:?}")]
+    SlashInSetId(String),
+    #[error("set id contains invalid character: {0:?}")]
+    InvalidSetIdCharacter(String),
+
+    #[error("artifact name is empty")]
+    EmptyArtifactName,
+    #[error("artifact name exceeds 256 bytes: {0}")]
+    ArtifactNameTooLong(usize),
+    #[error("artifact name has leading slash: {0:?}")]
+    LeadingSlashInArtifactName(String),
+    #[error("artifact name has trailing slash: {0:?}")]
+    TrailingSlashInArtifactName(String),
+    #[error("artifact name contains empty segment: {0:?}")]
+    EmptyArtifactNameSegment(String),
+    #[error("artifact name segment is dot or dot-dot: {0:?}")]
+    DotSegmentInArtifactName(String),
+    #[error("artifact name segment contains invalid character: {0:?}")]
+    InvalidArtifactNameCharacter(String),
+}
+
+fn is_valid_char(b: u8) -> bool {
+    b.is_ascii_alphanumeric() || b == b'.' || b == b'_' || b == b'-'
+}
+
+fn validate_prefix(prefix: &str) -> Result<(), SetKeyError> {
+    if prefix.is_empty() {
+        return Ok(());
+    }
+    if prefix.starts_with('/') {
+        return Err(SetKeyError::LeadingSlashInPrefix(prefix.to_string()));
+    }
+    if prefix.ends_with('/') {
+        return Err(SetKeyError::TrailingSlashInPrefix(prefix.to_string()));
+    }
+    for seg in prefix.split('/') {
+        if seg.is_empty() {
+            return Err(SetKeyError::EmptyPrefixSegment(prefix.to_string()));
+        }
+        if seg == "." || seg == ".." {
+            return Err(SetKeyError::DotSegmentInPrefix(prefix.to_string()));
+        }
+        if !seg.bytes().all(is_valid_char) {
+            return Err(SetKeyError::InvalidPrefixCharacter(prefix.to_string()));
+        }
+    }
+    Ok(())
+}
+
+fn validate_set_id(set_id: &str) -> Result<(), SetKeyError> {
+    if set_id.is_empty() {
+        return Err(SetKeyError::EmptySetId);
+    }
+    if set_id.len() > 128 {
+        return Err(SetKeyError::SetIdTooLong(set_id.len()));
+    }
+    if set_id.contains('/') {
+        return Err(SetKeyError::SlashInSetId(set_id.to_string()));
+    }
+    if !set_id.bytes().all(is_valid_char) {
+        return Err(SetKeyError::InvalidSetIdCharacter(set_id.to_string()));
+    }
+    Ok(())
+}
+
+fn validate_artifact_name(name: &str) -> Result<(), SetKeyError> {
+    if name.is_empty() {
+        return Err(SetKeyError::EmptyArtifactName);
+    }
+    if name.len() > 256 {
+        return Err(SetKeyError::ArtifactNameTooLong(name.len()));
+    }
+    if name.starts_with('/') {
+        return Err(SetKeyError::LeadingSlashInArtifactName(name.to_string()));
+    }
+    if name.ends_with('/') {
+        return Err(SetKeyError::TrailingSlashInArtifactName(name.to_string()));
+    }
+    for seg in name.split('/') {
+        if seg.is_empty() {
+            return Err(SetKeyError::EmptyArtifactNameSegment(name.to_string()));
+        }
+        if seg == "." || seg == ".." {
+            return Err(SetKeyError::DotSegmentInArtifactName(name.to_string()));
+        }
+        if !seg.bytes().all(is_valid_char) {
+            return Err(SetKeyError::InvalidArtifactNameCharacter(name.to_string()));
+        }
+    }
+    Ok(())
+}
+
+/// Key layout mapping set ids to immutable object storage keys.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct SetKeyLayout {
+    prefix: String,
+}
+
+impl SetKeyLayout {
+    /// Construct a layout under `prefix`.
+    ///
+    /// An empty prefix places keys at top-level. Non-empty prefixes must be
+    /// `/`-separated segments of `[A-Za-z0-9._-]` without leading/trailing `/`
+    /// or empty/dot segments.
+    pub fn new(prefix: &str) -> Result<Self, SetKeyError> {
+        validate_prefix(prefix)?;
+        Ok(Self {
+            prefix: prefix.to_string(),
+        })
+    }
+
+    /// Return the configured prefix.
+    pub fn prefix(&self) -> &str {
+        &self.prefix
+    }
+
+    /// Key for an immutable artifact within a set: `<prefix>/sets/<set_id>/artifacts/<name>`.
+    pub fn artifact_key(&self, set_id: &str, name: &str) -> Result<String, SetKeyError> {
+        validate_set_id(set_id)?;
+        validate_artifact_name(name)?;
+        if self.prefix.is_empty() {
+            Ok(format!("sets/{set_id}/artifacts/{name}"))
+        } else {
+            Ok(format!("{}/sets/{set_id}/artifacts/{name}", self.prefix))
+        }
+    }
+
+    /// Key for the manifest marking set completion: `<prefix>/sets/<set_id>/manifest.json`.
+    pub fn manifest_key(&self, set_id: &str) -> Result<String, SetKeyError> {
+        validate_set_id(set_id)?;
+        if self.prefix.is_empty() {
+            Ok(format!("sets/{set_id}/manifest.json"))
+        } else {
+            Ok(format!("{}/sets/{set_id}/manifest.json", self.prefix))
+        }
+    }
+
+    /// Key for the latest successful backup pointer: `<prefix>/latest.json`.
+    pub fn latest_key(&self) -> String {
+        if self.prefix.is_empty() {
+            "latest.json".to_string()
+        } else {
+            format!("{}/latest.json", self.prefix)
+        }
+    }
+
+    /// Common prefix for all objects in a set: `<prefix>/sets/<set_id>/`.
+    pub fn set_prefix(&self, set_id: &str) -> Result<String, SetKeyError> {
+        validate_set_id(set_id)?;
+        if self.prefix.is_empty() {
+            Ok(format!("sets/{set_id}/"))
+        } else {
+            Ok(format!("{}/sets/{set_id}/", self.prefix))
+        }
+    }
+
+    /// Parse the set id from any key that lies under `<prefix>/sets/<id>/`.
+    ///
+    /// Returns `None` if the key does not match this layout's prefix or does
+    /// not lie directly under a valid set prefix.
+    pub fn parse_set_id<'a>(&self, key: &'a str) -> Option<&'a str> {
+        let rem = if self.prefix.is_empty() {
+            key.strip_prefix("sets/")?
+        } else {
+            key.strip_prefix(&self.prefix)?.strip_prefix("/sets/")?
+        };
+        let (id, _rest) = rem.split_once('/')?;
+        if id.is_empty() || id.len() > 128 || !id.bytes().all(is_valid_char) {
+            return None;
+        }
+        Some(id)
+    }
+}

crates/service-backup/src/lib.rs

@@ -27,8 +27,9 @@ pub use app::run_admin_snapshot_backup;
 pub use app::sink_from_destination;
 pub use application::{run_backup_once, BackupObject, BackupRunResult};
 pub use domain::{
-    BackupDestination, BackupPolicy, BackupSink, BackupSinkError, DestinationError, PolicyError,
-    RetentionPolicy, ScheduledBackupPolicy, SchemeInfo, SUPPORTED_SCHEMES,
+    plan_retention, BackupDestination, BackupPolicy, BackupSink, BackupSinkError, DestinationError,
+    PolicyError, RetentionPlan, RetentionPolicy, ScheduledBackupPolicy, SchemeInfo, SetInventory,
+    SetKeyError, SetKeyLayout, SetState, SUPPORTED_SCHEMES,
 };
 pub use infrastructure::{fetch_backup_object, GcsSink, LocalFsSink, UnsupportedCloudSink};
 #[cfg(feature = "http-client")]

crates/service-backup/tests/it/main.rs

@@ -5,5 +5,6 @@ mod admin_snapshot_backup;
 #[cfg(feature = "http-client")]
 mod projected_admin_transport;
 mod schema_golden;
+mod set_layout;
 mod sink_error;
 mod wire_golden;

crates/service-backup/tests/it/projected_admin_transport.rs

@@ -82,7 +82,7 @@ async fn non_success_diagnostic_is_bounded_and_structured() {
         .unwrap_err();
     let diagnostic = error.diagnostic().expect("status error has diagnostics");
     assert_eq!(diagnostic.status, 503);
-    assert_eq!(diagnostic.body.as_bytes().len(), 10);
+    assert_eq!(diagnostic.body.len(), 10);
     assert!(diagnostic.truncated);
     assert_eq!(diagnostic.operation.as_str(), "backup");
 }