feat: mass-deletion protection (cascade guard, per-repo breaker, batched deletes)

.mise/config.toml

@@ -101,7 +101,7 @@ set -eu
 # the worst offender — so a NEW overly-complex function fails immediately
 # even while the legacy offenders are still being paid down. Refactored one
 # under the threshold? Lower BUDGET to match (the task reminds you).
-BUDGET=31
+BUDGET=29
 COUNT=$(cargo clippy --workspace --all-targets --all-features --locked -q -- -W clippy::cognitive_complexity 2>&1 \
   | grep -A1 'cognitive complexity of' | grep -oE '[-]{2}> [^ ]+' | sort -u | wc -l)
 echo "==> functions above the cognitive-complexity threshold: $COUNT (budget: $BUDGET)"

CLAUDE.md

@@ -92,7 +92,7 @@ add controller-runtime dependencies to `crates/api`.
 | Observability       | `kopiur-telemetry` crate: instrument once on the OTel API → Prometheus pull (`/metrics`) + optional OTLP push; all metrics `kopiur_*`; OTLP env-gated/off by default; one span per reconcile. See `docs/dev/observability.md`                                                                                        |
 | API version         | `v1alpha1` only; no conversion webhooks yet                                                                                                                                                                                                                                                                          |
 | Retention           | GFS-only (`SnapshotPolicy.spec.retention`); failures bounded by flat `failedJobsHistoryLimit`; NO `successfulJobsHistoryLimit`                                                                                                                                                                                       |
