kopiur_api/
restore.rs

1//! The `Restore` CRD — a restore from a snapshot/identity to a PVC, or a passive
2//! populator source. ADR-0001 §3.6, ADR-0003 §4.6.
3
4use crate::common::{
5    CredentialProjection, FailurePolicy, MoverSpec, ObjectRef, PvcAccessMode, RepositoryRef,
6    ResolvedIdentity,
7};
8use k8s_openapi::apimachinery::pkg::apis::meta::v1::Condition;
9use kube::CustomResource;
10use schemars::JsonSchema;
11use serde::{Deserialize, Serialize};
12
13/// A restore operation from a snapshot/identity into a PVC, or a passive populator source.
14#[derive(CustomResource, Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
15#[kube(
16    group = "kopiur.home-operations.com",
17    version = "v1alpha1",
18    kind = "Restore",
19    namespaced,
20    status = "RestoreStatus",
21    shortname = "kopiarestore",
22    category = "kopiur",
23    printcolumn = r#"{"name":"Phase","type":"string","jsonPath":".status.phase"}"#,
24    printcolumn = r#"{"name":"Source","type":"string","jsonPath":".status.sourceKind"}"#,
25    printcolumn = r#"{"name":"Age","type":"date","jsonPath":".metadata.creationTimestamp"}"#
26)]
27// §15: operator-authored CEL in the CRD schema — exactly one of
28// target.pvc/target.pvcRef/target.populator. Validates in the apiserver + CI
29// (`kubeconform`), complementing the webhook. `target` is required so `has(self.target)`
30// is always true; the rule counts the present sub-keys.
31#[schemars(extend("x-kubernetes-validations" = [{
32    "rule": "[has(self.target.pvc), has(self.target.pvcRef), has(self.target.populator)].filter(x, x).size() == 1",
33    "message": "exactly one of target.pvc, target.pvcRef, target.populator"
34}]))]
35#[serde(rename_all = "camelCase")]
36/// Desired state of a Restore: where to read from, where to write to, and how to behave when the snapshot is missing.
37pub struct RestoreSpec {
38    /// The repository to read from; derived from `source` when omitted, required only with `source.identity`.
39    #[serde(default, skip_serializing_if = "Option::is_none")]
40    pub repository: Option<RepositoryRef>,
41    /// Where to read data from (snapshotRef, fromPolicy, or identity).
42    pub source: RestoreSource,
43    /// Where to write the restored data (pvc, pvcRef, or populator).
44    pub target: RestoreTarget,
45    /// kopia restore behavior (file deletion, permission/atomicity handling).
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub options: Option<RestoreOptions>,
48    /// What to do when the referenced snapshot doesn't exist yet, and how long to wait.
49    #[serde(default, skip_serializing_if = "Option::is_none")]
50    pub policy: Option<RestorePolicy>,
51    /// Opt-in copying of the repository's credential Secret(s) into the mover's namespace (default off).
52    #[serde(default, skip_serializing_if = "Option::is_none")]
53    pub credential_projection: Option<CredentialProjection>,
54    /// Per-run mover overrides for this restore's Job (resources, cache, `securityContext`).
55    #[serde(default, skip_serializing_if = "Option::is_none")]
56    pub mover: Option<MoverSpec>,
57    /// Mover `Job` retry/deadline limits (`backoffLimit`, `activeDeadlineSeconds`).
58    #[serde(default, skip_serializing_if = "Option::is_none")]
59    pub failure_policy: Option<FailurePolicy>,
60}
61
62/// Where to restore from; exactly one variant.
63#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
64#[serde(rename_all = "camelCase")]
65pub enum RestoreSource {
66    /// A `Snapshot` CR (scheduled, manual, or discovered).
67    SnapshotRef(ObjectRef),
68    /// A `SnapshotPolicy` CR, resolved via identity even with no `Snapshot` CR present (deploy-or-restore).
69    FromPolicy(FromPolicy),
70    /// A raw kopia identity (foreign writers / aged-out catalog); requires `spec.repository`.
71    Identity(IdentitySource),
72}
73
74impl RestoreSource {
75    /// Stable discriminant string for status/metrics.
76    ///
77    /// ```
78    /// use kopiur_api::common::ObjectRef;
79    /// use kopiur_api::restore::RestoreSource;
80    ///
81    /// let src = RestoreSource::SnapshotRef(ObjectRef { name: "pg-20260524".into(), namespace: None });
82    /// assert_eq!(src.kind_str(), "SnapshotRef");
83    ///
84    /// // Externally tagged: each variant deserializes under its own camelCase key.
85    /// let from_cfg: RestoreSource =
86    ///     serde_json::from_value(serde_json::json!({ "fromPolicy": { "name": "pg" } })).unwrap();
87    /// assert_eq!(from_cfg.kind_str(), "FromPolicy");
88    /// ```
89    pub fn kind_str(&self) -> &'static str {
90        match self {
91            RestoreSource::SnapshotRef(_) => "SnapshotRef",
92            RestoreSource::FromPolicy(_) => "FromPolicy",
93            RestoreSource::Identity(_) => "Identity",
94        }
95    }
96}
97
98/// The `fromPolicy` source: resolve a snapshot via a `SnapshotPolicy`'s identity, even when no `Snapshot` CR exists yet.
99#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
100#[serde(rename_all = "camelCase")]
101pub struct FromPolicy {
102    /// Name of the `SnapshotPolicy` whose identity selects the snapshot.
103    pub name: String,
104    /// Namespace of the `SnapshotPolicy`; absent = the `Restore`'s own namespace.
105    #[serde(default, skip_serializing_if = "Option::is_none")]
106    pub namespace: Option<String>,
107    /// Restore the newest snapshot at or before this RFC3339 timestamp (point-in-time).
108    #[serde(default, skip_serializing_if = "Option::is_none")]
109    pub as_of: Option<String>,
110    /// Which snapshot to pick: 0 = latest, 1 = previous, and so on.
111    #[serde(default = "default_offset")]
112    #[schemars(default = "default_offset")]
113    pub offset: i64,
114    /// The kopia source path to restore FROM, overriding the path kopiur derives
115    /// from the policy.
116    ///
117    /// Normally the path is derived: a policy with one plain `pvc:`/`nfs` source
118    /// contributes its own path, and a `pvcSelector` policy contributes the path
119    /// its `sourcePathStrategy` would have produced for the PVC being restored
120    /// (`/pvc/<name>` or `/pvc/<namespace>/<name>`) — the same rule the backup
121    /// side used when it wrote the snapshot, so a fan-out restore fills each PVC
122    /// from ITS OWN snapshot instead of the newest snapshot of any member.
123    ///
124    /// Set this when the derivation is ambiguous or wrong: selector sources that
125    /// disagree on `sourcePathStrategy`/`sourcePathOverride` (kopiur fails closed
126    /// rather than guess), a policy whose selector sources share one
127    /// `sourcePathOverride`, or a cross-namespace `target.pvcRef` whose derived
128    /// `/pvc/<namespace>/<name>` names a namespace the repository never saw.
129    ///
130    /// Mirrors `IdentitySource::source_path`: it selects which kopia source to
131    /// READ, it does not change where the data is written.
132    #[serde(default, skip_serializing_if = "Option::is_none")]
133    #[schemars(length(max = 4096))]
134    pub source_path: Option<String>,
135}
136
137/// serde/schemars `default` for [`FromPolicy::offset`] — `0`, the latest snapshot
138/// (ADR-0005 §1). A named fn so it backs BOTH `#[serde(default = ...)]` and
139/// `#[schemars(default = ...)]`, which is what makes schemars 1 emit the OpenAPI
140/// `default:` in the generated CRD schema.
141fn default_offset() -> i64 {
142    0
143}
144
145/// The `identity` source: a raw kopia `username@hostname:path` identity; requires `spec.repository`.
146#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
147#[serde(rename_all = "camelCase")]
148pub struct IdentitySource {
149    /// The kopia `username` to match.
150    pub username: String,
151    /// The kopia `hostname` to match.
152    pub hostname: String,
153    /// The kopia source path to match; absent matches any path for the identity.
154    #[serde(default, skip_serializing_if = "Option::is_none")]
155    pub source_path: Option<String>,
156    /// Pin an exact kopia snapshot by ID.
157    #[serde(
158        default,
159        rename = "snapshotID",
160        skip_serializing_if = "Option::is_none"
161    )]
162    pub snapshot_id: Option<String>,
163    /// Restore the newest snapshot at or before this RFC3339 timestamp (point-in-time).
164    #[serde(default, skip_serializing_if = "Option::is_none")]
165    pub as_of: Option<String>,
166    /// Which snapshot to pick: 0 = latest, 1 = previous, and so on.
167    #[serde(default, skip_serializing_if = "Option::is_none")]
168    pub offset: Option<i64>,
169}
170
171/// Where to restore to; exactly one variant.
172#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
173#[serde(rename_all = "camelCase")]
174pub enum RestoreTarget {
175    /// Operator creates the PVC.
176    Pvc(PvcTemplate),
177    /// Write into an existing PVC.
178    PvcRef(ObjectRef),
179    /// Passive populator mode: the restore is claimed by a PVC's `spec.dataSourceRef`.
180    Populator(PopulatorTarget),
181}
182
183impl RestoreTarget {
184    /// Stable discriminant string for status/metrics.
185    ///
186    /// ```
187    /// use kopiur_api::common::ObjectRef;
188    /// use kopiur_api::restore::RestoreTarget;
189    ///
190    /// let into_existing = RestoreTarget::PvcRef(ObjectRef { name: "data".into(), namespace: None });
191    /// assert_eq!(into_existing.kind_str(), "PvcRef");
192    ///
193    /// // Externally tagged: `{ pvc: {...} }` selects the create-PVC variant.
194    /// let created: RestoreTarget =
195    ///     serde_json::from_value(serde_json::json!({ "pvc": { "name": "restored" } })).unwrap();
196    /// assert_eq!(created.kind_str(), "Pvc");
197    /// ```
198    pub fn kind_str(&self) -> &'static str {
199        match self {
200            RestoreTarget::Pvc(_) => "Pvc",
201            RestoreTarget::PvcRef(_) => "PvcRef",
202            RestoreTarget::Populator(_) => "Populator",
203        }
204    }
205}
206
207/// Passive-populator target marker; its presence selects populator mode.
208#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
209#[serde(rename_all = "camelCase")]
210pub struct PopulatorTarget {}
211
212/// Template for a PVC the operator creates as the restore target.
213#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
214#[serde(rename_all = "camelCase")]
215pub struct PvcTemplate {
216    /// Name of the PVC to create.
217    pub name: String,
218    /// StorageClass for the new PVC; absent uses the cluster default.
219    #[serde(default, skip_serializing_if = "Option::is_none")]
220    pub storage_class_name: Option<String>,
221    /// Requested size of the new PVC (e.g. `100Gi`).
222    #[serde(default, skip_serializing_if = "Option::is_none")]
223    pub capacity: Option<String>,
224    /// Access modes for the new PVC; empty defaults to `[ReadWriteOnce]`. Closed
225    /// enum in the schema; a non-canonical value persisted before enforcement
226    /// decodes as [`PvcAccessMode::Unknown`] and is rejected per-CR with the
227    /// value quoted (webhook + reconciler, via `validate_access_modes`) instead
228    /// of poisoning the typed watcher.
229    #[serde(default, skip_serializing_if = "Vec::is_empty")]
230    pub access_modes: Vec<PvcAccessMode>,
231}
232
233/// kopia restore behavior knobs (M2 flag sweep). Every `Option` field's `None`
234/// reproduces kopia's own default — an all-`None`, `enableFileDeletion: false`
235/// instance yields the exact same `restore_args` argv produced before these
236/// fields existed. The tri-state booleans map to kopia's `--[no-]flag` grammar
237/// (`Some(true)` → `--flag`, `Some(false)` → `--no-flag`, `None` → omit).
238#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
239#[serde(rename_all = "camelCase")]
240pub struct RestoreOptions {
241    /// Delete files in the target that are not present in the snapshot (exact mirror); off by default.
242    /// Wired to kopia's `--[no-]delete-extra` (previously a silent no-op — see issue #216 gap sweep).
243    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
244    pub enable_file_deletion: bool,
245    /// Continue past permission errors during restore (default true).
246    #[serde(default, skip_serializing_if = "Option::is_none")]
247    pub ignore_permission_errors: Option<bool>,
248    /// Write files atomically via a temp file + rename (default true).
249    #[serde(default, skip_serializing_if = "Option::is_none")]
250    pub write_files_atomically: Option<bool>,
251    /// `--parallel`: restore parallelism (kopia default `8`; `1` disables parallelism).
252    #[serde(default, skip_serializing_if = "Option::is_none")]
253    pub parallel: Option<u32>,
254    /// `--[no-]write-sparse-files`: attempt to write files sparsely, allocating the
255    /// minimum disk space needed (kopia default `false`).
256    #[serde(default, skip_serializing_if = "Option::is_none")]
257    pub write_sparse_files: Option<bool>,
258    /// `--[no-]skip-owners`: skip restoring file owners (kopia default `false`).
259    #[serde(default, skip_serializing_if = "Option::is_none")]
260    pub skip_owners: Option<bool>,
261    /// `--[no-]skip-permissions`: skip restoring file permissions (kopia default `false`).
262    #[serde(default, skip_serializing_if = "Option::is_none")]
263    pub skip_permissions: Option<bool>,
264    /// `--[no-]skip-times`: skip restoring file modification times (kopia default `false`).
265    #[serde(default, skip_serializing_if = "Option::is_none")]
266    pub skip_times: Option<bool>,
267    /// `--[no-]overwrite-files`: overwrite existing files in the target (kopia default `true`).
268    #[serde(default, skip_serializing_if = "Option::is_none")]
269    pub overwrite_files: Option<bool>,
270    /// `--[no-]overwrite-directories`: overwrite existing directories in the target
271    /// (kopia default `true`).
272    #[serde(default, skip_serializing_if = "Option::is_none")]
273    pub overwrite_directories: Option<bool>,
274    /// `--[no-]overwrite-symlinks`: overwrite existing symlinks in the target
275    /// (kopia default `true`).
276    #[serde(default, skip_serializing_if = "Option::is_none")]
277    pub overwrite_symlinks: Option<bool>,
278    /// `--[no-]ignore-errors`: ignore all restore errors and continue (kopia default `false`).
279    #[serde(default, skip_serializing_if = "Option::is_none")]
280    pub ignore_errors: Option<bool>,
281    /// `--[no-]skip-existing`: skip files/symlinks that already exist in the target
282    /// (kopia default `false`).
283    #[serde(default, skip_serializing_if = "Option::is_none")]
284    pub skip_existing: Option<bool>,
285}
286
287/// How the restore reacts to a missing snapshot and how long it waits.
288#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
289#[serde(rename_all = "camelCase")]
290pub struct RestorePolicy {
291    /// What to do when the resolved source matches no snapshot (`Fail`/`Continue`).
292    #[serde(default, skip_serializing_if = "Option::is_none")]
293    pub on_missing_snapshot: Option<OnMissingSnapshot>,
294    /// How long to wait for the source snapshot to appear before giving up (e.g. `5m`).
295    #[serde(default, skip_serializing_if = "Option::is_none")]
296    pub wait_timeout: Option<String>,
297}
298
299/// What to do when the resolved source matches no snapshot. Defaults to `Fail`
300/// (fail-closed) so an explicit restore can never silently no-op; choose
301/// `Continue` to provision an empty volume instead (deploy-or-restore).
302#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, Default, JsonSchema)]
303pub enum OnMissingSnapshot {
304    /// Fail-closed; the default for explicit `snapshotRef`/`identity` sources.
305    #[default]
306    Fail,
307    /// Proceed with an empty volume (deploy-or-restore); the default for `fromPolicy`.
308    Continue,
309}
310
311/// Lifecycle phase of a restore.
312///
313/// ```
314/// use kopiur_api::RestorePhase;
315///
316/// assert_eq!(serde_json::to_value(RestorePhase::Restoring).unwrap(), "Restoring");
317/// // An unrecognized phase from a newer operator decodes instead of erroring.
318/// let p: RestorePhase = serde_json::from_value(serde_json::json!("Staging")).unwrap();
319/// assert_eq!(p, RestorePhase::Unknown("Staging".into()));
320/// assert_eq!(serde_json::to_value(&p).unwrap(), "Staging");
321/// assert!(!p.is_terminal());
322/// ```
323#[derive(Clone, Debug, PartialEq, Eq, Default)]
324pub enum RestorePhase {
325    /// Admitted but not yet acted on; the default initial phase.
326    #[default]
327    Pending,
328    /// Resolving the source to a concrete snapshot and pinning it to status.
329    Resolving,
330    /// The mover `Job` is actively writing data into the target.
331    Restoring,
332    /// The restore finished successfully.
333    Completed,
334    /// The restore terminally failed; see `conditions` for the reason.
335    Failed,
336    /// A phase string this build does not recognize (newer operator, or legacy
337    /// stored data). Decode-compat only — hidden from the CRD schema, never
338    /// produced by this build, never terminal.
339    Unknown(String),
340}
341
342crate::common::phase_serde!(RestorePhase, "Lifecycle phase of a restore.");
343
344impl RestorePhase {
345    /// Whether this phase is **terminal**: the restore reached an end state and
346    /// the operator will do no further work on it of its own accord.
347    ///
348    /// `Pending`, `Resolving`, and `Restoring` are all in-flight — including a
349    /// `Pending` restore parked on a structural gate (see
350    /// [`crate::gates::STRUCTURAL_GATES`]), which never progresses but is
351    /// emphatically not finished. A `Restore` has no deletion phase of its own,
352    /// so unlike `SnapshotPhase` there is no wedged-finalizer case to exclude.
353    ///
354    /// Pure + exhaustive so the single definition lives in one tested place.
355    ///
356    /// ```
357    /// use kopiur_api::RestorePhase;
358    ///
359    /// assert!(RestorePhase::Completed.is_terminal());
360    /// assert!(RestorePhase::Failed.is_terminal());
361    /// assert!(!RestorePhase::Pending.is_terminal());
362    /// assert!(!RestorePhase::Resolving.is_terminal());
363    /// assert!(!RestorePhase::Restoring.is_terminal());
364    /// assert!(!RestorePhase::Unknown("Staging".into()).is_terminal());
365    /// ```
366    pub fn is_terminal(&self) -> bool {
367        match self {
368            Self::Completed | Self::Failed => true,
369            Self::Pending | Self::Resolving | Self::Restoring => false,
370            // Conservative surface-it policy: a phase this build cannot
371            // interpret is never reported as finished, so a newer operator's
372            // in-flight restore stays visible to an older CLI/reconciler.
373            Self::Unknown(_) => false,
374        }
375    }
376
377    /// Whether this phase is the **decode sentinel** — a value the running build
378    /// cannot interpret, kept verbatim by [`Unknown`](Self::Unknown) instead of
379    /// failing the whole typed `list()`/watch (#359, defect 3).
380    ///
381    /// Same narrow contract as [`SnapshotPhase::is_unknown`](crate::SnapshotPhase::is_unknown):
382    /// `true` means only "this string is not a phase this binary knows". A
383    /// canonical variant added later is not the sentinel, which is why the
384    /// `match` is written out exhaustively rather than left as a `matches!`.
385    ///
386    /// ```
387    /// use kopiur_api::RestorePhase;
388    ///
389    /// assert!(RestorePhase::Unknown("Staging".into()).is_unknown());
390    /// assert!(!RestorePhase::Completed.is_unknown());
391    /// assert!(!RestorePhase::Failed.is_unknown());
392    /// ```
393    pub fn is_unknown(&self) -> bool {
394        match self {
395            Self::Unknown(_) => true,
396            Self::Pending | Self::Resolving | Self::Restoring | Self::Completed | Self::Failed => {
397                false
398            }
399        }
400    }
401}
402
403impl crate::common::PhaseLabel for RestorePhase {
404    const ALL: &'static [Self] = &[
405        Self::Pending,
406        Self::Resolving,
407        Self::Restoring,
408        Self::Completed,
409        Self::Failed,
410    ];
411    fn label(&self) -> &str {
412        match self {
413            Self::Pending => "Pending",
414            Self::Resolving => "Resolving",
415            Self::Restoring => "Restoring",
416            Self::Completed => "Completed",
417            Self::Failed => "Failed",
418            Self::Unknown(s) => s,
419        }
420    }
421    fn unknown(raw: String) -> Self {
422        Self::Unknown(raw)
423    }
424}
425
426/// Observed state of a Restore, written by the controller/mover.
427#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Default, JsonSchema)]
428#[serde(rename_all = "camelCase")]
429pub struct RestoreStatus {
430    /// Current lifecycle phase.
431    #[serde(default, skip_serializing_if = "Option::is_none")]
432    pub phase: Option<RestorePhase>,
433    /// The pinned source kind (`SnapshotRef`/`FromPolicy`/`Identity`); backs the `SOURCE` printer column.
434    #[serde(default, skip_serializing_if = "Option::is_none")]
435    pub source_kind: Option<String>,
436    /// `metadata.generation` last reconciled, so stale status is detectable.
437    #[serde(default, skip_serializing_if = "Option::is_none")]
438    pub observed_generation: Option<i64>,
439    /// The source resolved and pinned at admission; never re-resolved.
440    #[serde(default, skip_serializing_if = "Option::is_none")]
441    pub resolved: Option<ResolvedRestore>,
442    /// Resolved target details (the PVC written to / populator handshake).
443    #[serde(default, skip_serializing_if = "Option::is_none")]
444    pub target: Option<RestoreTargetStatus>,
445    /// Start/end timestamps for the restore run.
446    #[serde(default, skip_serializing_if = "Option::is_none")]
447    pub timing: Option<RestoreTiming>,
448    /// When the `policy.waitTimeout` window OPENED (RFC3339) — the first reconcile on which
449    /// the restore could actually proceed (its repository reached `Ready`, and for a
450    /// `target.populator` a PVC already claims it), NOT when the Restore was created. The
451    /// window does NOT open while the referenced `Repository` object or `fromPolicy`
452    /// `SnapshotPolicy` doesn't exist: the restore parks in `Pending`
453    /// (`ReferentAvailable=False`, reason `RestoreReferentMissing`) and stays unstamped. It
454    /// DOES still open for a `snapshotRef` whose `Snapshot` row doesn't exist yet (so
455    /// `onMissingSnapshot` can fire for a ref that never appears) and for a restore whose
456    /// mover Job already launched. Stamped once and then honored verbatim, so the window
457    /// survives controller restarts and Job pod retries; cleared when a populator re-opens
458    /// resolution for a re-created claim, so that claim gets the full window again. Absent
459    /// means the window has not opened yet (or no `policy.waitTimeout` is configured, in
460    /// which case there is no window to anchor).
461    #[serde(default, skip_serializing_if = "Option::is_none")]
462    pub wait_started_at: Option<String>,
463    /// Bytes/files restored so far, patched periodically by the mover.
464    #[serde(default, skip_serializing_if = "Option::is_none")]
465    pub progress: Option<RestoreProgress>,
466    /// Standard Kubernetes conditions carrying the human-readable status/reason.
467    #[serde(default, skip_serializing_if = "Vec::is_empty")]
468    pub conditions: Vec<Condition>,
469    /// The last lines of the run's output, written by the mover at the terminal transition.
470    #[serde(default, skip_serializing_if = "Option::is_none")]
471    pub log_tail: Option<String>,
472    /// Structured terminal-failure detail (kopia error class, stderr tail, retry hint).
473    #[serde(default, skip_serializing_if = "Option::is_none")]
474    pub failure: Option<crate::common::FailureBlock>,
475    /// Per-claimant state for a `target.populator` restore, keyed by the claiming
476    /// PVC's name (#443).
477    ///
478    /// A populator `Restore` is claimed by EVERY PVC whose `spec.dataSourceRef`
479    /// names it, not just the first, and each claimant gets its own prime PVC, its
480    /// own mover `Job` and its own kopia source path — so each needs its own state.
481    /// A **map**, not a list, because an RFC-7386 merge patch merges map keys but
482    /// REPLACES arrays: N concurrent populate movers each patch only their own key
483    /// and can never clobber a sibling (the same reason as
484    /// `SnapshotPolicyStatus.verificationStamps`).
485    ///
486    /// Absent for a direct `target.pvc`/`target.pvcRef` restore, which keeps using
487    /// the top-level `resolved`/`target`/`waitStartedAt`/`logTail`/`failure`.
488    ///
489    /// The schema renders as an object with `additionalProperties`, which PRUNES
490    /// unknown keys — so every field any writer (controller or mover) puts under
491    /// `claims.<pvc>` must exist on `RestoreClaimStatus`.
492    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
493    pub claims: std::collections::BTreeMap<String, RestoreClaimStatus>,
494}
495
496/// The state of ONE claiming PVC of a `target.populator` restore (#443).
497///
498/// Written by the controller (every field except `resolved`/`observedAt`/
499/// `logTail`/`failure`) and by that claim's mover `Job`, which nests its status
500/// patch under `status.claims.<pvc>` and deliberately omits `phase` — the
501/// controller owns claim phase exactly as it owns the top-level one.
502#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
503#[serde(rename_all = "camelCase")]
504pub struct RestoreClaimStatus {
505    /// `metadata.uid` of the claiming PVC this record describes. A record whose
506    /// uid no longer matches the live claimant is STALE — the PVC was deleted and
507    /// re-created — so the claim re-arms and the old claim's prime PVC / Job / PV
508    /// are reaped under the recorded uid.
509    #[serde(default, skip_serializing_if = "Option::is_none")]
510    pub uid: Option<String>,
511    /// Lifecycle phase of this claim; absent means "not observed yet".
512    #[serde(default, skip_serializing_if = "Option::is_none")]
513    pub phase: Option<RestoreClaimPhase>,
514    /// Machine-readable reason for the current phase (`PopulatingPrimePvc`,
515    /// `MoverJobFailed`, `SourcePathAmbiguous`, …).
516    #[serde(default, skip_serializing_if = "Option::is_none")]
517    pub reason: Option<String>,
518    /// Human-readable what / why / fix for the current phase.
519    #[serde(default, skip_serializing_if = "Option::is_none")]
520    pub message: Option<String>,
521    /// The kopia source path this claim restores FROM, derived from the policy's
522    /// `sourcePathStrategy` (or pinned from `source.fromPolicy.sourcePath`).
523    /// Recorded so a fan-out restore is auditable: each claim shows which member's
524    /// data it read.
525    #[serde(default, skip_serializing_if = "Option::is_none")]
526    pub source_path: Option<String>,
527    /// The source this claim resolved and pinned; never re-resolved.
528    #[serde(default, skip_serializing_if = "Option::is_none")]
529    pub resolved: Option<ResolvedRestore>,
530    /// The prime PVC provisioned for this claim's populate handshake.
531    #[serde(default, skip_serializing_if = "Option::is_none")]
532    pub pvc_prime: Option<String>,
533    /// The mover `Job` populating this claim's prime PVC.
534    #[serde(default, skip_serializing_if = "Option::is_none")]
535    pub job: Option<String>,
536    /// When THIS claim's `policy.waitTimeout` window opened (RFC3339) — the first
537    /// pass on which this claim could actually proceed. Per-claim because
538    /// claimants appear at different times: a sibling created an hour later gets
539    /// its own full window.
540    #[serde(default, skip_serializing_if = "Option::is_none")]
541    pub wait_started_at: Option<String>,
542    /// When this claim's mover last patched (RFC3339). Mover-owned: every
543    /// `StatusUpdate` carries it, and the field must exist here or the schema's
544    /// `additionalProperties` pruning would silently drop the mover's whole patch.
545    #[serde(default, skip_serializing_if = "Option::is_none")]
546    pub observed_at: Option<String>,
547    /// The last lines of this claim's mover output, written once at its terminal
548    /// transition. Mover-owned.
549    #[serde(default, skip_serializing_if = "Option::is_none")]
550    pub log_tail: Option<String>,
551    /// Structured terminal-failure detail for this claim. Mover-owned.
552    #[serde(default, skip_serializing_if = "Option::is_none")]
553    pub failure: Option<crate::common::FailureBlock>,
554}
555
556/// Lifecycle phase of ONE claiming PVC of a populator `Restore` (#443).
557///
558/// Deliberately a separate closed enum from [`RestorePhase`]: a claim has states
559/// the Restore does not (`Rebinding`, `AlreadyBound`), and the Restore-level
560/// phase is the AGGREGATE over these. Named `*Phase` on purpose, so
561/// `cargo xtask check-phases` polices every branch on it.
562///
563/// ```
564/// use kopiur_api::RestoreClaimPhase;
565///
566/// assert_eq!(serde_json::to_value(RestoreClaimPhase::Populating).unwrap(), "Populating");
567/// // An unrecognized phase from a newer operator decodes instead of erroring.
568/// let p: RestoreClaimPhase = serde_json::from_value(serde_json::json!("Staging")).unwrap();
569/// assert_eq!(p, RestoreClaimPhase::Unknown("Staging".into()));
570/// assert_eq!(serde_json::to_value(&p).unwrap(), "Staging");
571/// assert!(!p.is_terminal());
572/// ```
573#[derive(Clone, Debug, PartialEq, Eq, Default)]
574pub enum RestoreClaimPhase {
575    /// Observed, but nothing has run for it yet — waiting on a scheduling hint
576    /// (`WaitForFirstConsumer`), on the repository, or on the source snapshot.
577    #[default]
578    Pending,
579    /// A mover `Job` is writing this claim's prime PVC.
580    Populating,
581    /// The prime volume is written and its `PersistentVolume` was rebound to the
582    /// claim; waiting for the PV controller to complete the bind.
583    Rebinding,
584    /// The claim is bound to the volume this restore populated. Terminal.
585    Populated,
586    /// The claim was ALREADY bound when first observed, so there was nothing to
587    /// populate — a CSI volume-populator only fills an UNBOUND claim. A truthful
588    /// terminal no-op (#233): no prime PVC, no mover run.
589    AlreadyBound,
590    /// This claim terminally failed; `reason`/`message` say why. Terminal for the
591    /// CLAIM only — siblings keep going, and re-creating this claiming PVC
592    /// re-arms it.
593    Failed,
594    /// A phase string this build does not recognize (newer operator, or legacy
595    /// stored data). Decode-compat only — hidden from the CRD schema, never
596    /// produced by this build, never terminal.
597    Unknown(String),
598}
599
600crate::common::phase_serde!(
601    RestoreClaimPhase,
602    "Lifecycle phase of one claiming PVC of a populator restore."
603);
604
605impl RestoreClaimPhase {
606    /// Whether this claim reached an end state and the operator will do no
607    /// further work on it of its own accord.
608    ///
609    /// `Failed` IS terminal here, and that is what lets the fan-out stop
610    /// short-circuiting the WHOLE `Restore` on `Failed`: the failure is scoped to
611    /// this claim, its siblings keep going, and what re-arms it is deleting and
612    /// re-creating the claiming PVC (which mints a new uid, hence a new record) —
613    /// never a re-drive of the old one.
614    ///
615    /// Pure + exhaustive, so the single definition lives in one tested place.
616    ///
617    /// ```
618    /// use kopiur_api::RestoreClaimPhase;
619    ///
620    /// assert!(RestoreClaimPhase::Populated.is_terminal());
621    /// assert!(RestoreClaimPhase::AlreadyBound.is_terminal());
622    /// assert!(RestoreClaimPhase::Failed.is_terminal());
623    /// assert!(!RestoreClaimPhase::Pending.is_terminal());
624    /// assert!(!RestoreClaimPhase::Populating.is_terminal());
625    /// assert!(!RestoreClaimPhase::Rebinding.is_terminal());
626    /// assert!(!RestoreClaimPhase::Unknown("Staging".into()).is_terminal());
627    /// ```
628    pub fn is_terminal(&self) -> bool {
629        match self {
630            Self::Populated | Self::AlreadyBound | Self::Failed => true,
631            Self::Pending | Self::Populating | Self::Rebinding => false,
632            // Conservative surface-it policy, same as `RestorePhase::is_terminal`:
633            // a phase this build cannot interpret is never reported as finished.
634            Self::Unknown(_) => false,
635        }
636    }
637
638    /// Whether this phase is the **decode sentinel** — a value the running build
639    /// cannot interpret, kept verbatim by [`Unknown`](Self::Unknown) instead of
640    /// failing the whole typed `list()`/watch for the Kind.
641    ///
642    /// ```
643    /// use kopiur_api::RestoreClaimPhase;
644    ///
645    /// assert!(RestoreClaimPhase::Unknown("Staging".into()).is_unknown());
646    /// assert!(!RestoreClaimPhase::Populated.is_unknown());
647    /// ```
648    pub fn is_unknown(&self) -> bool {
649        match self {
650            Self::Unknown(_) => true,
651            Self::Pending
652            | Self::Populating
653            | Self::Rebinding
654            | Self::Populated
655            | Self::AlreadyBound
656            | Self::Failed => false,
657        }
658    }
659}
660
661impl crate::common::PhaseLabel for RestoreClaimPhase {
662    const ALL: &'static [Self] = &[
663        Self::Pending,
664        Self::Populating,
665        Self::Rebinding,
666        Self::Populated,
667        Self::AlreadyBound,
668        Self::Failed,
669    ];
670    fn label(&self) -> &str {
671        match self {
672            Self::Pending => "Pending",
673            Self::Populating => "Populating",
674            Self::Rebinding => "Rebinding",
675            Self::Populated => "Populated",
676            Self::AlreadyBound => "AlreadyBound",
677            Self::Failed => "Failed",
678            Self::Unknown(s) => s,
679        }
680    }
681    fn unknown(raw: String) -> Self {
682        Self::Unknown(raw)
683    }
684}
685
686/// Which outcome the source resolution pinned, once and never re-resolved.
687#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, JsonSchema)]
688pub enum ResolutionOutcome {
689    /// The source resolved to a concrete kopia snapshot (see `kopiaSnapshotID`).
690    Snapshot,
691    /// The source matched no snapshot; `Continue` chose an empty (deploy-or-restore) volume.
692    NoSnapshot,
693}
694
695/// The source resolved and pinned at admission, so a restore never silently retargets.
696#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
697#[serde(rename_all = "camelCase")]
698pub struct ResolvedRestore {
699    /// Which outcome the resolution pinned (`Snapshot`/`NoSnapshot`).
700    #[serde(default, skip_serializing_if = "Option::is_none")]
701    pub resolution: Option<ResolutionOutcome>,
702    /// The exact kopia snapshot manifest id the source resolved to; pinned once.
703    #[serde(
704        default,
705        rename = "kopiaSnapshotID",
706        skip_serializing_if = "Option::is_none"
707    )]
708    pub kopia_snapshot_id: Option<String>,
709    /// The concrete `Snapshot` CR the source resolved to, when applicable.
710    #[serde(default, skip_serializing_if = "Option::is_none")]
711    pub snapshot_ref: Option<ObjectRef>,
712    /// The repository the snapshot lives in, resolved from the source.
713    #[serde(default, skip_serializing_if = "Option::is_none")]
714    pub repository: Option<RepositoryRef>,
715    /// Timestamp at which the source was pinned (RFC3339).
716    #[serde(default, skip_serializing_if = "Option::is_none")]
717    pub pinned_at: Option<String>,
718    /// The resolved kopia identity (`username@hostname:path`) of the snapshot.
719    #[serde(default, skip_serializing_if = "Option::is_none")]
720    pub identity: Option<ResolvedIdentity>,
721}
722
723/// Resolved restore target details written to status.
724#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
725#[serde(rename_all = "camelCase")]
726pub struct RestoreTargetStatus {
727    /// Populator handshake (passive / pvc-create modes).
728    #[serde(default, skip_serializing_if = "Option::is_none")]
729    pub pvc_prime: Option<String>,
730    /// The PVC actually written to (created or pre-existing).
731    #[serde(default, skip_serializing_if = "Option::is_none")]
732    pub pvc_ref: Option<ObjectRef>,
733}
734
735/// Start/end timestamps of a restore run.
736#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
737#[serde(rename_all = "camelCase")]
738pub struct RestoreTiming {
739    /// When the mover began restoring (RFC3339).
740    #[serde(default, skip_serializing_if = "Option::is_none")]
741    pub start_time: Option<String>,
742    /// When the restore reached a terminal phase (RFC3339).
743    #[serde(default, skip_serializing_if = "Option::is_none")]
744    pub end_time: Option<String>,
745}
746
747/// Live progress counters patched by the mover during a restore.
748#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
749#[serde(rename_all = "camelCase")]
750pub struct RestoreProgress {
751    /// Total bytes restored so far.
752    #[serde(default, skip_serializing_if = "Option::is_none")]
753    pub bytes_restored: Option<i64>,
754    /// Total files restored so far.
755    #[serde(default, skip_serializing_if = "Option::is_none")]
756    pub files_restored: Option<i64>,
757}
758
759#[cfg(test)]
760mod tests {
761    use super::*;
762    use crate::common::PhaseLabel;
763    use crate::testutil::from_yaml;
764    use kube::core::CustomResourceExt;
765
766    #[test]
767    fn restore_phase_all_covers_every_variant_uniquely() {
768        // Mirrors the `SnapshotPhase` tripwire: every variant is in ALL with a
769        // unique, non-empty label. A variant added without updating ALL fails
770        // here (and `label`'s exhaustive match won't compile at all).
771        let labels: Vec<&str> = RestorePhase::ALL.iter().map(|p| p.label()).collect();
772        assert_eq!(RestorePhase::ALL.len(), 5);
773        assert!(labels.iter().all(|l| !l.is_empty()));
774        let mut sorted = labels.clone();
775        sorted.sort_unstable();
776        sorted.dedup();
777        assert_eq!(sorted.len(), labels.len(), "phase labels must be unique");
778        assert!(RestorePhase::ALL.contains(&RestorePhase::default()));
779    }
780
781    #[test]
782    fn restore_terminal_set_is_pinned() {
783        // Driven off ALL so a new variant must be classified deliberately.
784        let terminal: Vec<&str> = RestorePhase::ALL
785            .iter()
786            .filter(|p| p.is_terminal())
787            .map(|p| p.label())
788            .collect();
789        assert_eq!(terminal, ["Completed", "Failed"]);
790        let in_flight: Vec<&str> = RestorePhase::ALL
791            .iter()
792            .filter(|p| !p.is_terminal())
793            .map(|p| p.label())
794            .collect();
795        assert_eq!(in_flight, ["Pending", "Resolving", "Restoring"]);
796    }
797
798    #[test]
799    fn restore_crd_metadata_is_correct() {
800        let crd = Restore::crd();
801        assert_eq!(crd.spec.group, "kopiur.home-operations.com");
802        assert_eq!(crd.spec.names.kind, "Restore");
803        assert_eq!(crd.spec.scope, "Namespaced");
804        assert_eq!(crd.spec.versions[0].name, "v1alpha1");
805    }
806
807    #[test]
808    fn restore_crd_carries_target_xor_x_kubernetes_validation() {
809        // §15: the generated CRD spec schema must carry the operator-authored
810        // x-kubernetes-validations rule (exactly-one-of target.*) at the spec level,
811        // surviving kube's structural-schema rewriter.
812        let crd = Restore::crd();
813        let json = serde_json::to_value(&crd).expect("serialize CRD");
814        let spec_schema =
815            &json["spec"]["versions"][0]["schema"]["openAPIV3Schema"]["properties"]["spec"];
816        let rules = spec_schema["x-kubernetes-validations"]
817            .as_array()
818            .expect("spec.x-kubernetes-validations present");
819        assert!(
820            rules.iter().any(|r| r["rule"]
821                .as_str()
822                .is_some_and(|s| s.contains("target.populator"))),
823            "expected the target XOR rule; got {rules:?}"
824        );
825    }
826
827    #[test]
828    fn from_policy_offset_carries_static_openapi_default_in_crd() {
829        // ADR-0005 §1: source.fromPolicy.offset must carry a real schema `default: 0`.
830        let crd = Restore::crd();
831        let json = serde_json::to_value(&crd).expect("serialize CRD");
832        let default = &json["spec"]["versions"][0]["schema"]["openAPIV3Schema"]["properties"]["spec"]
833            ["properties"]["source"]["properties"]["fromPolicy"]["properties"]["offset"]["default"];
834        assert_eq!(
835            default, 0,
836            "fromPolicy.offset must emit `default: 0` in the CRD schema; got {default:?}"
837        );
838    }
839
840    #[test]
841    fn from_policy_offset_defaults_to_zero_when_absent() {
842        let spec: RestoreSpec =
843            from_yaml("source: { fromPolicy: { name: pg } }\ntarget: { populator: {} }\n");
844        match &spec.source {
845            RestoreSource::FromPolicy(c) => assert_eq!(c.offset, 0),
846            other => panic!("expected FromPolicy, got {}", other.kind_str()),
847        }
848    }
849
850    #[test]
851    fn restore_backup_ref_roundtrip_matches_adr_shape() {
852        // Mirrors ADR-0001 §3.6 / §5.3.
853        let yaml = r#"
854source:
855  snapshotRef: { name: postgres-data-20260524-021300, namespace: billing }
856target:
857  pvc:
858    name: postgres-data-restored
859    storageClassName: fast-ssd
860    capacity: 100Gi
861    accessModes: [ReadWriteOnce]
862options:
863  enableFileDeletion: false
864  ignorePermissionErrors: true
865  writeFilesAtomically: true
866policy:
867  onMissingSnapshot: Fail
868  waitTimeout: 5m
869"#;
870        let spec: RestoreSpec = from_yaml(yaml);
871        assert_eq!(spec.source.kind_str(), "SnapshotRef");
872        match &spec.source {
873            RestoreSource::SnapshotRef(r) => {
874                assert_eq!(r.name, "postgres-data-20260524-021300");
875                assert_eq!(r.namespace.as_deref(), Some("billing"));
876            }
877            other => panic!("expected SnapshotRef, got {}", other.kind_str()),
878        }
879        let target = &spec.target;
880        assert_eq!(target.kind_str(), "Pvc");
881        match target {
882            RestoreTarget::Pvc(t) => {
883                assert_eq!(t.name, "postgres-data-restored");
884                assert_eq!(t.access_modes, vec![PvcAccessMode::ReadWriteOnce]);
885            }
886            other => panic!("expected Pvc, got {}", other.kind_str()),
887        }
888        assert_eq!(
889            spec.policy.as_ref().unwrap().on_missing_snapshot,
890            Some(OnMissingSnapshot::Fail)
891        );
892
893        let json = serde_json::to_value(&spec).expect("serialize");
894        let reparsed: RestoreSpec = serde_json::from_value(json).expect("reparse");
895        assert_eq!(spec, reparsed);
896    }
897
898    #[test]
899    fn restore_pvc_access_modes_render_a_closed_enum_in_the_crd_schema() {
900        // The Vec<String> → Vec<PvcAccessMode> migration must surface in the CRD:
901        // items are a closed string enum (apiserver rejects typos on new writes),
902        // and the legacy-decode `Unknown` variant never appears.
903        let crd = Restore::crd();
904        let json = serde_json::to_value(&crd).expect("serialize CRD");
905        let items = &json["spec"]["versions"][0]["schema"]["openAPIV3Schema"]["properties"]["spec"]
906            ["properties"]["target"]["properties"]["pvc"]["properties"]["accessModes"]["items"];
907        assert_eq!(
908            items["type"], "string",
909            "items must be strings; got {items}"
910        );
911        assert_eq!(
912            items["enum"],
913            serde_json::json!([
914                "ReadWriteOnce",
915                "ReadOnlyMany",
916                "ReadWriteMany",
917                "ReadWriteOncePod"
918            ]),
919            "items enum must be exactly the canonical modes; got {items}"
920        );
921    }
922
923    #[test]
924    fn restore_legacy_access_mode_decodes_to_unknown_not_an_error() {
925        // A pre-enforcement stored Restore with a bogus mode must still
926        // deserialize (a serde error would poison the typed watch stream for the
927        // whole Kind) — the value lands in `Unknown`, verbatim, and the shared
928        // validator rejects it per-CR with the value quoted.
929        let spec: RestoreSpec = from_yaml(
930            "source: { snapshotRef: { name: b } }\n\
931             target: { pvc: { name: restored, capacity: 10Gi, accessModes: [ReadWriteOnze] } }\n",
932        );
933        match &spec.target {
934            RestoreTarget::Pvc(t) => assert_eq!(
935                t.access_modes,
936                vec![PvcAccessMode::Unknown("ReadWriteOnze".into())]
937            ),
938            other => panic!("expected Pvc, got {}", other.kind_str()),
939        }
940        // And it re-serializes unchanged (read-modify-write never mutates it).
941        let json = serde_json::to_value(&spec).unwrap();
942        assert_eq!(
943            json["target"]["pvc"]["accessModes"],
944            serde_json::json!(["ReadWriteOnze"])
945        );
946    }
947
948    #[test]
949    fn restore_options_full_flag_sweep_roundtrip() {
950        // M2 flag sweep: every new `options` knob round-trips through the cluster's
951        // YAML → serde_json::Value → typed path.
952        let yaml = r#"
953source: { snapshotRef: { name: b } }
954target: { pvcRef: { name: d } }
955options:
956  enableFileDeletion: true
957  ignorePermissionErrors: false
958  writeFilesAtomically: true
959  parallel: 4
960  writeSparseFiles: true
961  skipOwners: true
962  skipPermissions: false
963  skipTimes: true
964  overwriteFiles: false
965  overwriteDirectories: false
966  overwriteSymlinks: true
967  ignoreErrors: false
968  skipExisting: true
969"#;
970        let spec: RestoreSpec = from_yaml(yaml);
971        let o = spec.options.as_ref().expect("options set");
972        assert!(o.enable_file_deletion);
973        assert_eq!(o.ignore_permission_errors, Some(false));
974        assert_eq!(o.write_files_atomically, Some(true));
975        assert_eq!(o.parallel, Some(4));
976        assert_eq!(o.write_sparse_files, Some(true));
977        assert_eq!(o.skip_owners, Some(true));
978        assert_eq!(o.skip_permissions, Some(false));
979        assert_eq!(o.skip_times, Some(true));
980        assert_eq!(o.overwrite_files, Some(false));
981        assert_eq!(o.overwrite_directories, Some(false));
982        assert_eq!(o.overwrite_symlinks, Some(true));
983        assert_eq!(o.ignore_errors, Some(false));
984        assert_eq!(o.skip_existing, Some(true));
985
986        let json = serde_json::to_value(&spec).expect("serialize");
987        assert_eq!(json["options"]["parallel"], 4);
988        assert_eq!(json["options"]["skipExisting"], true);
989        let reparsed: RestoreSpec = serde_json::from_value(json).expect("reparse");
990        assert_eq!(spec, reparsed);
991    }
992
993    #[test]
994    fn restore_options_omits_unset_leaf_fields() {
995        // A minimal `options` block that only sets `enableFileDeletion` must not
996        // serialize the other (unset) knobs.
997        let yaml = r#"
998source: { snapshotRef: { name: b } }
999target: { pvcRef: { name: d } }
1000options:
1001  enableFileDeletion: true
1002"#;
1003        let spec: RestoreSpec = from_yaml(yaml);
1004        let json = serde_json::to_value(&spec).unwrap();
1005        let opts_json = &json["options"];
1006        assert_eq!(opts_json["enableFileDeletion"], true);
1007        for key in [
1008            "ignorePermissionErrors",
1009            "writeFilesAtomically",
1010            "parallel",
1011            "writeSparseFiles",
1012            "skipOwners",
1013            "skipPermissions",
1014            "skipTimes",
1015            "overwriteFiles",
1016            "overwriteDirectories",
1017            "overwriteSymlinks",
1018            "ignoreErrors",
1019            "skipExisting",
1020        ] {
1021            assert!(opts_json.get(key).is_none(), "{key} should be absent");
1022        }
1023    }
1024
1025    #[test]
1026    fn restore_passive_populator_mode_uses_explicit_populator_target() {
1027        // ADR-0005 §9: passive populator mode is now an EXPLICIT `target.populator: {}`
1028        // (the empty-`target` form is removed). Mirrors ADR-0001 §5.5
1029        // deploy-or-restore: fromPolicy + Continue + populator.
1030        let yaml = r#"
1031source: { fromPolicy: { name: postgres-data, offset: 0 } }
1032target: { populator: {} }
1033policy: { onMissingSnapshot: Continue }
1034"#;
1035        let spec: RestoreSpec = from_yaml(yaml);
1036        assert_eq!(spec.source.kind_str(), "FromPolicy");
1037        assert_eq!(spec.target.kind_str(), "Populator");
1038        assert!(matches!(spec.target, RestoreTarget::Populator(_)));
1039        match &spec.source {
1040            RestoreSource::FromPolicy(c) => {
1041                assert_eq!(c.name, "postgres-data");
1042                assert_eq!(c.offset, 0);
1043            }
1044            other => panic!("expected FromPolicy, got {}", other.kind_str()),
1045        }
1046
1047        // Externally tagged: `{ populator: {} }`.
1048        let json = serde_json::to_value(&spec).unwrap();
1049        assert!(json["target"]["populator"].is_object());
1050        let reparsed: RestoreSpec = serde_json::from_value(json).unwrap();
1051        assert_eq!(spec, reparsed);
1052    }
1053
1054    #[test]
1055    fn restore_without_target_fails_to_deserialize() {
1056        // ADR-0005 §9 breaking change: a Restore with no `target` is invalid.
1057        let value: serde_json::Value =
1058            serde_yaml::from_str("source: { snapshotRef: { name: b } }\n").unwrap();
1059        assert!(
1060            serde_json::from_value::<RestoreSpec>(value).is_err(),
1061            "an absent target must be rejected (no empty-target form, ADR-0005 §9)"
1062        );
1063    }
1064
1065    #[test]
1066    fn restore_populator_rejects_inherit_security_context() {
1067        // ADR-0005 §9: inheritSecurityContextFrom is meaningless with a populator
1068        // target (no workload pod exists at provision time) — the validator rejects it.
1069        use crate::common::{InheritSecurityContextFrom, MoverSpec, PodSelector};
1070        use crate::validate::validate_restore;
1071        use k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector;
1072        let spec = RestoreSpec {
1073            repository: None,
1074            source: RestoreSource::FromPolicy(FromPolicy {
1075                name: "pg".into(),
1076                namespace: None,
1077                as_of: None,
1078                offset: 0,
1079                source_path: None,
1080            }),
1081            target: RestoreTarget::Populator(PopulatorTarget {}),
1082            options: None,
1083            policy: None,
1084            credential_projection: None,
1085            mover: Some(MoverSpec {
1086                inherit_security_context_from: Some(InheritSecurityContextFrom::WorkloadSelector(
1087                    PodSelector {
1088                        pod_selector: LabelSelector::default(),
1089                        container: None,
1090                    },
1091                )),
1092                ..Default::default()
1093            }),
1094            failure_policy: None,
1095        };
1096        assert!(matches!(
1097            validate_restore(&spec),
1098            Err(crate::error::ValidationError::InvalidFieldValue { .. })
1099        ));
1100
1101        // The same populator target WITHOUT inherit is fine.
1102        let ok = RestoreSpec {
1103            mover: None,
1104            ..spec
1105        };
1106        assert!(validate_restore(&ok).is_ok());
1107    }
1108
1109    #[test]
1110    fn restore_identity_source_requires_repository_in_practice() {
1111        // The `identity` source variant; spec.repository is webhook-required (not type-required).
1112        let yaml = r#"
1113repository: { kind: Repository, name: nas-primary, namespace: backups }
1114source:
1115  identity:
1116    username: postgres-data
1117    hostname: billing
1118    sourcePath: /data
1119    snapshotID: k1f1ec0a8
1120target:
1121  pvcRef: { name: postgres-data-restored }
1122"#;
1123        let spec: RestoreSpec = from_yaml(yaml);
1124        assert_eq!(spec.source.kind_str(), "Identity");
1125        assert!(spec.repository.is_some());
1126        match &spec.source {
1127            RestoreSource::Identity(i) => {
1128                assert_eq!(i.username, "postgres-data");
1129                assert_eq!(i.snapshot_id.as_deref(), Some("k1f1ec0a8"));
1130            }
1131            other => panic!("expected Identity, got {}", other.kind_str()),
1132        }
1133        assert_eq!(spec.target.kind_str(), "PvcRef");
1134
1135        let json = serde_json::to_value(&spec).unwrap();
1136        let reparsed: RestoreSpec = serde_json::from_value(json).unwrap();
1137        assert_eq!(spec, reparsed);
1138    }
1139
1140    #[test]
1141    fn restore_mover_and_failure_policy_roundtrip() {
1142        // Restore carries the same mover surface a backup gets (resources +
1143        // securityContext for UID/GID match + cache) plus a failurePolicy.
1144        let yaml = r#"
1145source: { snapshotRef: { name: app-data-backup } }
1146target: { pvcRef: { name: app-data-restored } }
1147mover:
1148  resources:
1149    requests: { cpu: 250m, memory: 512Mi }
1150    limits: { cpu: "2", memory: 4Gi }
1151  cache:
1152    capacity: 16Gi
1153    storageClassName: fast-ssd
1154  securityContext:
1155    runAsUser: 1000
1156    runAsGroup: 1000
1157    runAsNonRoot: true
1158    allowPrivilegeEscalation: false
1159    capabilities: { drop: ["ALL"] }
1160    seccompProfile: { type: RuntimeDefault }
1161  podSecurityContext:
1162    fsGroup: 1000
1163    fsGroupChangePolicy: OnRootMismatch
1164failurePolicy:
1165  backoffLimit: 4
1166  activeDeadlineSeconds: 3600
1167"#;
1168        let spec: RestoreSpec = from_yaml(yaml);
1169        let mover = spec.mover.as_ref().expect("mover");
1170        assert!(mover.resources.is_some());
1171        assert_eq!(
1172            mover.cache.as_ref().and_then(|c| c.capacity.as_deref()),
1173            Some("16Gi")
1174        );
1175        assert_eq!(
1176            mover.security_context.as_ref().and_then(|s| s.run_as_user),
1177            Some(1000)
1178        );
1179        // fsGroup is carried on the pod-level securityContext (makes a fresh restore
1180        // volume group-writable for an unprivileged mover).
1181        assert_eq!(
1182            mover.pod_security_context.as_ref().and_then(|p| p.fs_group),
1183            Some(1000)
1184        );
1185        // A hardened non-root container + fsGroup is NOT privileged: the gate lets it run.
1186        assert!(!mover.requires_privilege());
1187        let fp = spec.failure_policy.as_ref().expect("failurePolicy");
1188        assert_eq!(fp.backoff_limit, Some(4));
1189        assert_eq!(fp.active_deadline_seconds, Some(3600));
1190
1191        let json = serde_json::to_value(&spec).expect("serialize");
1192        let reparsed: RestoreSpec = serde_json::from_value(json).expect("reparse");
1193        assert_eq!(spec, reparsed);
1194    }
1195
1196    #[test]
1197    fn restore_mover_root_context_is_privileged() {
1198        // `runAsUser: 0` on a restore mover trips the same privileged-mover gate as a
1199        // backup — the controller refuses it unless the namespace opts in.
1200        let yaml = r#"
1201source: { snapshotRef: { name: app-data-backup } }
1202target: { pvcRef: { name: app-data-restored } }
1203mover:
1204  securityContext:
1205    runAsUser: 0
1206    runAsNonRoot: false
1207"#;
1208        let spec: RestoreSpec = from_yaml(yaml);
1209        assert!(spec.mover.as_ref().unwrap().requires_privilege());
1210    }
1211
1212    #[test]
1213    fn restore_crd_schema_carries_the_snapshot_inherit_variant() {
1214        // `Restore::crd()`/`SnapshotPolicy::crd()` smoke for the new externally-tagged
1215        // variant: schema generation must not panic, and the Restore schema must
1216        // surface `inheritSecurityContextFrom.snapshot` as an object property (the
1217        // SnapshotPolicy schema carries it too — the restriction is webhook-level,
1218        // not structural).
1219        let crd = Restore::crd();
1220        let json = serde_json::to_value(&crd).expect("serialize CRD");
1221        let inherit = &json["spec"]["versions"][0]["schema"]["openAPIV3Schema"]["properties"]["spec"]
1222            ["properties"]["mover"]["properties"]["inheritSecurityContextFrom"]["properties"];
1223        assert!(
1224            inherit["snapshot"].is_object(),
1225            "inheritSecurityContextFrom must carry the `snapshot` variant; got {inherit}"
1226        );
1227        assert_eq!(inherit["snapshot"]["type"], "object");
1228        // The other variants are untouched.
1229        assert!(inherit["workloadSelector"].is_object());
1230        assert!(inherit["pvcConsumer"].is_object());
1231        let _ = crate::SnapshotPolicy::crd();
1232    }
1233
1234    #[test]
1235    fn restore_snapshot_inherit_mover_roundtrip() {
1236        // The restore-only recorded-identity inherit mode, parsed the cluster's way.
1237        use crate::common::{InheritSecurityContextFrom, SnapshotInherit};
1238        let yaml = r#"
1239source: { snapshotRef: { name: app-data-backup } }
1240target: { pvcRef: { name: app-data-restored } }
1241mover:
1242  inheritSecurityContextFrom:
1243    snapshot: {}
1244"#;
1245        let spec: RestoreSpec = from_yaml(yaml);
1246        assert!(matches!(
1247            spec.mover
1248                .as_ref()
1249                .and_then(|m| m.inherit_security_context_from.as_ref()),
1250            Some(InheritSecurityContextFrom::Snapshot(SnapshotInherit {})),
1251        ));
1252        let json = serde_json::to_value(&spec).expect("serialize");
1253        assert!(json["mover"]["inheritSecurityContextFrom"]["snapshot"].is_object());
1254        let reparsed: RestoreSpec = serde_json::from_value(json).expect("reparse");
1255        assert_eq!(spec, reparsed);
1256    }
1257
1258    #[test]
1259    fn restore_source_unknown_variant_is_rejected() {
1260        let value: serde_json::Value = serde_yaml::from_str("snapshotUrl:\n  url: x\n").unwrap();
1261        assert!(serde_json::from_value::<RestoreSource>(value).is_err());
1262    }
1263
1264    #[test]
1265    fn on_missing_snapshot_and_phase_serialize_to_expected_strings() {
1266        assert_eq!(
1267            serde_json::to_value(OnMissingSnapshot::Fail).unwrap(),
1268            "Fail"
1269        );
1270        assert_eq!(
1271            serde_json::to_value(OnMissingSnapshot::Continue).unwrap(),
1272            "Continue"
1273        );
1274        assert_eq!(
1275            serde_json::to_value(RestorePhase::Restoring).unwrap(),
1276            "Restoring"
1277        );
1278        assert_eq!(
1279            serde_json::to_value(RestorePhase::Completed).unwrap(),
1280            "Completed"
1281        );
1282    }
1283
1284    // --- #443: per-claim status ------------------------------------------
1285
1286    #[test]
1287    fn restore_claim_phase_all_covers_every_variant_uniquely() {
1288        // Same tripwire as `RestorePhase`: every canonical variant is in ALL with
1289        // a unique, non-empty label. A variant added without updating ALL fails
1290        // here (and `label`'s exhaustive match won't compile at all).
1291        let labels: Vec<&str> = RestoreClaimPhase::ALL.iter().map(|p| p.label()).collect();
1292        assert_eq!(RestoreClaimPhase::ALL.len(), 6);
1293        assert!(labels.iter().all(|l| !l.is_empty()));
1294        let mut sorted = labels.clone();
1295        sorted.sort_unstable();
1296        sorted.dedup();
1297        assert_eq!(sorted.len(), labels.len(), "phase labels must be unique");
1298        assert!(RestoreClaimPhase::ALL.contains(&RestoreClaimPhase::default()));
1299    }
1300
1301    #[test]
1302    fn restore_claim_terminal_set_is_pinned() {
1303        // Driven off ALL so a new variant must be classified deliberately.
1304        let terminal: Vec<&str> = RestoreClaimPhase::ALL
1305            .iter()
1306            .filter(|p| p.is_terminal())
1307            .map(|p| p.label())
1308            .collect();
1309        assert_eq!(terminal, ["Populated", "AlreadyBound", "Failed"]);
1310        let in_flight: Vec<&str> = RestoreClaimPhase::ALL
1311            .iter()
1312            .filter(|p| !p.is_terminal())
1313            .map(|p| p.label())
1314            .collect();
1315        assert_eq!(in_flight, ["Pending", "Populating", "Rebinding"]);
1316        // The decode sentinel is never in ALL and never terminal.
1317        assert!(!RestoreClaimPhase::ALL.iter().any(|p| p.is_unknown()));
1318    }
1319
1320    #[test]
1321    fn restore_claim_phase_round_trips_and_tolerates_an_unknown_value() {
1322        for p in RestoreClaimPhase::ALL {
1323            let json = serde_json::to_value(p).expect("serialize");
1324            assert_eq!(json, p.label());
1325            let back: RestoreClaimPhase = serde_json::from_value(json).expect("decode");
1326            assert_eq!(&back, p);
1327        }
1328        // A phase written by a newer operator must not poison the typed watch.
1329        let unknown: RestoreClaimPhase =
1330            serde_json::from_value(serde_json::json!("Quiescing")).expect("decodes");
1331        assert_eq!(unknown, RestoreClaimPhase::Unknown("Quiescing".into()));
1332        assert!(unknown.is_unknown());
1333        assert!(!unknown.is_terminal());
1334        // …and is echoed back verbatim, so a read-modify-write never mutates it.
1335        assert_eq!(serde_json::to_value(&unknown).unwrap(), "Quiescing");
1336    }
1337
1338    #[test]
1339    fn restore_claim_phase_schema_publishes_only_canonical_values() {
1340        // `Unknown` is a decode-compat artifact, never an admissible write.
1341        let crd = Restore::crd();
1342        let json = serde_json::to_value(&crd).expect("serialize CRD");
1343        let phase = &json["spec"]["versions"][0]["schema"]["openAPIV3Schema"]["properties"]["status"]
1344            ["properties"]["claims"]["additionalProperties"]["properties"]["phase"];
1345        let values: Vec<String> = phase["enum"]
1346            .as_array()
1347            .map(|a| {
1348                a.iter()
1349                    .map(|v| v.as_str().unwrap_or_default().to_string())
1350                    .collect()
1351            })
1352            .unwrap_or_default();
1353        assert_eq!(
1354            values,
1355            RestoreClaimPhase::canonical(),
1356            "claim phase schema must be exactly the canonical set; got {phase}"
1357        );
1358    }
1359
1360    #[test]
1361    fn restore_status_claims_render_as_an_additional_properties_map() {
1362        // The map schema is what makes N concurrent movers safe (merge-patch
1363        // merges map keys, replaces arrays). `additionalProperties` PRUNES unknown
1364        // keys, so the per-claim object must declare every field any writer sets —
1365        // including the mover-owned `observedAt`, which every `StatusUpdate`
1366        // carries and whose absence would drop the mover's whole patch.
1367        let crd = Restore::crd();
1368        let json = serde_json::to_value(&crd).expect("serialize CRD");
1369        let claims = &json["spec"]["versions"][0]["schema"]["openAPIV3Schema"]["properties"]["status"]
1370            ["properties"]["claims"];
1371        assert_eq!(claims["type"], "object", "got {claims}");
1372        let props = claims["additionalProperties"]["properties"]
1373            .as_object()
1374            .expect("per-claim properties");
1375        for key in [
1376            "uid",
1377            "phase",
1378            "reason",
1379            "message",
1380            "sourcePath",
1381            "resolved",
1382            "pvcPrime",
1383            "job",
1384            "waitStartedAt",
1385            "observedAt",
1386            "logTail",
1387            "failure",
1388        ] {
1389            assert!(props.contains_key(key), "missing `{key}` in {claims}");
1390        }
1391    }
1392
1393    #[test]
1394    fn restore_status_claims_round_trip_through_the_apiserver_shape() {
1395        let status: RestoreStatus = from_yaml(
1396            r#"
1397phase: Restoring
1398claims:
1399  data-0:
1400    uid: 11111111-2222-3333-4444-555555555555
1401    phase: Populated
1402    reason: RestoreSucceeded
1403    sourcePath: /pvc/data-0
1404    pvcPrime: prime-1111
1405    job: r-populate-deadbeef
1406    observedAt: "2026-09-01T00:00:00Z"
1407    resolved:
1408      resolution: Snapshot
1409      kopiaSnapshotID: k1
1410  data-1:
1411    uid: 66666666-2222-3333-4444-555555555555
1412    phase: Failed
1413    reason: MoverJobFailed
1414"#,
1415        );
1416        assert_eq!(status.claims.len(), 2);
1417        let a = &status.claims["data-0"];
1418        assert_eq!(a.phase, Some(RestoreClaimPhase::Populated));
1419        assert_eq!(a.source_path.as_deref(), Some("/pvc/data-0"));
1420        assert_eq!(a.job.as_deref(), Some("r-populate-deadbeef"));
1421        assert_eq!(
1422            a.resolved
1423                .as_ref()
1424                .and_then(|r| r.kopia_snapshot_id.as_deref()),
1425            Some("k1")
1426        );
1427        assert_eq!(
1428            status.claims["data-1"].phase,
1429            Some(RestoreClaimPhase::Failed)
1430        );
1431
1432        // Structural round-trip: what we serialize decodes back identically.
1433        let json = serde_json::to_value(&status).expect("serialize");
1434        let reparsed: RestoreStatus = serde_json::from_value(json).expect("reparse");
1435        assert_eq!(status, reparsed);
1436
1437        // An empty map is omitted entirely, so a DIRECT restore's status is
1438        // byte-identical to what it was before #443.
1439        let direct = RestoreStatus::default();
1440        let json = serde_json::to_value(&direct).expect("serialize");
1441        assert!(
1442            json.get("claims").is_none(),
1443            "an empty claims map must not be written: {json}"
1444        );
1445    }
1446
1447    #[test]
1448    fn from_policy_round_trips_with_and_without_source_path() {
1449        let with: RestoreSpec = from_yaml(
1450            "source: { fromPolicy: { name: pg, sourcePath: /pvc/pgdata } }\n\
1451             target: { populator: {} }\n",
1452        );
1453        match &with.source {
1454            RestoreSource::FromPolicy(c) => {
1455                assert_eq!(c.source_path.as_deref(), Some("/pvc/pgdata"));
1456                assert_eq!(c.offset, 0);
1457            }
1458            other => panic!("expected FromPolicy, got {}", other.kind_str()),
1459        }
1460        let json = serde_json::to_value(&with).expect("serialize");
1461        assert_eq!(json["source"]["fromPolicy"]["sourcePath"], "/pvc/pgdata");
1462        let reparsed: RestoreSpec = serde_json::from_value(json).expect("reparse");
1463        assert_eq!(with, reparsed);
1464
1465        // Absent stays absent on the wire — no new key on an existing object.
1466        let without: RestoreSpec =
1467            from_yaml("source: { fromPolicy: { name: pg } }\ntarget: { populator: {} }\n");
1468        match &without.source {
1469            RestoreSource::FromPolicy(c) => assert_eq!(c.source_path, None),
1470            other => panic!("expected FromPolicy, got {}", other.kind_str()),
1471        }
1472        let json = serde_json::to_value(&without).expect("serialize");
1473        assert!(
1474            json["source"]["fromPolicy"].get("sourcePath").is_none(),
1475            "{json}"
1476        );
1477    }
1478
1479    #[test]
1480    fn from_policy_source_path_is_a_bounded_string_in_the_crd_schema() {
1481        let crd = Restore::crd();
1482        let json = serde_json::to_value(&crd).expect("serialize CRD");
1483        let field = &json["spec"]["versions"][0]["schema"]["openAPIV3Schema"]["properties"]["spec"]
1484            ["properties"]["source"]["properties"]["fromPolicy"]["properties"]["sourcePath"];
1485        assert_eq!(field["type"], "string", "got {field}");
1486        assert_eq!(field["maxLength"], 4096, "got {field}");
1487    }
1488}
kopiur_api/restore.rs

kopiur_api/
restore.rs
