From 5abbf2536e657f4003d1c3124a11f969cf0ff715 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Wed, 4 Nov 2026 03:44:12 +0100 Subject: [PATCH] Preserve committed Fabric identities across capability uncertainty agent-identity: agent/dotfiles/fabric-mesh agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: anthropic/claude-opus-5-5 --- .../fractal-fabric-adapter/src/reconcile.rs | 35 ++++-- .../src/reconcile_tests.rs | 115 +++++++++++++----- .../fabric-rekey-committed-retention.json | 11 ++ 3 files changed, 121 insertions(+), 40 deletions(-) create mode 100644 release-notes/fabric-rekey-committed-retention.json diff --git a/crates/fractal-fabric-adapter/src/reconcile.rs b/crates/fractal-fabric-adapter/src/reconcile.rs index 605f89ae157..8de48ffff32 100644 --- a/crates/fractal-fabric-adapter/src/reconcile.rs +++ b/crates/fractal-fabric-adapter/src/reconcile.rs @@ -613,11 +613,15 @@ async fn resolve_pending( journal: &mut Journal, desired: &DesiredPolicy, expiries: &Expiries, + preserved: &BTreeSet, ) -> Result> { 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, withdrawing: bool, + preserved: BTreeSet, ) -> Result { 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(¤t, &owner, &desired); &retained }; - let operations = plan(¤t, &owner, policy, &expiries); + let mut operations = plan(¤t, &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(¤t, &owner, &desired, &expiries).is_empty(), + plan(¤t, &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 } diff --git a/crates/fractal-fabric-adapter/src/reconcile_tests.rs b/crates/fractal-fabric-adapter/src/reconcile_tests.rs index 3b083fb894d..7bf310fef82 100644 --- a/crates/fractal-fabric-adapter/src/reconcile_tests.rs +++ b/crates/fractal-fabric-adapter/src/reconcile_tests.rs @@ -1134,12 +1134,46 @@ fn own_metadata_and_noncurrent_members_never_authorize_overlap_or_receipts() { } +fn serve_capability(root: &Path, capability: &'static str, replies: Vec) + -> (PathBuf, tokio::task::JoinHandle>) { + 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"}), @@ -1147,35 +1181,7 @@ async fn fabric_without_expiry_blocks_rekey_but_applies_unrelated_changes() { 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::>(); + 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()); +} diff --git a/release-notes/fabric-rekey-committed-retention.json b/release-notes/fabric-rekey-committed-retention.json new file mode 100644 index 00000000000..b71cede0c3f --- /dev/null +++ b/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."} +}