-| Deletion            | `Snapshot` CR owns its kopia snapshot via finalizer; `deletionPolicy: Delete`(default produced) / `Retain`(forced for discovered) / `Orphan`                                                                                                                                                                         |
+| Deletion            | `Snapshot` CR owns its kopia snapshot via finalizer; `deletionPolicy: Delete`(default produced) / `Retain`(forced for discovered) / `Orphan`. Schedule-deletion cascade guarded by `SnapshotSchedule.spec.deletion.onScheduleDelete` (default `Retain`, opt-in `Delete`). Per-repo mass-deletion breaker `deletionProtection.threshold` (default 10, `0` disables, timestamp-ack release) holds bulk EXTERNAL deletes; operator prunes always bypass it. Execution is batched per repository (`snapdel-*` mover Jobs), never one Job per Snapshot — see ADR-0006.                                       |
 | Maintenance         | Default-managed: `Repository`/`ClusterRepository` `spec.maintenance` (default-on) is projected into an _owned_ `Maintenance` CR; an externally-authored `Maintenance` is always honored (never duplicated), even with `enabled: false`. ClusterRepo placement: `spec.maintenance.namespace` else `KOPIUR_NAMESPACE`. |
 
 Pinned deps (Rust 1.95): `kube` 4.0, `k8s-openapi` 0.28 (feature `v1_33`,

crates/api/src/cluster_repository.rs

@@ -7,9 +7,9 @@
 
 use crate::backend::Backend;
 use crate::common::{
-    CatalogBounds, CreateBehavior, Encryption, IdentityDefaults, MoverDefaults,
-    NamespaceDeletePolicy, RepositoryMode, ScheduleDefaults, default_namespace_delete_policy,
-    default_repository_mode,
+    CatalogBounds, CreateBehavior, DeletionProtectionSpec, Encryption, IdentityDefaults,
+    MoverDefaults, NamespaceDeletePolicy, RepositoryMode, ScheduleDefaults,
+    default_namespace_delete_policy, default_repository_mode,
 };
 use crate::maintenance::RepositoryMaintenanceSpec;
 use crate::repository::{
@@ -91,6 +91,9 @@ pub struct ClusterRepositorySpec {
     #[serde(default = "default_namespace_delete_policy")]
     #[schemars(default = "default_namespace_delete_policy")]
     pub on_namespace_delete: NamespaceDeletePolicy,
+    /// Mass-deletion circuit breaker for this repository's Snapshots.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub deletion_protection: Option<DeletionProtectionSpec>,
     /// Repository-owner gate for projecting credential Secrets into a foreign consumer namespace.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub credential_projection: Option<ClusterRepoCredentialProjection>,
@@ -379,6 +382,61 @@ identityDefaults:
         );
     }
 
+    #[test]
+    fn deletion_protection_threshold_schema_default_matches_the_constant() {
+        let crd = ClusterRepository::crd();
+        let json = serde_json::to_value(&crd).unwrap();
+        let spec = &json["spec"]["versions"][0]["schema"]["openAPIV3Schema"]["properties"]["spec"];
+        assert_eq!(
+            spec["properties"]["deletionProtection"]["properties"]["threshold"]["default"],
+            serde_json::json!(crate::consts::DEFAULT_MASS_DELETION_THRESHOLD)
+        );
+        assert_eq!(
+            crate::consts::effective_mass_deletion_threshold(None),
+            crate::consts::DEFAULT_MASS_DELETION_THRESHOLD
+        );
+    }
+
+    #[test]
+    fn deletion_protection_round_trips_on_cluster_repository() {
+        let yaml = r#"
+backend: { filesystem: { path: /repo } }
+encryption: { passwordSecretRef: { name: s, namespace: kopia-system } }
+allowedNamespaces: { all: true }
+deletionProtection:
+  threshold: 0
+"#;
+        let spec: ClusterRepositorySpec = from_yaml(yaml);
+        assert_eq!(
+            spec.deletion_protection.as_ref().and_then(|d| d.threshold),
+            Some(0)
+        );
+        assert_eq!(
+            crate::consts::effective_mass_deletion_threshold(spec.deletion_protection.as_ref()),
+            0,
+            "Some(0) must pass through as the disable sentinel"
+        );
+        let json = serde_json::to_value(&spec).expect("serialize");
+        assert_eq!(json["deletionProtection"]["threshold"], 0);
+        let reparsed: ClusterRepositorySpec = serde_json::from_value(json).expect("reparse");
+        assert_eq!(spec, reparsed);
+
+        // Absent stays None and is elided.
+        let bare: ClusterRepositorySpec = from_yaml(
+            "backend: { filesystem: { path: /repo } }\n\
+             encryption: { passwordSecretRef: { name: s, namespace: kopia-system } }\n\
+             allowedNamespaces: { all: true }\n",
+        );
+        assert!(bare.deletion_protection.is_none());
+        assert!(
+            serde_json::to_value(&bare)
+                .unwrap()
+                .get("deletionProtection")
+                .is_none(),
+            "absent deletionProtection must be elided"
+        );
+    }
+
     #[test]
     fn catalog_foreign_snapshots_round_trips_on_cluster_repository() {
         use crate::common::ForeignSnapshots;

crates/api/src/common/mod.rs

@@ -531,6 +531,54 @@ pub enum NamespaceDeletePolicy {
     Delete,
 }
 
+/// What the deletion of a `SnapshotSchedule` does to the `Snapshot` CRs it
+/// produced (which Kubernetes GC cascade-deletes via their ownerReference).
+/// Default `Retain`: the CRs are removed but their kopia snapshots survive and
+/// the catalog rediscovers them as `origin: discovered`. `Delete` opts into the
+/// cascade: each Snapshot's own `deletionPolicy` applies.
+///
+/// Deliberately 2-variant (not reusing [`DeletionPolicy`]): an `Orphan` in
+/// cascade position would differ from `Retain` only in per-CR event/metric
+/// bookkeeping — an invalid state made unrepresentable. The guard's `Retain`
+/// is exactly `DeletionPolicy::Retain`'s semantics (CR removed, kopia snapshot
+/// stays, catalog rediscovers it), deliberately NOT the `Orphan` event storm
+/// (no per-CR "orphaned" event/metric for every produced Snapshot).
+#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, Default, JsonSchema)]
+pub enum ScheduleDeletePolicy {
+    /// Keep the kopia snapshots: a produced Snapshot whose effective
+    /// deletionPolicy is `Delete` is downgraded to retain when its owning
+    /// schedule is gone (the fail-safe default).
+    #[default]
+    Retain,
+    /// Cascade: each produced Snapshot's own `deletionPolicy` applies even when
+    /// the owning schedule is gone (subject to the mass-deletion breaker).
+    Delete,
+}
+
+/// Mass-deletion circuit breaker for this repository's Snapshots.
+#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, Default, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct DeletionProtectionSpec {
+    /// Pending EXTERNAL destructive Snapshot deletions (deletionTimestamp set,
+    /// effective deletionPolicy Delete, not operator-pruned) that trip the
+    /// breaker for this repository: at or above this, those deletions are HELD
+    /// (finalizers wait) until acknowledged via the
+    /// `kopiur.home-operations.com/allow-mass-deletion` annotation.
+    /// `0` disables the breaker. Default 10.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    #[schemars(default = "default_mass_deletion_threshold")]
+    pub threshold: Option<u32>,
+}
+
+/// schemars default for [`DeletionProtectionSpec::threshold`] —
+/// [`DEFAULT_MASS_DELETION_THRESHOLD`](crate::consts::DEFAULT_MASS_DELETION_THRESHOLD)
+/// (`10`), matching `effective_mass_deletion_threshold`'s absent→CONST
+/// resolution. Returns the field's `Option` type so schemars 1 emits the
+/// schema `default:`.
+fn default_mass_deletion_threshold() -> Option<u32> {
+    Some(crate::consts::DEFAULT_MASS_DELETION_THRESHOLD)
+}
+
 /// Repository access mode; `ReadOnly` serves restores only (no backups, no maintenance).
 ///
 /// ```

crates/api/src/consts.rs

@@ -87,6 +87,29 @@ pub const RUN_MODE_ANNOTATION: &str = "kopiur.home-operations.com/run-mode";
 pub const ALLOW_IDENTITY_CHANGE_ANNOTATION: &str =
     "kopiur.home-operations.com/allow-identity-change";
 
+/// Marks a `Snapshot` the OPERATOR is deleting as part of its own lifecycle,
+/// stamped immediately before the delete call. Values: `PrunedBy` annotation
+/// values (`retention`, `failed-history`). The Snapshot finalizer uses it to
+/// distinguish Kopiur's own prunes from external deletions (GC cascades,
+/// kubectl, third-party controllers); external destructive deletions are
+/// subject to the schedule-cascade guard and the mass-deletion breaker,
+/// operator prunes are not. Any unrecognized value is treated as EXTERNAL
+/// (fail-safe). Wire-visible: users and tooling may read it on terminating CRs.
+pub const PRUNED_BY_ANNOTATION: &str = "kopiur.home-operations.com/pruned-by";
+
+/// Acknowledges a mass-deletion wave on a `Repository`/`ClusterRepository`.
+/// Value: an RFC3339 timestamp. A HELD external deletion is released iff its
+/// Snapshot's `metadata.deletionTimestamp` <= this value — "I approve what is
+/// pending NOW". Deliberately VALUED (unlike the presence-only
+/// allow-identity-change ack, consumed at a single admission instant): this
+/// annotation is read continuously by the controller, so a presence-only ack
+/// left behind (or committed to Git) would disarm the breaker forever. With
+/// the timestamp, a stale ack is inert against any LATER wave, nothing ever
+/// needs to remove it, and the operator never edits user metadata. The
+/// controller clamps the effective value to <= its own now (clock-skew guard);
+/// an unparseable value is ignored (Warning event on the repository).
+pub const ALLOW_MASS_DELETION_ANNOTATION: &str = "kopiur.home-operations.com/allow-mass-deletion";
+
 /// The API version string for kopiur CRDs (used in mover `TargetRef`s and
 /// `kubectl -o name`-style output).
 pub const API_VERSION: &str = "kopiur.home-operations.com/v1alpha1";
@@ -191,6 +214,22 @@ pub fn effective_failed_jobs_history_limit(limit: Option<u32>) -> u32 {
     limit.unwrap_or(DEFAULT_FAILED_JOBS_HISTORY_LIMIT)
 }
 
+/// Default `spec.deletionProtection.threshold` (0 disables). 10 pending
+/// external destructive deletions is far above legitimate manual cleanup but
+/// far below a tooling-driven cascade (the motivating incident was ~600).
+pub const DEFAULT_MASS_DELETION_THRESHOLD: u32 = 10;
+
+/// The effective mass-deletion breaker threshold: `deletionProtection.threshold`
+/// when set, else [`DEFAULT_MASS_DELETION_THRESHOLD`]. `Some(0)` disables the breaker.
+pub fn effective_mass_deletion_threshold(p: Option<&crate::common::DeletionProtectionSpec>) -> u32 {
+    p.and_then(|d| d.threshold)
+        .unwrap_or(DEFAULT_MASS_DELETION_THRESHOLD)
+}
+
+/// `Repository`/`ClusterRepository` condition: pending external destructive
+/// deletions for this repository are at/above the breaker threshold and held.
+pub const MASS_DELETION_HELD_CONDITION: &str = "MassDeletionHeld";
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -219,6 +258,8 @@ mod tests {
             RUN_REQUESTED_ANNOTATION,
             RUN_MODE_ANNOTATION,
             ALLOW_IDENTITY_CHANGE_ANNOTATION,
+            PRUNED_BY_ANNOTATION,
+            ALLOW_MASS_DELETION_ANNOTATION,
         ] {
             assert!(s.starts_with(crate::GROUP), "{s} must be group-prefixed");
         }

crates/api/src/error.rs

@@ -82,6 +82,20 @@ pub enum ValidationError {
         got: String,
     },
 
+    /// A `Snapshot` with `origin: discovered` set `spec.onScheduleDelete`. Discovered
+    /// snapshots carry an empty spec and their owner is a repository, not a
+    /// `SnapshotSchedule` — a stamped cascade policy on one is meaningless, so it is
+    /// forbidden, exactly like a non-`Retain` `deletionPolicy` ([`Self::DiscoveredMustRetain`]).
+    #[error(
+        "origin: discovered snapshots must not set onScheduleDelete (got {got:?}); a discovered \
+         snapshot has no owning SnapshotSchedule for this field to apply to. Remove \
+         spec.onScheduleDelete"
+    )]
+    DiscoveredCannotSetOnScheduleDelete {
+        /// The rejected `onScheduleDelete` value that was set.
+        got: String,
+    },
+
     /// A `Restore` with `source.identity` did not set `spec.repository`. Identity
     /// sources cannot derive a repository, so it is required (ADR §3.6/§4.6).
     #[error(

