fabric: accept both fleet adapter owners during recovery and revocation

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

@@ -25,6 +25,8 @@ const LIMIT: u64 = 1024 * 1024;
 #[derive(Serialize, Deserialize)]
 struct Journal {
     fleet: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    owner: Option<String>,
     creation: String,
     deadline: u64,
     attempt: Option<String>,
@@ -38,6 +40,24 @@ struct Journal {
 fn owner(fleet: &str) -> String {
     format!("st/fleet/{fleet}")
 }
+fn accepted_owner(value: &Value, fleet: &str) -> bool {
+    value.as_str().is_some_and(|value| {
+        value.strip_prefix("st/fleet/") == Some(fleet)
+            || value.strip_prefix(super::reconcile::RENAMED_OWNER_PREFIX) == Some(fleet)
+    })
+}
+fn journal_owner(journal: &Journal) -> Result<std::borrow::Cow<'_, str>> {
+    let selected = journal.owner.as_deref().map_or_else(
+        || std::borrow::Cow::Owned(owner(&journal.fleet)),
+        std::borrow::Cow::Borrowed,
+    );
+    ensure!(
+        selected.strip_prefix("st/fleet/") == Some(journal.fleet.as_str())
+            || selected.strip_prefix(super::reconcile::RENAMED_OWNER_PREFIX) == Some(journal.fleet.as_str()),
+        "pairing journal belongs to another fleet"
+    );
+    Ok(selected)
+}
 // Both the immutable permit and reciprocal policy are client-only. Static operator
 // terminal grants belong to Fabric configuration, never to this adapter owner.
 fn grants(fleet: &str) -> Value {
@@ -177,7 +197,8 @@ async fn snapshot(client: &reqwest::Client) -> Result<super::LeaveSnapshot> {
         bytes.extend_from_slice(&chunk);
     }
     let snapshot =
-        serde_json::from_slice::<smallclaims::sync::PeerResponse<super::LeaveSnapshot>>(&bytes)?.value;
+        serde_json::from_slice::<smallclaims::sync::PeerResponse<super::LeaveSnapshot>>(&bytes)?
+            .value;
     ensure!(
         snapshot.persistence_error.is_none() && snapshot.membership.anchor.is_some(),
         "st membership is not durable and verified"
@@ -261,8 +282,9 @@ pub async fn invite(client: &reqwest::Client, socket: &Path, state: &Path) -> Re
                 .iter()
                 .any(|endpoint| endpoint["transport"] == "fabric"
                     && endpoint["node"] == gateway
-                    && endpoint["protocol"].as_str().is_some_and(|protocol|
-                        smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, &fleet))),
+                    && endpoint["protocol"].as_str().is_some_and(|protocol| {
+                        smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, &fleet)
+                    })),
         "sponsor lacks a unique current signed endpoint for the Fabric gateway"
     );
     let deadline = code.expires_at.min((super::now_ms() / 1000) as u64 + 600);
@@ -275,6 +297,7 @@ pub async fn invite(client: &reqwest::Client, socket: &Path, state: &Path) -> Re
     );
     let mut journal = Journal {
         fleet: fleet.clone(),
+        owner: None,
         creation: creation.clone(),
         deadline,
         attempt: None,
@@ -351,7 +374,7 @@ pub async fn tick(client: &reqwest::Client, socket: &Path, state: &Path) -> Resu
         }
         let reply = control(
             socket,
-            &json!({"type":"pairing","request":{"operation":"query","owner":owner(&journal.fleet),"creation_id":journal.creation}}),
+            &json!({"type":"pairing","request":{"operation":"query","owner":journal_owner(&journal)?,"creation_id":journal.creation}}),
         )
         .await?;
         // Fabric prunes terminal receipts after 24 hours and reports this exact
@@ -385,7 +408,7 @@ pub async fn tick(client: &reqwest::Client, socket: &Path, state: &Path) -> Resu
             journal.confirmed = state_name == "confirmed";
             super::save(&path, &journal)?;
         } else if journal.attempt.is_none() {
-            local(socket, json!({"operation":"cancel","owner":owner(&journal.fleet),"creation_id":journal.creation})).await?;
+            local(socket, json!({"operation":"cancel","owner":journal_owner(&journal)?,"creation_id":journal.creation})).await?;
         } else if state_name == "pending"
             && process(client, socket, &path, &mut journal, status)
                 .await
