fix: manualRun explicit-null completedAt, tri-state Restore gate, throttle honored in every mover flow (#394, #393, #374)

crates/api/src/common/mod.rs

@@ -1273,7 +1273,30 @@ pub struct ReplicationManualRunStatus {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub phase: Option<ReplicationManualRunPhase>,
     /// RFC3339 instant the run reached a terminal phase.
-    #[serde(default, skip_serializing_if = "Option::is_none")]
+    // Deliberately serialized EVEN WHEN `None` — no `skip_serializing_if`, unlike
+    // every other optional in this struct. A non-terminal phase must emit
+    // `"completedAt": null` so the merge-patch the controller sends CLEARS the
+    // previous run's stamp. On receiving that null the apiserver either deletes
+    // the key (plain RFC-7386) or stores the null verbatim — a nullable CRD
+    // field on k8s 1.33 was observed doing the latter — and BOTH converge here:
+    // an absent key decodes to `None`, an explicit null decodes to `None`, and
+    // re-serializing either yields `null` again.
+    //
+    // Omitting the key instead leaves the old instant standing, which is both a
+    // lie in `kubectl get -o yaml` and — because the replication controllers
+    // build the noop guard's `current` by re-serializing this very struct — a
+    // guard that never converges: `desired` would omit the key while `current`
+    // carries the stale value, so every queued pass re-fires a PATCH the
+    // apiserver no-ops (#394).
+    //
+    // This depends on `patch_status` sending `kube::api::Patch::Merge`
+    // (`crates/controller/src/io/apply.rs`). Under `Patch::Apply` an explicit
+    // null does NOT clear the field, and this contract silently breaks.
+    //
+    // Kept a plain comment rather than rustdoc on purpose: doc comments become
+    // the CRD `description` (`kubectl explain`, docs/field-reference.md), and
+    // serde mechanics are not the user's business.
+    #[serde(default)]
     pub completed_at: Option<String>,
 }
 

crates/api/src/common/mover.rs

@@ -134,6 +134,36 @@ pub struct MoverDefaults {
 /// their pods self-GC instead of lingering (ADR-0005 §12).
 pub const DEFAULT_JOB_TTL_SECONDS: i64 = 3600;
 
+/// Per-side throttle overrides for a `kopia snapshot migrate` run, which reads
+/// from a SOURCE repository and writes into a DESTINATION repository under two
+/// separate kopia connections.
+///
+/// `snapshot migrate` has **no speed flags of its own** — the only lever is
+/// `kopia repository throttle set` on each side's connection (kopia persists the
+/// limits in that connection's client config, and the migrate honors them when it
+/// reopens it). So a cap here is expressed per side, and each side overrides
+/// **that side's repository's** `moverDefaults.throttle` field by field: a field
+/// set here wins, a field left unset falls back to the repository default, and
+/// all four [`Throttle`] knobs are available on each side independently.
+///
+/// Not `Copy` (neither is [`Throttle`]).
+#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct MigrateThrottle {
+    /// Caps for the SOURCE (read) side, overriding the source repository's
+    /// `moverDefaults.throttle` field by field. Applied with `repository
+    /// throttle set` on the migrate's read-only source connection — accepted
+    /// there, so a read-only source is throttled like any other.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub source: Option<Throttle>,
+    /// Caps for the DESTINATION (write) side, overriding the destination
+    /// repository's `moverDefaults.throttle` field by field. Applied with
+    /// `repository throttle set` on the destination connection the migrate
+    /// writes through.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub destination: Option<Throttle>,
+}
+
 /// Repository-wide throttling for a mover's kopia connection; each `None` leaves kopia's current limit.
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, Default, JsonSchema)]
 #[serde(rename_all = "camelCase")]

crates/api/src/common/tests.rs