crates/api/src/repository.rs

@@ -2,9 +2,9 @@
 
 use crate::backend::Backend;
 use crate::common::{
-    CatalogBounds, CreateBehavior, Encryption, FailurePolicy, IdentityDefaults, MoverDefaults,
-    NamespaceDeletePolicy, RepositoryMode, ScheduleDefaults, default_namespace_delete_policy,
-    default_repository_mode,
+    CatalogBounds, CreateBehavior, DeletionProtectionSpec, Encryption, FailurePolicy,
+    IdentityDefaults, MoverDefaults, NamespaceDeletePolicy, RepositoryMode, ScheduleDefaults,
+    default_namespace_delete_policy, default_repository_mode,
 };
 use crate::maintenance::RepositoryMaintenanceSpec;
 use crate::server::{ServerSpec, ServerStatus};
@@ -91,6 +91,9 @@ pub struct RepositorySpec {
     #[serde(default = "default_namespace_delete_policy")]
     #[schemars(default = "default_namespace_delete_policy")]
     pub on_namespace_delete: NamespaceDeletePolicy,
+    /// Mass-deletion circuit breaker for this repository's Snapshots.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub deletion_protection: Option<DeletionProtectionSpec>,
     /// Access mode: `ReadWrite` (default) or `ReadOnly` (serves restores only).
     #[serde(default = "default_repository_mode")]
     #[schemars(default = "default_repository_mode")]
@@ -684,6 +687,68 @@ health:
         );
     }
 
