Fabric re-key R5: adapter runtime

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

@@ -18,6 +18,8 @@ use smallclaims::fleet::{FleetView, MemberView};
 
 const REQUEST_LIMIT: usize = 1024 * 1024;
 pub(super) const RENAMED_OWNER_PREFIX: &str = concat!("fractal", "/fleet/");
+#[cfg(test)]
+const RENAMED_PTY_PREFIX: &str = concat!("fractal", "/pty/");
 const RESPONSE_LIMIT: u64 = 4 * 1024 * 1024;
 const CONTROL_TIMEOUT: Duration = Duration::from_secs(5);
 
@@ -80,16 +82,19 @@ pub fn control_socket() -> Result<PathBuf> {
 #[derive(Debug, Deserialize)]
 #[serde(tag = "kind", rename_all = "snake_case")]
 enum Source {
-    Runtime { owner: String },
+    Runtime {
+        owner: String,
+        expires_at: Option<u64>,
+    },
     Static { path: PathBuf },
 }
 impl Source {
     fn owned_by(&self, owner: &str) -> bool {
-        matches!(self, Self::Runtime { owner: source_owner } if source_owner == owner)
+        matches!(self, Self::Runtime { owner: source_owner, .. } if source_owner == owner)
     }
     fn describe(&self) -> String {
         match self {
-            Self::Runtime { owner } => format!("runtime owner {owner}"),
+            Self::Runtime { owner, .. } => format!("runtime owner {owner}"),
             Self::Static { path } => format!("static {}", path.display()),
         }
     }
@@ -314,7 +319,9 @@ fn desired(
 
 type DesiredPolicy = BTreeMap<String, BTreeSet<String>>;
 
-fn plan(snapshot: &Snapshot, owner: &str, desired: &DesiredPolicy) -> Vec<Value> {
+type Expiries = BTreeMap<String, u64>;
+
+fn plan(snapshot: &Snapshot, owner: &str, desired: &DesiredPolicy, expiries: &Expiries) -> Vec<Value> {
     let owned = snapshot
         .peers
         .iter()
@@ -331,21 +338,38 @@ fn plan(snapshot: &Snapshot, owner: &str, desired: &DesiredPolicy) -> Vec<Value>
         operations.push(json!({"op":"remove_peer", "node_id":node}));
     }
     for (node, grants) in desired {
-        if let Some(peer) = owned.get(node.as_str()) {
-            let actual = peer
-                .services
-                .iter()
-                .filter(|(_, sources)| sources.iter().any(|source| source.owned_by(owner)))
-                .map(|(service, _)| service.clone())
-                .collect::<BTreeSet<_>>();
-            if actual == *grants {
-                continue;
+        let peer = owned.get(node.as_str());
+        let actual_expiry = peer.and_then(|peer| peer.trust_sources.iter().find_map(|source| match source {
+            Source::Runtime { owner: source_owner, expires_at } if source_owner == owner => Some(*expires_at),
+            _ => None,
+        })).flatten();
+        let expected_expiry = expiries.get(node).copied();
+        if owner.starts_with(RENAMED_OWNER_PREFIX) {
+            // Recovery is a reader: expiry changes may revoke the old record,
+            // never promote it by removing and re-adding grants.
+            if actual_expiry != expected_expiry {
+                operations.push(json!({"op":"remove_peer", "node_id":node}));
             }
-        } else {
-            // No label: avoid conflicting with a static source's user-selected name.
+            continue;
+        }
+        // Fabric expiry is immutable. Promotion/abort replaces the record in
+        // this same atomic batch, never renews a staged record by add_peer alone.
+        let replace = actual_expiry.is_some() && actual_expiry != expected_expiry;
+        if replace {
+            operations.push(json!({"op":"remove_peer", "node_id":node}));
+        }
+        let actual = peer.map(|peer| peer.services.iter()
+            .filter(|(_, sources)| sources.iter().any(|source| source.owned_by(owner)))
+            .map(|(service, _)| service.clone()).collect::<BTreeSet<_>>());
+        if peer.is_none() || replace {
             operations.push(json!({"op":"add_peer", "node_id":node}));
         }
-        operations.push(json!({"op":"set_grants", "node_id":node, "grants":grants}));
+        if replace || actual.as_ref() != Some(grants) {
+            operations.push(json!({"op":"set_grants", "node_id":node, "grants":grants}));
+        }
+        if expected_expiry.is_some() && (replace || actual_expiry != expected_expiry) {
+            operations.push(json!({"op":"set_expiry", "node_id":node, "expires_at":expected_expiry}));
+        }
     }
     operations
 }
@@ -433,7 +457,7 @@ async fn save_journal(state_dir: &Path, journal: &Journal) -> Result<()> {
     .await?
 }
 
-fn pending_matches_desired(pending: &Pending, desired: &DesiredPolicy) -> bool {
+fn pending_matches_desired(pending: &Pending, desired: &DesiredPolicy, expiries: &Expiries) -> 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"));
@@ -447,7 +471,9 @@ fn pending_matches_desired(pending: &Pending, desired: &DesiredPolicy) -> bool {
                 };
                 match operation["op"].as_str() {
                     Some("add_peer") => desired.contains_key(node),
-                    Some("remove_peer") => !desired.contains_key(node),
+                    Some("remove_peer") => !desired.contains_key(node)
+                        || pending.request["operations"].as_array().is_some_and(|ops| ops.iter().any(|op| op["op"] == "add_peer" && op["node_id"] == node)),
+                    Some("set_expiry") => expiries.get(node).is_some_and(|deadline| operation["expires_at"].as_u64() == Some(*deadline)),
                     Some("set_grants") => desired.get(node).is_some_and(|grants| {
                         operation["grants"].as_array().is_some_and(|services| {
                             services.len() == grants.len()
@@ -511,11 +537,12 @@ async fn resolve_pending(
     socket: &Path,
     journal: &mut Journal,
     desired: &DesiredPolicy,
+    expiries: &Expiries,
 ) -> Result<Option<Mutation>> {
     let Some(pending) = &journal.pending else {
         return Ok(None);
     };
-    let replay_allowed = pending_matches_desired(pending, desired);
+    let replay_allowed = pending_matches_desired(pending, desired, expiries);
     match complete_pending(socket, pending, replay_allowed).await {
         Ok(mutation) => {
             journal.pending = None;
@@ -563,6 +590,7 @@ async fn apply(
     socket: &Path,
     fleet_id: &str,
     desired: DesiredPolicy,
+    expiries: Expiries,
     ended: BTreeSet<String>,
     withdrawing: bool,
 ) -> Result<ReconcileStatus> {
@@ -576,21 +604,21 @@ async fn apply(
             "pending Fabric batch belongs to another fleet"
         );
     }
-    resolve_pending(state_dir, socket, &mut journal, &desired).await?;
+    resolve_pending(state_dir, socket, &mut journal, &desired, &expiries).await?;
     let mut journal_changed = false;
     for node in ended {
         journal_changed |= journal.retired.insert(node);
     }
     let mut current = snapshot(socket).await?;
-    for owner in [owner, format!("{RENAMED_OWNER_PREFIX}{fleet_id}")] {
+    for owner in [owner.clone(), 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);
+        let operations = plan(&current, &owner, policy, &expiries);
         ensure!(
             operations.len() <= 128,
             "Fabric reconciliation requires {} operations (limit 128); no partial policy was installed",
@@ -622,7 +650,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)
+            let mutation = resolve_pending(state_dir, socket, &mut journal, policy, &expiries)
                 .await?
                 .expect("pending transaction");
             journal_changed = false;
@@ -633,6 +661,10 @@ async fn apply(
             );
         }
     }
+    ensure!(
+        plan(&current, &owner, &desired, &expiries).is_empty(),
+        "Fabric final snapshot does not match the desired exact grants and expiry; no rekey receipt is safe"
+    );
     let retired_count = journal.retired.len();
     journal.retired.retain(|node| {
         !desired.contains_key(node)
@@ -682,7 +714,7 @@ pub async fn reconcile(
         .members
         .iter()
         .any(|member| member.member_key == own_key && member.state == "current");
-    let desired = desired
+    let desired: DesiredPolicy = desired
         .into_iter()
         .map(|node| {
             let peer_current = view.members.iter().any(|member| {
@@ -699,6 +731,7 @@ pub async fn reconcile(
             (node, services)
         })
         .collect();
+    let expiries = Expiries::new();
     let ended = view
         .members
         .iter()
@@ -707,7 +740,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, ended, false).await?;
+    let mut status = apply(state_dir, socket, fleet_id, desired, expiries, ended, false).await?;
     if status.state == "applied"
         && view.members.iter().any(|member| {
             member.state == "ended"
@@ -729,6 +762,7 @@ pub async fn withdraw(state_dir: &Path, socket: &Path, fleet_id: &str) -> Result
         socket,
         fleet_id,
         DesiredPolicy::new(),
+        Expiries::new(),
         BTreeSet::new(),
         true,
     )

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

@@ -38,7 +38,7 @@ fn renamed_owner_reduction_and_recovered_batches_are_reader_only() {
         BTreeSet::from(["st3/fleet/test-fleet".into()]),
     )]);
     let retained = retained_policy(&snapshot, renamed, &desired);
-    let operations = plan(&snapshot, renamed, &retained);
+    let operations = plan(&snapshot, renamed, &retained, &Expiries::new());
     assert_eq!(
         operations[0],
         json!({"op":"remove_peer","node_id":"old-node"})
@@ -47,9 +47,9 @@ fn renamed_owner_reduction_and_recovered_batches_are_reader_only() {
     let mut pending = uncertain_add();
     pending.request["owner"] = json!(renamed);
     pending.request["operations"][1]["grants"] = json!(["fractal/fleet/test-fleet"]);
-    assert!(!pending_matches_desired(&pending, &desired));
-    pending.request["operations"][1]["grants"] = json!(["fractal/pty/test-fleet"]);
-    assert!(!pending_matches_desired(&pending, &desired));
+    assert!(!pending_matches_desired(&pending, &desired, &Expiries::new()));
+    pending.request["operations"][1]["grants"] = json!([format!("{RENAMED_PTY_PREFIX}{FLEET}")]);
+    assert!(!pending_matches_desired(&pending, &desired, &Expiries::new()));
 }
 
 #[tokio::test]
@@ -59,7 +59,7 @@ async fn renamed_pending_addition_is_observed_never_sent() {
     pending.request["owner"] = json!(format!("{RENAMED_OWNER_PREFIX}{FLEET}"));
     let (socket, server) = serve(root.path(), vec![Some(mutation(vec![], "complete"))]);
     complete_pending(&socket, &pending, pending_matches_desired(&pending,
-        &DesiredPolicy::from([("peer-node".into(), grants(FLEET))]))).await.unwrap();
+        &DesiredPolicy::from([("peer-node".into(), grants(FLEET))]), &Expiries::new())).await.unwrap();
     assert_eq!(server.await.unwrap()[0]["request"]["operation"], "cancellation");
 }
 
@@ -214,7 +214,7 @@ async fn adds_peer_and_replaces_exact_grants_in_one_atomic_batch() {
         vec![
             Some(list(vec![], 7)),
             Some(mutation(vec![], "complete")),
-            Some(list(vec![], 8)),
+            Some(list(vec![peer("peer-node", OWNER, &["st3-client-v0", "st3/fleet/test-fleet"])], 8)),
         ],
     );
     let status = reconcile(
@@ -247,32 +247,47 @@ async fn adds_peer_and_replaces_exact_grants_in_one_atomic_batch() {
 #[tokio::test]
 async fn rekey_removes_old_identity_and_adds_new_in_same_transaction() {
     let root = tempfile::tempdir().unwrap();
+    let services = ["st3-client-v0", "st3/fleet/test-fleet"];
+    let mut membership = view();
+    let member_key = membership.members[1].member_key.clone();
+    membership.members[1].endpoints[0]["node"] = json!("new-node");
     let (socket, server) = serve(
         root.path(),
         vec![
-            Some(list(vec![peer("old-node", OWNER, &["exec", "echo"])], 7)),
+            Some(list(vec![peer("old-node", OWNER, &services)], 7)),
             Some(mutation(vec![], "complete")),
-            Some(list(vec![], 8)),
+            Some(list(vec![peer("new-node", OWNER, &services)], 8)),
         ],
     );
     reconcile(
         root.path(),
         &socket,
         FLEET,
         "own-key",
-        &view(),
+        &membership,
         &BTreeSet::new(),
     )
     .await
     .unwrap();
     let requests = server.await.unwrap();
-    let operations = requests[1]["request"]["operations"].as_array().unwrap();
+    assert_eq!(membership.members[1].member_key, member_key);
+    // Only list -> one mutation -> list crosses the control boundary. There is
+    // no separate remove/add request that could expose an intermediate policy.
+    assert_eq!(requests.len(), 3);
+    assert_eq!(requests[0]["request"]["operation"], "list");
+    assert_eq!(requests[2]["request"]["operation"], "list");
+    let batch = &requests[1]["request"];
+    assert_eq!(batch["operation"], "batch");
+    assert_eq!(batch["owner"], OWNER);
+    assert_eq!(batch["expected_revision"], 7);
     assert_eq!(
-        operations[0],
-        json!({"op":"remove_peer", "node_id":"old-node"})
+        batch["operations"],
+        json!([
+            {"op":"remove_peer", "node_id":"old-node"},
+            {"op":"add_peer", "node_id":"new-node"},
+            {"op":"set_grants", "node_id":"new-node", "grants":services}
+        ])
     );
-    assert_eq!(operations[1]["node_id"], "peer-node");
-    assert_eq!(operations.len(), 3);
 }
 
 #[tokio::test]
@@ -289,7 +304,7 @@ async fn existing_owner_grants_are_replaced_without_touching_other_owners() {
                 7,
             )),
             Some(mutation(vec![], "complete")),
-            Some(list(vec![], 8)),
+            Some(list(vec![peer("peer-node", OWNER, &["st3-client-v0", "st3/fleet/test-fleet"])], 8)),
         ],
     );
     reconcile(
@@ -352,7 +367,7 @@ async fn lost_reply_retries_identical_request_id_revision_and_contents() {
             Some(list(vec![], 7)),
             None,
             Some(mutation(vec![], "complete")),
-            Some(list(vec![], 8)),
+            Some(list(vec![peer("peer-node", OWNER, &["st3-client-v0", "st3/fleet/test-fleet"])], 8)),
         ],
     );
     reconcile(
@@ -677,7 +692,7 @@ async fn real_fault_class_revision_conflict_allows_fresh_reconciliation() {
             Some(conflict),
             Some(list(vec![], 8)),
             Some(mutation(vec![], "complete")),
-            Some(list(vec![], 9)),
+            Some(list(vec![peer("peer-node", OWNER, &["st3-client-v0", "st3/fleet/test-fleet"])], 9)),
         ],
     );
     assert!(
@@ -977,3 +992,15 @@ async fn leave_drain_never_replays_full_service_grants() {
         ])
     );
 }
+
+#[tokio::test]
+async fn successful_mutation_without_verified_install_is_an_error() {
+    let root = tempfile::tempdir().unwrap();
+    let (socket, server) = serve(root.path(), vec![
+        Some(list(vec![], 7)), Some(mutation(vec![], "complete")), Some(list(vec![], 8)),
+    ]);
+    assert!(reconcile(root.path(), &socket, FLEET, "own-key", &view(), &BTreeSet::new())
+        .await.unwrap_err().to_string().contains("final snapshot"));
+    server.await.unwrap();
+}
+

release-notes/fabric-rekey-runtime-expiry.json

@@ -0,0 +1 @@
+{"version": 1,"summary": "Verify exact Fabric runtime grants and immutable expiry before accepting policy installation","replay": {"status": "none","detail": "No graph replay."},"schema": {"status": "none","detail": "Reuses fleet.member-endpoints claims; no database migration."},"rules": {"status": "none","detail": "No checkpoint rule change."},"compatibility": {"status": "changed","detail": "Non-routing rekey metadata is preserved during worker refresh and removed by explicit transport withdrawal."},"downtime": {"status": "none","detail": "No service restart required."},"manual": {"status": "none","detail": "No manual step."},"recovery": {"status": "changed","detail": "Renamed-owner recovery can retain or revoke existing access only; immutable expiry changes replace writer-owned records atomically."}}