kopiur_api/
repository_replication.rs

1//! The `RepositoryReplication` CRD — mirror a repository's blobs to a second
2//! backend on a schedule (ADR-0005 §13(d)). The one net-new CRD: it is the "2" in
3//! 3-2-1 backup, wrapping `kopia repository sync-to`.
4//!
5//! It is **namespaced** (it lives alongside its source repository, mirroring
6//! `Maintenance`) and references either a namespaced `Repository` or a cluster-scoped
7//! `ClusterRepository` via a [`RepositoryRef`]. The controller schedules a per-slot
8//! mover Job (croner + deterministic jitter, single-flight, repo-ready gate,
9//! transition-guarded status) exactly like `Maintenance`.
10
11use crate::backend::Backend;
12use crate::common::{CronSpec, MoverSpec, RepositoryRef};
13use k8s_openapi::apimachinery::pkg::apis::meta::v1::Condition;
14use kube::CustomResource;
15use schemars::JsonSchema;
16use serde::{Deserialize, Serialize};
17
18/// Mirror a source repository's blobs to a destination backend on a schedule (`kopia repository sync-to`).
19///
20/// Not `Eq`: `mover` transitively embeds k8s-openapi types.
21#[derive(CustomResource, Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
22#[kube(
23    group = "kopiur.home-operations.com",
24    version = "v1alpha1",
25    kind = "RepositoryReplication",
26    plural = "repositoryreplications",
27    namespaced,
28    status = "RepositoryReplicationStatus",
29    shortname = "kopiarepl",
30    category = "kopiur",
31    printcolumn = r#"{"name":"Source","type":"string","jsonPath":".spec.sourceRef.name"}"#,
32    printcolumn = r#"{"name":"Destination","type":"string","jsonPath":".status.destinationBackend"}"#,
33    printcolumn = r#"{"name":"Schedule","type":"string","jsonPath":".spec.schedule.cron"}"#,
34    printcolumn = r#"{"name":"Last","type":"date","jsonPath":".status.lastReplicated"}"#,
35    printcolumn = r#"{"name":"Age","type":"date","jsonPath":".metadata.creationTimestamp"}"#
36)]
37#[serde(rename_all = "camelCase")]
38pub struct RepositoryReplicationSpec {
39    /// Reference to the `Repository` or `ClusterRepository` to mirror from.
40    pub source_ref: RepositoryRef,
41    /// The backend to mirror to; must differ from the source's backend (webhook-enforced).
42    ///
43    /// `kopia repository sync-to` is a blob-level copy: the destination inherits the
44    /// source repository's format and encryption password verbatim, so there is no
45    /// separate destination password to configure. The destination backend's own
46    /// access credentials (e.g. S3 keys) ride its `auth.secretRef`, which — like the
47    /// source's — must live in this CR's namespace.
48    pub destination: Backend,
49    /// Cron and deterministic jitter for the replication runs.
50    pub schedule: CronSpec,
51    /// Mover (Job pod) overrides for the replication run.
52    #[serde(default, skip_serializing_if = "Option::is_none")]
53    pub mover: Option<MoverSpec>,
54    /// Pause this replication; a suspended replication runs no syncs (default `false`).
55    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
56    pub suspend: bool,
57    /// Tuning knobs for the underlying `kopia repository sync-to` invocation
58    /// (issue #216). `None` reproduces today's behavior: sequential copy
59    /// (`--parallel` unset), additive sync (no `--delete`), kopia's own
60    /// `--must-exist`/`--times`/`--update` defaults, and no throughput cap.
61    #[serde(default, skip_serializing_if = "Option::is_none")]
62    pub sync: Option<SyncOptions>,
63}
64
65/// Tuning knobs for `kopia repository sync-to` (issue #216): copy parallelism,
66/// destination pruning, and the blob-sync tri-states/throughput caps kopia
67/// exposes on the command. Every field's `None`/`false` reproduces kopia's own
68/// default, so an absent `sync` block is exactly today's behavior. Pure scalars
69/// (no k8s-openapi embeds), so this derives `Eq` unlike its parent spec.
70#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, Default, JsonSchema)]
71#[serde(rename_all = "camelCase")]
72pub struct SyncOptions {
73    /// `--parallel`: number of concurrent blob-copy workers (kopia default `1` —
74    /// sequential, the root cause of #216's multi-week seed times to R2).
75    #[serde(default, skip_serializing_if = "Option::is_none")]
76    pub parallel: Option<u32>,
77    /// `--delete`: prune destination-only blobs so the mirror is an exact copy
78    /// (kopia default `false` — additive sync, never removes destination
79    /// content). Named `deleteExtra`, not kopia's bare `delete`: a `delete: true`
80    /// key on backup-adjacent YAML is dangerously ambiguous at a glance.
81    ///
82    /// CAUTION: with this `true`, blobs present at the destination but absent
83    /// from the source are deleted on every run.
84    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
85    pub delete_extra: bool,
86    /// `--[no-]must-exist`: fail the sync instead of initializing the
87    /// destination's repository-format blob (kopia default `false` — sync-to may
88    /// create the destination layout on first run).
89    #[serde(default, skip_serializing_if = "Option::is_none")]
90    pub must_exist: Option<bool>,
91    /// `--[no-]times`: synchronize blob modification times to the destination,
92    /// when the destination backend supports it (kopia default `true`).
93    #[serde(default, skip_serializing_if = "Option::is_none")]
94    pub times: Option<bool>,
95    /// `--[no-]update`: update blobs already present at the destination when the
96    /// source copy is newer (kopia default `true`).
97    #[serde(default, skip_serializing_if = "Option::is_none")]
98    pub update: Option<bool>,
99    /// `--max-download-speed`: cap read throughput from the source, in
100    /// bytes/sec (kopia default: unlimited).
101    #[serde(default, skip_serializing_if = "Option::is_none")]
102    pub max_download_speed_bytes_per_second: Option<i64>,
103    /// `--max-upload-speed`: cap write throughput to the destination, in
104    /// bytes/sec (kopia default: unlimited).
105    #[serde(default, skip_serializing_if = "Option::is_none")]
106    pub max_upload_speed_bytes_per_second: Option<i64>,
107}
108
109/// Lifecycle phase of a replication.
110#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, Default, JsonSchema)]
111pub enum RepositoryReplicationPhase {
112    /// Admitted, not yet run (also the default).
113    #[default]
114    Pending,
115    /// A replication mover Job is in flight.
116    Replicating,
117    /// The most recent replication completed successfully (idle until the next slot).
118    Succeeded,
119    /// The most recent replication run failed; see conditions.
120    Failed,
121    /// Suspended via `spec.suspend`.
122    Suspended,
123}
124
125impl crate::common::PhaseLabel for RepositoryReplicationPhase {
126    const ALL: &'static [Self] = &[
127        Self::Pending,
128        Self::Replicating,
129        Self::Succeeded,
130        Self::Failed,
131        Self::Suspended,
132    ];
133    fn label(&self) -> &'static str {
134        match self {
135            Self::Pending => "Pending",
136            Self::Replicating => "Replicating",
137            Self::Succeeded => "Succeeded",
138            Self::Failed => "Failed",
139            Self::Suspended => "Suspended",
140        }
141    }
142}
143
144/// Observed state of a `RepositoryReplication`.
145#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Default, JsonSchema)]
146#[serde(rename_all = "camelCase")]
147pub struct RepositoryReplicationStatus {
148    /// Current lifecycle phase.
149    #[serde(default, skip_serializing_if = "Option::is_none")]
150    pub phase: Option<RepositoryReplicationPhase>,
151    /// `metadata.generation` last reconciled, for staleness detection / kstatus.
152    #[serde(default, skip_serializing_if = "Option::is_none")]
153    pub observed_generation: Option<i64>,
154    /// The destination backend kind, for the `DESTINATION` print column.
155    #[serde(default, skip_serializing_if = "Option::is_none")]
156    pub destination_backend: Option<String>,
157    /// RFC3339 timestamp of the most recent successful replication run.
158    #[serde(default, skip_serializing_if = "Option::is_none")]
159    pub last_replicated: Option<String>,
160    /// RFC3339 timestamp of the next scheduled replication run (cron + jitter, pinned).
161    #[serde(default, skip_serializing_if = "Option::is_none")]
162    pub next_scheduled_at: Option<String>,
163    /// Bytes replicated by the last successful run (best-effort from kopia output).
164    #[serde(default, skip_serializing_if = "Option::is_none")]
165    pub last_replicated_bytes: Option<i64>,
166    /// Blobs replicated by the last successful run (best-effort).
167    #[serde(default, skip_serializing_if = "Option::is_none")]
168    pub last_replicated_blobs: Option<i64>,
169    /// Standard Kubernetes conditions (`Ready`, `Reconciling`, `Stalled`).
170    #[serde(default, skip_serializing_if = "Vec::is_empty")]
171    pub conditions: Vec<Condition>,
172}
173
174#[cfg(test)]
175mod tests {
176    use super::*;
177    use crate::common::RepositoryKind;
178    use crate::testutil::from_yaml;
179    use kube::core::CustomResourceExt;
180
181    #[test]
182    fn repository_replication_crd_metadata_is_correct() {
183        let crd = RepositoryReplication::crd();
184        assert_eq!(crd.spec.group, "kopiur.home-operations.com");
185        assert_eq!(crd.spec.names.kind, "RepositoryReplication");
186        assert_eq!(crd.spec.names.plural, "repositoryreplications");
187        assert_eq!(crd.spec.scope, "Namespaced");
188        assert_eq!(crd.spec.versions[0].name, "v1alpha1");
189    }
190
191    #[test]
192    fn repository_replication_roundtrip() {
193        // sourceRef + destination (externally-tagged backend) + schedule.
194        let yaml = r#"
195sourceRef:
196  kind: Repository
197  name: nas-primary
198destination:
199  s3:
200    bucket: offsite-mirror
201    region: us-east-1
202    auth:
203      secretRef:
204        name: offsite-creds
205schedule:
206  cron: "0 5 * * *"
207  jitter: 1h
208suspend: false
209"#;
210        let spec: RepositoryReplicationSpec = from_yaml(yaml);
211        assert_eq!(spec.source_ref.kind, RepositoryKind::Repository);
212        assert_eq!(spec.source_ref.name, "nas-primary");
213        // Destination is exactly one backend variant (the type guarantees it).
214        match &spec.destination {
215            Backend::S3(s3) => assert_eq!(s3.bucket, "offsite-mirror"),
216            other => panic!("expected S3 destination, got {}", other.kind_str()),
217        }
218        assert_eq!(spec.schedule.cron, "0 5 * * *");
219        assert_eq!(spec.schedule.jitter.as_deref(), Some("1h"));
220        assert!(!spec.suspend);
221
222        let json = serde_json::to_value(&spec).expect("serialize");
223        // Externally tagged destination backend.
224        assert_eq!(json["destination"]["s3"]["bucket"], "offsite-mirror");
225        let reparsed: RepositoryReplicationSpec = serde_json::from_value(json).expect("reparse");
226        assert_eq!(spec, reparsed);
227    }
228
229    #[test]
230    fn minimal_true_mirror_spec_omits_optionals() {
231        // A true mirror reuses the source password (sync-to is a blob copy), so there
232        // is no destination-encryption knob to set.
233        let yaml = r#"
234sourceRef: { name: nas-primary }
235destination: { filesystem: { path: /mirror } }
236schedule: { cron: "0 6 * * 0" }
237"#;
238        let spec: RepositoryReplicationSpec = from_yaml(yaml);
239        // sourceRef.kind defaults to Repository.
240        assert_eq!(spec.source_ref.kind, RepositoryKind::Repository);
241        let json = serde_json::to_value(&spec).unwrap();
242        assert!(json.get("suspend").is_none());
243        // #216: no `sync` block set → the field is entirely absent on the wire,
244        // reproducing today's argv exactly (no dormant defaults sneak in).
245        assert!(spec.sync.is_none());
246        assert!(json.get("sync").is_none());
247    }
248
249    #[test]
250    fn sync_options_roundtrip_full_block() {
251        // #216: every `spec.sync` tuning knob round-trips through the cluster's
252        // YAML → serde_json::Value → typed path.
253        let yaml = r#"
254sourceRef: { name: nas-primary }
255destination: { filesystem: { path: /mirror } }
256schedule: { cron: "0 5 * * *" }
257sync:
258  parallel: 8
259  deleteExtra: true
260  mustExist: false
261  times: true
262  update: false
263  maxDownloadSpeedBytesPerSecond: 1000000
264  maxUploadSpeedBytesPerSecond: 500000
265"#;
266        let spec: RepositoryReplicationSpec = from_yaml(yaml);
267        let sync = spec.sync.expect("sync block set");
268        assert_eq!(sync.parallel, Some(8));
269        assert!(sync.delete_extra);
270        assert_eq!(sync.must_exist, Some(false));
271        assert_eq!(sync.times, Some(true));
272        assert_eq!(sync.update, Some(false));
273        assert_eq!(sync.max_download_speed_bytes_per_second, Some(1_000_000));
274        assert_eq!(sync.max_upload_speed_bytes_per_second, Some(500_000));
275
276        let json = serde_json::to_value(&spec).expect("serialize");
277        assert_eq!(json["sync"]["parallel"], 8);
278        assert_eq!(json["sync"]["deleteExtra"], true);
279        assert_eq!(json["sync"]["mustExist"], false);
280        let reparsed: RepositoryReplicationSpec = serde_json::from_value(json).expect("reparse");
281        assert_eq!(spec, reparsed);
282    }
283
284    #[test]
285    fn sync_options_omits_unset_leaf_fields() {
286        // A `sync` block that only sets `parallel` must not serialize the other
287        // (unset) knobs — `deleteExtra`'s `false` default also skips (Not::not).
288        let yaml = r#"
289sourceRef: { name: nas-primary }
290destination: { filesystem: { path: /mirror } }
291schedule: { cron: "0 5 * * *" }
292sync:
293  parallel: 4
294"#;
295        let spec: RepositoryReplicationSpec = from_yaml(yaml);
296        let json = serde_json::to_value(&spec).unwrap();
297        let sync_json = &json["sync"];
298        assert_eq!(sync_json["parallel"], 4);
299        assert!(sync_json.get("deleteExtra").is_none());
300        assert!(sync_json.get("mustExist").is_none());
301        assert!(sync_json.get("times").is_none());
302        assert!(sync_json.get("update").is_none());
303        assert!(sync_json.get("maxDownloadSpeedBytesPerSecond").is_none());
304        assert!(sync_json.get("maxUploadSpeedBytesPerSecond").is_none());
305    }
306
307    #[test]
308    fn stored_cr_with_removed_destination_encryption_still_deserializes() {
309        // The field was removed (sync-to is a blob copy; it never did anything). A CR
310        // stored while the field existed must still round-trip: serde silently drops
311        // the now-unknown key (no `deny_unknown_fields`), so existing objects keep
312        // reconciling instead of failing to decode.
313        let yaml = r#"
314sourceRef: { name: nas-primary }
315destination: { filesystem: { path: /mirror } }
316destinationEncryption:
317  passwordSecretRef: { name: legacy-creds, key: KOPIA_PASSWORD }
318schedule: { cron: "0 6 * * 0" }
319"#;
320        let spec: RepositoryReplicationSpec = from_yaml(yaml);
321        assert_eq!(spec.source_ref.name, "nas-primary");
322        assert_eq!(spec.schedule.cron, "0 6 * * 0");
323    }
324
325    #[test]
326    fn replication_phase_all_covers_every_variant() {
327        use crate::common::PhaseLabel;
328        let labels: Vec<&str> = RepositoryReplicationPhase::ALL
329            .iter()
330            .map(|p| p.label())
331            .collect();
332        assert_eq!(RepositoryReplicationPhase::ALL.len(), 5);
333        assert!(labels.iter().all(|l| !l.is_empty()));
334    }
335
336    #[test]
337    fn status_roundtrips() {
338        let status: RepositoryReplicationStatus = from_yaml(
339            "phase: Succeeded\ndestinationBackend: s3\nlastReplicated: 2026-06-09T05:00:00Z\nlastReplicatedBytes: 12345\n",
340        );
341        assert_eq!(status.phase, Some(RepositoryReplicationPhase::Succeeded));
342        assert_eq!(status.destination_backend.as_deref(), Some("s3"));
343        assert_eq!(status.last_replicated_bytes, Some(12345));
344        let json = serde_json::to_value(&status).unwrap();
345        let reparsed: RepositoryReplicationStatus = serde_json::from_value(json).unwrap();
346        assert_eq!(status, reparsed);
347    }
348}
kopiur_api/repository_replication.rs

kopiur_api/
repository_replication.rs
