Fabric re-key R6b: committed-phase retention fix (hosted #4996, re-key v2)

crates/fractal-fabric-adapter/src/reconcile.rs

@@ -613,11 +613,15 @@ async fn resolve_pending(
     journal: &mut Journal,
     desired: &DesiredPolicy,
     expiries: &Expiries,
+    preserved: &BTreeSet<String>,
 ) -> Result<Option<Mutation>> {
     let Some(pending) = &journal.pending else {
         return Ok(None);
     };
-    let replay_allowed = pending_matches_desired(pending, desired, expiries);
+    let replay_allowed = pending_matches_desired(pending, desired, expiries)
+        && pending.request["operations"].as_array().is_some_and(|operations|
+            operations.iter().all(|operation|
+                !preserved.contains(operation["node_id"].as_str().unwrap_or_default())));
     match complete_pending(socket, pending, replay_allowed).await {
         Ok(mutation) => {
             journal.pending = None;
@@ -668,6 +672,7 @@ async fn apply(
     expiries: Expiries,
     ended: BTreeSet<String>,
     withdrawing: bool,
+    preserved: BTreeSet<String>,
 ) -> Result<ReconcileStatus> {
     let owner = format!("st/fleet/{fleet_id}");
     let grants = accepted_grants(fleet_id);
@@ -679,7 +684,7 @@ async fn apply(
             "pending Fabric batch belongs to another fleet"
         );
     }
-    resolve_pending(state_dir, socket, &mut journal, &desired, &expiries).await?;
+    resolve_pending(state_dir, socket, &mut journal, &desired, &expiries, &preserved).await?;
     let mut journal_changed = false;
     for node in ended {
         journal_changed |= journal.retired.insert(node);
@@ -693,7 +698,9 @@ async fn apply(
             retained = retained_policy(&current, &owner, &desired);
             &retained
         };
-        let operations = plan(&current, &owner, policy, &expiries);
+        let mut operations = plan(&current, &owner, policy, &expiries);
+        operations.retain(|operation|
+            !preserved.contains(operation["node_id"].as_str().unwrap_or_default()));
         ensure!(
             operations.len() <= 128,
             "Fabric reconciliation requires {} operations (limit 128); no partial policy was installed",
@@ -725,7 +732,7 @@ async fn apply(
             // Durable before sending: restarting st must not forget an uncertain commit
             // or replace a pending/timed-out cancellation with an empty-diff success.
             save_journal(state_dir, &journal).await?;
-            let mutation = resolve_pending(state_dir, socket, &mut journal, policy, &expiries)
+            let mutation = resolve_pending(state_dir, socket, &mut journal, policy, &expiries, &preserved)
                 .await?
                 .expect("pending transaction");
             journal_changed = false;
@@ -737,7 +744,8 @@ async fn apply(
         }
     }
     ensure!(
-        plan(&current, &owner, &desired, &expiries).is_empty(),
+        plan(&current, &owner, &desired, &expiries).iter().all(|operation|
+            preserved.contains(operation["node_id"].as_str().unwrap_or_default())),
         "Fabric final snapshot does not match the desired exact grants and expiry; no rekey receipt is safe"
     );
     let retired_count = journal.retired.len();
@@ -814,10 +822,18 @@ pub async fn reconcile(
             Err(error) => Some(format!("fabric_capability_unknown: {error:#}")),
         }
     };
+    let mut preserved = BTreeSet::new();
     if blocked_reason.is_some() {
-        // Never turn expiring overlap into permanent authorization on an older daemon.
-        // Keep the old route; ordinary peer additions and revocations still reconcile.
-        for (_, rotation) in rotations { desired.remove(&rotation.next_node); }
+        // Preparation may stop staging, but an installed cutover must not be
+        // undone by a transient probe failure. Freeze committed records exactly
+        // as installed, including immutable expiry and both owner namespaces.
+        for (_, rotation) in rotations {
+            if rotation.state == "committed" {
+                preserved.extend([rotation.old_node, rotation.next_node]);
+            } else {
+                desired.remove(&rotation.next_node);
+            }
+        }
         expiries.clear();
     }
     let ended = view
@@ -828,7 +844,7 @@ pub async fn reconcile(
         .filter(|endpoint| endpoint["transport"] == "fabric")
         .filter_map(|endpoint| endpoint["node"].as_str().map(str::to_owned))
         .collect();
-    let mut status = apply(state_dir, socket, fleet_id, desired, expiries, ended, false).await?;
+    let mut status = apply(state_dir, socket, fleet_id, desired, expiries, ended, false, preserved).await?;
     if let Some(reason) = blocked_reason.filter(|_| status.state == "applied") {
         status.state = "rekey_blocked".into();
         status.message = Some(reason);
@@ -857,6 +873,7 @@ pub async fn withdraw(state_dir: &Path, socket: &Path, fleet_id: &str) -> Result
         Expiries::new(),
         BTreeSet::new(),
         true,
+        BTreeSet::new(),
     )
     .await
 }

crates/fractal-fabric-adapter/src/reconcile_tests.rs

@@ -1134,48 +1134,54 @@ fn own_metadata_and_noncurrent_members_never_authorize_overlap_or_receipts() {
 }
 
 
+fn serve_capability(root: &Path, capability: &'static str, replies: Vec<Value>)
+    -> (PathBuf, tokio::task::JoinHandle<Vec<Value>>) {
+    let socket = root.join("control.sock");
+    let listener = UnixListener::bind(&socket).unwrap();
+    let server = tokio::spawn(async move {
+        let mut requests = Vec::new();
+        for mut reply in replies {
+            let (mut stream, _) = listener.accept().await.unwrap();
+            let mut line = String::new();
+            BufReader::new(&mut stream).read_line(&mut line).await.unwrap();
+            let request: Value = serde_json::from_str(&line).unwrap();
+            if request["request"]["operations"].as_array().is_some_and(|ops|
+                ops.iter().any(|op| op["op"] == "set_expiry")) {
+                reply = refusal("unsupported_operation");
+            }
+            let probing = request["type"] == "reachability_status";
+            requests.push(request);
+            if probing && capability != "old" {
+                match capability {
+                    "timeout" => tokio::time::sleep(CONTROL_TIMEOUT + Duration::from_millis(20)).await,
+                    "malformed" => stream.write_all(b"not-json\n").await.unwrap(),
+                    "truncated" => stream.write_all(b"{\"type\":\"reachability_status\"").await.unwrap(),
+                    "closed" => (),
+                    _ => unreachable!(),
+                }
+                continue;
+            }
+            let mut bytes = serde_json::to_vec(&reply).unwrap();
+            bytes.push(b'\n');
+            stream.write_all(&bytes).await.unwrap();
+        }
+        requests
+    });
+    (socket, server)
+}
+
 #[tokio::test]
 async fn fabric_without_expiry_blocks_rekey_but_applies_unrelated_changes() {
     for capability in ["old", "timeout", "malformed", "truncated", "closed"] {
         let root = tempfile::tempdir().unwrap();
-        let socket = root.path().join("control.sock");
-        let listener = UnixListener::bind(&socket).unwrap();
         let permanent = |node| peer(node, OWNER, &["st3-client-v0", "st3/fleet/test-fleet"]);
         let replies = vec![
             json!({"type":"reachability_status","version":"0.9.3+old"}),
             list(vec![permanent("peer-node"), permanent("gone")], 7),
             mutation(vec![], "complete"),
             list(vec![permanent("peer-node"), permanent("new-node")], 8),
         ];
-        let server = tokio::spawn(async move {
-            let mut requests = Vec::new();
-            for mut reply in replies {
-                let (mut stream, _) = listener.accept().await.unwrap();
-                let mut line = String::new();
-                BufReader::new(&mut stream).read_line(&mut line).await.unwrap();
-                let request: Value = serde_json::from_str(&line).unwrap();
-                if request["request"]["operations"].as_array().is_some_and(|ops|
-                    ops.iter().any(|op| op["op"] == "set_expiry")) {
-                    reply = refusal("unsupported_operation");
-                }
-                let probing = request["type"] == "reachability_status";
-                requests.push(request);
-                if probing && capability != "old" {
-                    match capability {
-                        "timeout" => tokio::time::sleep(CONTROL_TIMEOUT + Duration::from_millis(20)).await,
-                        "malformed" => stream.write_all(b"not-json\n").await.unwrap(),
-                        "truncated" => stream.write_all(b"{\"type\":\"reachability_status\"").await.unwrap(),
-                        "closed" => (),
-                        _ => unreachable!(),
-                    }
-                    continue;
-                }
-                let mut bytes = serde_json::to_vec(&reply).unwrap();
-                bytes.push(b'\n');
-                stream.write_all(&bytes).await.unwrap();
-            }
-            requests
-        });
+        let (socket, server) = serve_capability(root.path(), capability, replies);
         let mut view = rotation_view("prepared", (now_ms() / 1000) as u64 + 300);
         view.members.push(member("new", "new-key", "new-node", "current"));
         let status = reconcile(root.path(), &socket, FLEET, "own-key", &view, &BTreeSet::new()).await.unwrap();
@@ -1190,3 +1196,50 @@ async fn fabric_without_expiry_blocks_rekey_but_applies_unrelated_changes() {
         assert!(operations.iter().all(|op| op["op"] != "set_expiry" && op["node_id"] != "next-node"));
     }
 }
+
+#[tokio::test]
+async fn committed_cutover_survives_unknown_capability_without_any_policy_write() {
+    let deadline = (now_ms() / 1000) as u64 + 300;
+    for capability in ["timeout", "malformed"] {
+        for owner in [OWNER.to_owned(), format!("{RENAMED_OWNER_PREFIX}{FLEET}")] {
+            let root = tempfile::tempdir().unwrap();
+            let services = if owner == OWNER { grants(FLEET) } else {
+                BTreeSet::from([format!("{RENAMED_OWNER_PREFIX}{FLEET}"), "fractal-client-v0".into()])
+            };
+            let services = services.iter().map(String::as_str).collect::<Vec<_>>();
+            let mut old = peer("peer-node", &owner, &services);
+            old["trust_sources"][0]["expires_at"] = json!(deadline);
+            let installed = vec![old, peer("next-node", &owner, &services)];
+            let (socket, server) = serve_capability(root.path(), capability, vec![
+                json!({"type":"reachability_status","version":"0.9.4"}), list(installed, 7),
+            ]);
+            let status = reconcile(root.path(), &socket, FLEET, "own-key",
+                &rotation_view("committed", deadline), &BTreeSet::new()).await.unwrap();
+            assert_eq!(status.state, "rekey_blocked");
+            assert!(status.message.unwrap().contains("fabric_capability_unknown"));
+            let requests = server.await.unwrap();
+            assert_eq!(requests.len(), 2);
+            assert_eq!(requests[0]["type"], "reachability_status");
+            assert_eq!(requests[1]["request"]["operation"], "list",
+                "no expiry write, promotion, revocation or new grant may alter the installed cutover");
+        }
+    }
+}
+
+#[tokio::test]
+async fn blocked_committed_cutover_never_replays_a_pending_replacement() {
+    let root = tempfile::tempdir().unwrap();
+    let policy = DesiredPolicy::from([("next-node".into(), grants(FLEET))]);
+    let request = json!({"operation":"batch","owner":OWNER,"request_id":"replace-next",
+        "operations":[{"op":"remove_peer","node_id":"next-node"},
+            {"op":"add_peer","node_id":"next-node"},
+            {"op":"set_grants","node_id":"next-node","grants":grants(FLEET)}]});
+    let mut journal = Journal { pending: Some(Pending {
+        request, created_unix_ms: now_ms(),
+    }), ..Journal::default() };
+    let (socket, server) = serve(root.path(), vec![Some(mutation(vec![], "complete"))]);
+    resolve_pending(root.path(), &socket, &mut journal, &policy, &Expiries::new(),
+        &BTreeSet::from(["next-node".into(), "peer-node".into()])).await.unwrap();
+    assert_eq!(server.await.unwrap()[0]["request"]["operation"], "cancellation");
+    assert!(journal.pending.is_none());
+}

release-notes/fabric-rekey-committed-retention.json

@@ -0,0 +1,11 @@
+{
+  "version": 1,
+  "summary": "Preserve installed committed Fabric cutovers when expiry capability cannot be established",
+  "replay": {"status": "none", "detail": "No graph replay."},
+  "schema": {"status": "none", "detail": "No stored format change."},
+  "rules": {"status": "none", "detail": "No checkpoint rule change."},
+  "compatibility": {"status": "changed", "detail": "Blocked preparation stops staging. Blocked committed rotations preserve both installed identities and expiry exactly, for either owner namespace."},
+  "downtime": {"status": "none", "detail": "No restart required."},
+  "manual": {"status": "none", "detail": "Fabric v0.9.4 must still precede Fractal rekey deployment; expiry state still requires explicit migration before a v0.9.3 rollback."},
+  "recovery": {"status": "changed", "detail": "Capability failures cannot replay changes to committed identities, promote the expiring old identity, or revoke the permanent new identity. Unrelated policy operations continue."}
+}