@@ -1523,7 +1523,33 @@ fn replication_manual_run_status_roundtrips_camel_case() {
     assert_eq!(json["completedAt"], "2026-06-11T12:01:42Z");
     let back: ReplicationManualRunStatus = serde_json::from_value(json).unwrap();
     assert_eq!(back, st);
-    // Absent fields serialize away entirely (no explicit nulls in status).
+    // The contract (#394): `requestedAt`/`phase` serialize away when absent, but
+    // `completedAt` ALWAYS emits — as an explicit null when there is no
+    // completion instant — so the merge-patch CLEARS a previous run's stamp
+    // instead of leaving it standing over a fresh non-terminal phase. The
+    // apiserver either deletes the key or stores the null; both decode back to
+    // `None`, so neither leaves a stale timestamp.
     let empty = serde_json::to_value(ReplicationManualRunStatus::default()).unwrap();
-    assert_eq!(empty, serde_json::json!({}));
+    assert_eq!(empty, serde_json::json!({ "completedAt": null }));
+    // …and that explicit null reads back as "no completion instant", not as a
+    // decode error (the round trip both directions).
+    let back: ReplicationManualRunStatus = serde_json::from_value(empty).unwrap();
+    assert_eq!(back, ReplicationManualRunStatus::default());
+    assert!(back.completed_at.is_none());
+
+    // A non-terminal run keeps its other fields AND clears the stamp.
+    let running = serde_json::to_value(ReplicationManualRunStatus {
+        requested_at: Some("2026-06-11T13:00:00Z".into()),
+        phase: Some(ReplicationManualRunPhase::Running),
+        completed_at: None,
+    })
+    .unwrap();
+    assert_eq!(
+        running,
+        serde_json::json!({
+            "requestedAt": "2026-06-11T13:00:00Z",
+            "phase": "Running",
+            "completedAt": null,
+        })
+    );
 }

crates/api/src/consts.rs

@@ -386,6 +386,36 @@ pub const SOURCE_PVC_AVAILABLE_CONDITION: &str = "SourcePvcAvailable";
 /// `reason`/Event reason for [`SOURCE_PVC_AVAILABLE_CONDITION`] = `False`.
 pub const SOURCE_PVC_MISSING_REASON: &str = "SourcePvcMissing";
 
+/// `Restore` condition recording whether the object this restore's repository
+/// is DERIVED from exists. Set `False` (with
+/// [`RESTORE_REFERENT_MISSING_REASON`]) when the readiness gate cannot even
+/// look up the repository because a referent is absent: the explicit
+/// `spec.repository` object itself, or the `source.fromPolicy` `SnapshotPolicy`
+/// the repository ref is read from (issue #393).
+///
+/// Its own condition rather than the `Ready` one: the registry's coarse
+/// [`crate::gates::StructuralGate::trips`] filter keys on condition+polarity,
+/// so registering a `Ready=False` row would make every unrelated `Ready=False`
+/// reason on a `Snapshot`/`Restore` read as an unknown gate from a newer
+/// operator. Only this gate ever writes this condition, so it cannot misfire.
+///
+/// Structural ([`crate::gates::RESTORE_REFERENT_MISSING_GATE`]): a referent
+/// that is never created never self-heals, and while it is missing the restore
+/// sits at `phase: Pending` with its `policy.waitTimeout` window deliberately
+/// NOT started — the phase-invisible park shape of #359.
+pub const RESTORE_REFERENT_AVAILABLE_CONDITION: &str = "ReferentAvailable";
+/// `reason`/Event reason for [`RESTORE_REFERENT_AVAILABLE_CONDITION`] =
+/// `False`. Distinct from [`REPOSITORY_NOT_READY_REASON`] on purpose: that one
+/// means "the repository object exists and its backend is unreachable", which
+/// would be a lie for a `SnapshotPolicy` that was never applied.
+pub const RESTORE_REFERENT_MISSING_REASON: &str = "RestoreReferentMissing";
+/// `reason` for [`RESTORE_REFERENT_AVAILABLE_CONDITION`] = `True`: the referent
+/// appeared on a later pass (the clear-side counterpart of
+/// [`RESTORE_REFERENT_MISSING_REASON`]; controller-internal remediation copy,
+/// like the `Snapshot` gate's `SourcePvcFound`). Written ONLY when the
+/// condition already exists — the healthy wire never grows the condition.
+pub const RESTORE_REFERENT_FOUND_REASON: &str = "RestoreReferentFound";
+
 /// `Snapshot` condition: this deletion is HELD by the mass-deletion breaker
 /// (`Repository`/`ClusterRepository` `spec.deletionProtection.threshold`)
 /// until acknowledged via [`ALLOW_MASS_DELETION_ANNOTATION`] on the