@@ -428,7 +451,7 @@ async fn process(
     if journal.response.is_none() {
         let reply = local(
             socket,
-            json!({"operation":"pair_receive","owner":owner(&journal.fleet),"attempt_id":attempt}),
+            json!({"operation":"pair_receive","owner":journal_owner(journal)?,"attempt_id":attempt}),
         )
         .await?;
         let message = &reply["exchange"]["message"];
@@ -467,8 +490,8 @@ async fn process(
         super::save(path, journal)?;
     }
     let payload = journal.response.as_ref().context("missing answer")?;
-    local(socket, json!({"operation":"pair_send","owner":owner(&journal.fleet),"attempt_id":attempt,"sequence":1,"digest":digest(payload),"payload":payload})).await?;
-    let reply = local(socket, json!({"operation":"pair_receive","owner":owner(&journal.fleet),"attempt_id":attempt,"ack":1})).await?;
+    local(socket, json!({"operation":"pair_send","owner":journal_owner(journal)?,"attempt_id":attempt,"sequence":1,"digest":digest(payload),"payload":payload})).await?;
+    let reply = local(socket, json!({"operation":"pair_receive","owner":journal_owner(journal)?,"attempt_id":attempt,"ack":1})).await?;
     let message = &reply["exchange"]["message"];
     if message.is_null() {
         return Ok(());
@@ -499,7 +522,7 @@ async fn process(
         &journal.fleet,
     )?;
     let policy_revision = revision(socket).await?;
-    let confirmation = json!({"operation":"confirm","owner":owner(&journal.fleet),"attempt_id":attempt,"expected_node_id":node,"expected_attempt_revision":status["attempt_revision"],"expected_policy_revision":policy_revision,"request_id":format!("confirm-{attempt}-{policy_revision}")});
+    let confirmation = json!({"operation":"confirm","owner":journal_owner(journal)?,"attempt_id":attempt,"expected_node_id":node,"expected_attempt_revision":status["attempt_revision"],"expected_policy_revision":policy_revision,"request_id":format!("confirm-{attempt}-{policy_revision}")});
     journal.confirmation = Some(confirmation.clone());
     super::save(path, journal)?;
     finish_confirmation(socket, path, journal, confirmation).await?;
@@ -512,6 +535,7 @@ async fn finish_confirmation(
     journal: &mut Journal,
     confirmation: Value,
 ) -> Result<()> {
+    ensure!(journal_owner(journal)? == owner(&journal.fleet), "renamed-owner permit remains pending; reader phase cannot confirm new authority");
     let reply = control(socket, &json!({"type":"pairing","request":confirmation})).await?;
     if reply["type"] == "pairing" && reply["reply"]["error"]["code"] == "policy_revision_conflict" {
         // This explicit refusal cannot have committed. Only this case may discard
@@ -728,8 +752,12 @@ pub async fn join(socket: &Path, state: &Path) -> Result<()> {
             .iter()
             .any(|endpoint| endpoint["transport"] == "fabric"
                 && endpoint["node"] == pairing.gateway
-                && endpoint["protocol"].as_str().is_some_and(|protocol|
-                    smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, &sealed.fleet_id))),
+                && endpoint["protocol"].as_str().is_some_and(|protocol| {
+                    smallclaims::fleet::transport::is_fleet_fabric_protocol(
+                        protocol,
+                        &sealed.fleet_id,
+                    )
+                })),
         "authenticated sponsor lacks signed exact Fabric endpoint"
     );
     if let Some(resume) = &start.resume {
@@ -748,10 +776,12 @@ pub async fn join(socket: &Path, state: &Path) -> Result<()> {
     let mutation = if path.exists() {
         let retained: Value = serde_json::from_slice(&std::fs::read(&path)?)?;
         ensure!(
-            retained["request"]["owner"] == owner(&sealed.fleet_id)
+            accepted_owner(&retained["request"]["owner"], &sealed.fleet_id)
                 && retained["request"]["operations"][0]["node_id"] == pairing.gateway,
             "retained reciprocal mutation belongs to another identity"
         );
+        ensure!(retained["request"]["owner"] == owner(&sealed.fleet_id),
+            "renamed-owner reciprocal addition remains pending; reader phase cannot replay new authority");
         retained
     } else {
         json!({"type":"runtime_peers","request":{"operation":"batch","owner":owner(&sealed.fleet_id),"request_id":format!("pair-{}", pairing.attempt),"expected_revision":revision(socket).await?,"operations":[{"op":"add_peer","node_id":pairing.gateway,"direction":"normal"},{"op":"set_grants","node_id":pairing.gateway,"grants":grants(&sealed.fleet_id)}]}})

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

@@ -20,6 +20,7 @@ fn request(node: &str) -> JoinRequest {
 fn journal() -> Journal {
     Journal {
         fleet: "fleet".into(),
+        owner: None,
         creation: "creation".into(),
         deadline: 600,
         attempt: Some("attempt".into()),
@@ -66,6 +67,29 @@ fn permit_and_reciprocal_policy_grants_are_client_only() {
     assert_eq!(grants("fleet"), json!(["st3/fleet/fleet", "st3-client-v0"]));
 }
 
+#[test]
+fn recovered_pairing_owner_is_exact_and_legacy_journals_keep_their_owner() {
+    let old = serde_json::to_value(journal()).unwrap();
+    assert!(old.get("owner").is_none());
+    let mut recovered: Journal = serde_json::from_value(old).unwrap();
+    assert_eq!(journal_owner(&recovered).unwrap(), "st/fleet/fleet");
+    for owner in ["st/fleet/fleet", "fractal/fleet/fleet"] {
+        recovered.owner = Some(owner.into());
+        assert_eq!(journal_owner(&recovered).unwrap(), owner);
+    }
+    recovered.owner = Some("fractal/fleet/other".into());
+    assert!(journal_owner(&recovered).is_err());
+}
+
+#[tokio::test]
+async fn renamed_permit_cannot_confirm_or_install_grants() {
+    let mut journal = journal();
+    journal.owner = Some(format!("{}fleet", super::super::reconcile::RENAMED_OWNER_PREFIX));
+    assert!(finish_confirmation(Path::new("absent.sock"), Path::new("absent.json"),
+        &mut journal, json!({"operation":"confirm"})).await.unwrap_err()
+        .to_string().contains("remains pending"));
+}
+
 #[test]
 fn quarantined_pairing_writes_fail_closed_without_accepting_a_status() {
     let error = pairing_reply(json!({

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

@@ -17,6 +17,7 @@ use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
 use smallclaims::fleet::{FleetView, MemberView};
 
 const REQUEST_LIMIT: usize = 1024 * 1024;
+pub(super) const RENAMED_OWNER_PREFIX: &str = concat!("fractal", "/fleet/");
 const RESPONSE_LIMIT: u64 = 4 * 1024 * 1024;
 const CONTROL_TIMEOUT: Duration = Duration::from_secs(5);
 
@@ -220,6 +221,46 @@ fn grants(fleet_id: &str) -> BTreeSet<String> {
     [format!("st3/fleet/{fleet_id}"), "st3-client-v0".into()].into()
 }
 
+fn accepted_grants(fleet: &str) -> BTreeSet<String> {
+    grants(fleet)
+        .into_iter()
+        .chain([format!("{RENAMED_OWNER_PREFIX}{fleet}"), "fractal-client-v0".into()])
+        .collect()
+}
+
+fn service_allowed(expected: &BTreeSet<String>, service: &str) -> bool {
+    expected.contains(service)
+        || (service == "fractal-client-v0" && expected.contains("st3-client-v0"))
+        || service.strip_prefix(RENAMED_OWNER_PREFIX).is_some_and(|fleet| {
+            expected
+                .iter()
+                .any(|grant| grant.strip_prefix("st3/fleet/") == Some(fleet))
+        })
+}
+
+// A reader may reduce an existing renamed owner's authority, never publish new authority.
+fn retained_policy(snapshot: &Snapshot, owner: &str, desired: &DesiredPolicy) -> DesiredPolicy {
+    snapshot
+        .peers
+        .iter()
+        .filter_map(|peer| {
+            let expected = desired.get(&peer.node_id)?;
+            if !peer
+                .trust_sources
+                .iter()
+                .any(|source| source.owned_by(owner))
+            {
+                return None;
+            }
+            let actual: BTreeSet<_> = peer.services.iter()
+                .filter(|(_, sources)| sources.iter().any(|source| source.owned_by(owner)))
+                .map(|(service, _)| service.clone()).collect();
+            actual.iter().all(|service| service_allowed(expected, service))
+                .then(|| (peer.node_id.clone(), actual))
+        })
+        .collect()
+}
+
 fn desired(
     view: &FleetView,
     own_key: &str,
@@ -255,8 +296,9 @@ fn desired(
             }
             // Do not turn arbitrary signed endpoint names into Fabric grants.
             ensure!(
-                endpoint["protocol"].as_str().is_some_and(|protocol|
-                    smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, fleet_id)),
+                endpoint["protocol"].as_str().is_some_and(|protocol| {
+                    smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, fleet_id)
+                }),
                 "member {} advertises a nonstandard Fabric replication protocol; expected st3/fleet/{fleet_id} or fractal/fleet/{fleet_id}",
                 member.name
             );
@@ -318,7 +360,7 @@ fn residual(peers: &[Peer], desired: &DesiredPolicy, grants: &BTreeSet<String>)
                     grants.contains(*service)
                         && !desired
                             .get(&peer.node_id)
-                            .is_some_and(|expected| expected.contains(*service))
+                            .is_some_and(|expected| service_allowed(expected, service))
                 })
                 .map(|(service, sources)| {
                     format!(
@@ -392,6 +434,10 @@ async fn save_journal(state_dir: &Path, journal: &Journal) -> Result<()> {
 }
 
 fn pending_matches_desired(pending: &Pending, desired: &DesiredPolicy) -> bool {
+    if pending.request["owner"].as_str().is_some_and(|owner| owner.starts_with(RENAMED_OWNER_PREFIX)) {
+        return pending.request["operations"].as_array().is_some_and(|operations|
+            operations.iter().all(|operation| operation["op"] == "remove_peer"));
+    }
     pending.request["operations"]
         .as_array()
         .is_some_and(|operations| {
@@ -521,55 +567,71 @@ async fn apply(
     withdrawing: bool,
 ) -> Result<ReconcileStatus> {
     let owner = format!("st/fleet/{fleet_id}");
-    let grants = grants(fleet_id);
+    let grants = accepted_grants(fleet_id);
     let mut journal = load_journal(state_dir).await?;
+    if let Some(pending) = &journal.pending {
+        ensure!(
+            pending.request["owner"] == owner
+                || pending.request["owner"] == format!("{RENAMED_OWNER_PREFIX}{fleet_id}"),
+            "pending Fabric batch belongs to another fleet"
+        );
+    }
     resolve_pending(state_dir, socket, &mut journal, &desired).await?;
     let mut journal_changed = false;
     for node in ended {
         journal_changed |= journal.retired.insert(node);
     }
     let mut current = snapshot(socket).await?;
-    let operations = plan(&current, &owner, &desired);
-    ensure!(
-        operations.len() <= 128,
-        "Fabric reconciliation requires {} operations (limit 128); no partial policy was installed",
-        operations.len()
-    );
-    if !operations.is_empty() {
-        for operation in &operations {
-            if operation["op"] == "remove_peer" {
-                journal.retired.insert(
-                    operation["node_id"]
-                        .as_str()
-                        .context("removal has no NodeID")?
-                        .to_owned(),
-                );
-            }
-        }
-        let revision = current.policy_revision;
-        let digest = Sha256::digest(serde_json::to_vec(&json!([
-            &owner,
-            &current.instance,
-            revision,
-            &operations
-        ]))?);
-        journal.pending = Some(Pending {
-            request: json!({"operation":"batch", "owner":owner, "request_id":format!("st-{}", hex::encode(digest)),
-                "expected_revision":revision, "operations":operations}),
-            created_unix_ms: now_ms(),
-        });
-        // 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, &desired)
-            .await?
-            .expect("pending transaction");
-        journal_changed = false;
-        current = snapshot(socket).await?;
+    for owner in [owner, format!("{RENAMED_OWNER_PREFIX}{fleet_id}")] {
+        let retained;
+        let policy = if owner.starts_with("st/") {
+            &desired
+        } else {
+            retained = retained_policy(&current, &owner, &desired);
+            &retained
+        };
+        let operations = plan(&current, &owner, policy);
         ensure!(
-            current.policy_revision >= mutation.current_revision,
-            "Fabric policy revision regressed after mutation"
+            operations.len() <= 128,
+            "Fabric reconciliation requires {} operations (limit 128); no partial policy was installed",
+            operations.len()
         );
+        if !operations.is_empty() {
+            for operation in &operations {
+                if operation["op"] == "remove_peer" {
+                    journal.retired.insert(
+                        operation["node_id"]
+                            .as_str()
+                            .context("removal has no NodeID")?
+                            .to_owned(),
+                    );
+                }
+            }
+            let revision = current.policy_revision;
+            let digest = Sha256::digest(serde_json::to_vec(&json!([
+                &owner,
+                &current.instance,
+                revision,
+                &operations
+            ]))?);
+            journal.pending = Some(Pending {
+                request: json!({"operation":"batch", "owner":owner, "request_id":format!("st-{}", hex::encode(digest)),
+                "expected_revision":revision, "operations":operations}),
+                created_unix_ms: now_ms(),
+            });
+            // 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)
+                .await?
+                .expect("pending transaction");
+            journal_changed = false;
+            current = snapshot(socket).await?;
+            ensure!(
+                current.policy_revision >= mutation.current_revision,
+                "Fabric policy revision regressed after mutation"
+            );
+        }
     }
     let retired_count = journal.retired.len();
     journal.retired.retain(|node| {
@@ -590,7 +652,7 @@ async fn apply(
                 || journal.retired.contains(&peer.node_id)
                 || desired
                     .get(&peer.node_id)
-                    .is_some_and(|expected| expected.len() < grants.len())
+                    .is_some_and(|expected| expected.len() < 2)
         })
         .collect::<Vec<_>>();
     let message = residual(&relevant, &desired, &grants);