+    #[test]
+    fn deletion_protection_threshold_schema_default_matches_the_constant() {
+        // Mirrors repository_schema_emits_context_free_defaults: a context-free
+        // default is safe to server-side-materialize because
+        // effective_mass_deletion_threshold maps absent → this same value.
+        let crd = Repository::crd();
+        let json = serde_json::to_value(&crd).unwrap();
+        let spec = &json["spec"]["versions"][0]["schema"]["openAPIV3Schema"]["properties"]["spec"];
+        assert_eq!(
+            spec["properties"]["deletionProtection"]["properties"]["threshold"]["default"],
+            serde_json::json!(crate::consts::DEFAULT_MASS_DELETION_THRESHOLD)
+        );
+        assert_eq!(
+            crate::consts::effective_mass_deletion_threshold(None),
+            crate::consts::DEFAULT_MASS_DELETION_THRESHOLD
+        );
+    }
+
+    #[test]
+    fn deletion_protection_round_trips_and_zero_disables() {
+        use crate::common::DeletionProtectionSpec;
+
+        let spec: RepositorySpec = from_yaml(
+            "backend: { filesystem: { path: /repo } }\n\
+             encryption: { passwordSecretRef: { name: s } }\n\
+             deletionProtection:\n  threshold: 25\n",
+        );
+        assert_eq!(
+            spec.deletion_protection.as_ref().and_then(|d| d.threshold),
+            Some(25)
+        );
+        assert_eq!(
+            crate::consts::effective_mass_deletion_threshold(spec.deletion_protection.as_ref()),
+            25
+        );
+        let json = serde_json::to_value(&spec).expect("serialize");
+        assert_eq!(json["deletionProtection"]["threshold"], 25);
+        let reparsed: RepositorySpec = serde_json::from_value(json).expect("reparse");
+        assert_eq!(spec, reparsed);
+
+        // `Some(0)` is the disable sentinel — it passes through, not "fall back to default".
+        let disabled = DeletionProtectionSpec { threshold: Some(0) };
+        assert_eq!(
+            crate::consts::effective_mass_deletion_threshold(Some(&disabled)),
+            0
+        );
+
+        // Absent deletionProtection stays None and is elided (no stored-object churn).
+        let bare: RepositorySpec = from_yaml(
+            "backend: { filesystem: { path: /repo } }\n\
+             encryption: { passwordSecretRef: { name: s } }\n",
+        );
+        assert!(bare.deletion_protection.is_none());
+        assert!(
+            serde_json::to_value(&bare)
+                .unwrap()
+                .get("deletionProtection")
+                .is_none(),
+            "absent deletionProtection must be elided"
+        );
+    }
+
     #[test]
     fn storage_stats_index_blob_count_roundtrips() {
         let stats = StorageStats {