crates/api/src/gates.rs

@@ -528,6 +528,33 @@ pub const SEED_MOVER_TOO_OLD_GATE: StructuralGate = StructuralGate {
     severity: GateSeverity::Warn,
 };
 
+/// A `Restore` whose repository referent does not exist (issue #393): the
+/// explicit `spec.repository` object, or the `source.fromPolicy`
+/// `SnapshotPolicy` the repository ref is derived from.
+///
+/// The readiness gate cannot verify a repository it cannot even look up, so the
+/// restore parks at `phase: Pending` and — critically — its
+/// `policy.waitTimeout` window is NOT opened (`status.waitStartedAt` stays
+/// unstamped) until the referent appears and its repository becomes `Ready`.
+/// Before #393 the gate fell through unverified here, so a slow-arriving
+/// referent silently spent the window; for a `fromPolicy` source, whose
+/// `onMissingSnapshot` defaults to `Continue`, a spent window means an EMPTY
+/// volume.
+///
+/// WARN, not Fail, for the reason [`POLICY_REPOSITORY_NOT_READY_GATE`] is:
+/// referents applied moments apart by GitOps are the ordinary case and resolve
+/// themselves within a requeue or two, so this must explain the park rather
+/// than independently turn a diagnostic red. What it must never be is INVISIBLE
+/// — a restore parked on a `SnapshotPolicy` that was never applied looks
+/// identical, by phase, to one that is simply still resolving.
+pub const RESTORE_REFERENT_MISSING_GATE: StructuralGate = StructuralGate {
+    applies_to: GateScope::SnapshotOrRestore,
+    condition: consts::RESTORE_REFERENT_AVAILABLE_CONDITION,
+    blocked_status: CONDITION_FALSE,
+    reason: consts::RESTORE_REFERENT_MISSING_REASON,
+    severity: GateSeverity::Warn,
+};
+
 /// Every human-actionable structural gate kopiur's reconcilers can park an
 /// object on.
 ///
@@ -559,6 +586,7 @@ pub const STRUCTURAL_GATES: &[StructuralGate] = &[
     SCHEDULE_FANOUT_CAPPED_GATE,
     POLICY_REPOSITORY_NOT_READY_GATE,
     SOURCE_PVC_MISSING_GATE,
+    RESTORE_REFERENT_MISSING_GATE,
     SEED_SOURCE_NOT_READY_GATE,
     SEED_SOURCE_AUTH_CONFLICT_GATE,
     SEEDING_GATE,
@@ -893,6 +921,13 @@ mod tests {
                 GateScope::Snapshot,
                 GateSeverity::Fail,
             ),
+            (
+                consts::RESTORE_REFERENT_AVAILABLE_CONDITION,
+                CONDITION_FALSE,
+                consts::RESTORE_REFERENT_MISSING_REASON,
+                GateScope::SnapshotOrRestore,
+                GateSeverity::Warn,
+            ),
             (
                 consts::SEEDED_CONDITION,
                 CONDITION_FALSE,
@@ -967,6 +1002,47 @@ mod tests {
         }
     }
 
+    /// The #393 park must be visible to `doctor` on a `Restore` — and must not
+    /// make any OTHER condition read as a gate.
+    ///
+    /// The registry's coarse [`StructuralGate::trips`] filter is
+    /// reason-agnostic, so a row registered on a condition other reconcilers
+    /// also write (`Ready`, `Resolved`, …) would make every unrelated reason on
+    /// that condition report as "a gate from a newer operator". This row's
+    /// condition is therefore written by exactly one gate and one reason.
+    #[test]
+    fn the_restore_referent_gate_is_restore_scoped_and_owns_its_condition() {
+        let row = STRUCTURAL_GATES
+            .iter()
+            .find(|g| g.reason == consts::RESTORE_REFERENT_MISSING_REASON)
+            .expect("the #393 referent-missing park must be registered");
+        // Reachable from a Restore (the only scope whose `covers_restore` is true).
+        assert!(row.applies_to.covers_restore());
+        assert_eq!(row.applies_to, GateScope::SnapshotOrRestore);
+        // Warn: referents applied moments apart by GitOps resolve themselves.
+        assert_eq!(row.severity, GateSeverity::Warn);
+        assert_eq!(row.condition, consts::RESTORE_REFERENT_AVAILABLE_CONDITION);
+        assert!(row.trips(consts::RESTORE_REFERENT_AVAILABLE_CONDITION, "False"));
+        assert!(!row.trips(consts::RESTORE_REFERENT_AVAILABLE_CONDITION, "True"));
+        // Exactly one row owns this condition, so the coarse filter can only
+        // fire for the one reason this build writes on it.
+        assert_eq!(
+            STRUCTURAL_GATES
+                .iter()
+                .filter(|g| g.condition == consts::RESTORE_REFERENT_AVAILABLE_CONDITION)
+                .count(),
+            1
+        );
+        // ...and it is NOT the `Ready` condition, which every Restore writes with
+        // many reasons — registering there would misreport all of them.
+        assert_ne!(row.condition, consts::READY_CONDITION);
+        assert!(
+            !STRUCTURAL_GATES
+                .iter()
+                .any(|g| g.condition == consts::READY_CONDITION)
+        );
+    }
+
     #[test]
     fn every_scope_classifier_is_consistent() {
         // Each scope covers exactly one of the four kind families a consumer

crates/api/src/maintenance.rs

@@ -609,7 +609,24 @@ pub struct ManualRunStatus {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub phase: Option<ManualRunPhase>,
     /// RFC3339 instant the run reached a terminal phase.
-    #[serde(default, skip_serializing_if = "Option::is_none")]
+    // Deliberately serialized EVEN WHEN `None` (no `skip_serializing_if`), for
+    // the same reason as `common::ReplicationManualRunStatus::completed_at`: a
+    // non-terminal phase emits `"completedAt": null` so the merge-patch CLEARS
+    // the previous run's stamp instead of leaving it standing over a fresh
+    // `Running` (#394). The apiserver answers that null by deleting the key
+    // (plain RFC-7386) or by storing the null verbatim — a nullable CRD field on
+    // k8s 1.33 was observed doing the latter — and both decode back to `None`,
+    // so no stale timestamp survives either way. Maintenance patches `manualRun`
+    // unconditionally (no noop guard), so here the stake is a truthful status
+    // rather than a non-converging write loop.
+    //
+    // This depends on `patch_status` sending `kube::api::Patch::Merge`
+    // (`crates/controller/src/io/apply.rs`). Under `Patch::Apply` an explicit
+    // null does NOT clear the field, and this contract silently breaks.
+    //
+    // Kept a plain comment rather than rustdoc on purpose: doc comments become
+    // the CRD `description` (`kubectl explain`, docs/field-reference.md).
+    #[serde(default)]
     pub completed_at: Option<String>,
 }
 
@@ -677,6 +694,64 @@ mod tests {
         }
     }
 
+    #[test]
+    fn manual_run_status_roundtrips_camel_case_and_nulls_a_missing_completion() {
+        // Parsed the cluster's way (YAML -> serde_json::Value -> typed), which
+        // is the only path that proves the camelCase wire names land.
+        let status: MaintenanceStatus = from_yaml(
+            "manualRun:\n  requestedAt: 2026-06-11T12:00:00Z\n  mode: full\n  phase: Succeeded\n  completedAt: 2026-06-11T12:01:42Z\n",
+        );
+        let manual = status.manual_run.expect("manualRun decodes");
+        assert_eq!(manual.requested_at.as_deref(), Some("2026-06-11T12:00:00Z"));
+        assert_eq!(manual.mode, Some(ManualRunMode::Full));
+        assert_eq!(manual.phase, Some(ManualRunPhase::Succeeded));
+        assert_eq!(manual.completed_at.as_deref(), Some("2026-06-11T12:01:42Z"));
+        let json = serde_json::to_value(&manual).unwrap();
+        assert_eq!(json["requestedAt"], "2026-06-11T12:00:00Z");
+        assert_eq!(json["mode"], "full");
+        assert_eq!(json["phase"], "Succeeded");
+        assert_eq!(json["completedAt"], "2026-06-11T12:01:42Z");
+
+        // #394: a non-terminal run emits an EXPLICIT null completedAt, so the
+        // merge-patch clears the previous run's stamp rather than leaving it
+        // standing over a fresh `Running`. Whether the apiserver deletes the key
+        // or stores the null, it decodes back to `None`.
+        let running = serde_json::to_value(ManualRunStatus {
+            requested_at: Some("2026-06-11T13:00:00Z".into()),
+            mode: Some(ManualRunMode::Quick),
+            phase: Some(ManualRunPhase::Running),
+            completed_at: None,
+        })
+        .unwrap();
+        assert_eq!(
+            running,
+            serde_json::json!({
+                "requestedAt": "2026-06-11T13:00:00Z",
+                "mode": "quick",
+                "phase": "Running",
+                "completedAt": null,
+            })
+        );
+        // The explicit null reads back as "no completion instant".
+        let back: ManualRunStatus = serde_json::from_value(running).unwrap();
+        assert!(back.completed_at.is_none());
+        assert_eq!(
+            serde_json::to_value(ManualRunStatus::default()).unwrap(),
+            serde_json::json!({ "completedAt": null })
+        );
+
+        // A status that never requested a run still has NO manualRun key —
+        // the parent field keeps its own skip_serializing_if.
+        let never: MaintenanceStatus = from_yaml("observedGeneration: 3\n");
+        assert!(never.manual_run.is_none());
+        assert!(
+            serde_json::to_value(&never)
+                .unwrap()
+                .get("manualRun")
+                .is_none()
+        );
+    }
+
     #[test]
     fn lease_identity_is_hostname_safe_and_stable() {
         let (user, host) = kopia_lease_identity("kopiur/media/My_App.x");

crates/api/src/restore.rs

@@ -426,13 +426,17 @@ pub struct RestoreStatus {
     pub timing: Option<RestoreTiming>,
     /// When the `policy.waitTimeout` window OPENED (RFC3339) — the first reconcile on which
     /// the restore could actually proceed (its repository reached `Ready`, and for a
-    /// `target.populator` a PVC already claims it), NOT when the Restore was created. If the
-    /// referenced `Repository`/`SnapshotPolicy` object doesn't exist yet, the readiness gate
-    /// falls through unverified and the anchor can be stamped as early as creation. Stamped
-    /// once and then honored verbatim, so the window survives controller restarts and Job pod
-    /// retries; cleared when a populator re-opens resolution for a re-created claim, so that
-    /// claim gets the full window again. Absent means the window has not opened yet (or no
-    /// `policy.waitTimeout` is configured, in which case there is no window to anchor).
+    /// `target.populator` a PVC already claims it), NOT when the Restore was created. The
+    /// window does NOT open while the referenced `Repository` object or `fromPolicy`
+    /// `SnapshotPolicy` doesn't exist: the restore parks in `Pending`
+    /// (`ReferentAvailable=False`, reason `RestoreReferentMissing`) and stays unstamped. It
+    /// DOES still open for a `snapshotRef` whose `Snapshot` row doesn't exist yet (so
+    /// `onMissingSnapshot` can fire for a ref that never appears) and for a restore whose
+    /// mover Job already launched. Stamped once and then honored verbatim, so the window
+    /// survives controller restarts and Job pod retries; cleared when a populator re-opens
+    /// resolution for a re-created claim, so that claim gets the full window again. Absent
+    /// means the window has not opened yet (or no `policy.waitTimeout` is configured, in
+    /// which case there is no window to anchor).
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub wait_started_at: Option<String>,
     /// Bytes/files restored so far, patched periodically by the mover.