Fabric re-key R2: API (hosted #4991, re-key v2)

crates/fractal-api-core/src/api.rs

@@ -899,6 +899,10 @@ pub fn router_for_transport(state: AppState, transport: ClientTransportBoundary,
             "/v1/internal/fleet/endpoints",
             post(fleet::fleet_publish_endpoints),
         )
+        .route(
+            "/v1/internal/fleet/fabric-rekey",
+            post(fleet::fleet_fabric_rekey),
+        )
         .route("/v1/internal/replication-wake", post(replication::replication_wake))
         .route(
             "/v1/internal/replication/shard/{shard}/peers",
@@ -9548,6 +9552,7 @@ async fn replication_worker_status(
 }
 
 pub use crate::model::FleetEndpointsRequest;
+pub use smallclaims::store::fabric_rekey::{FabricRekeyAction, FabricRekeyRequest, FabricRekeyState};
 
 /// The peers a node reports on: its config peers and the current listening members it dials.
 /// A dial-out member is never dialed, and an ended name is history, so neither is reported.

crates/fractal-api-core/src/api/fleet.rs

@@ -6,7 +6,13 @@ pub async fn fleet_publish_endpoints(
 ) -> Result<Json<Value>, ApiError> {
     let store = state.store.clone();
     let written = blocking_store(move || {
-        store.publish_fleet_endpoints(&request.mode, &request.endpoints, env!("CARGO_PKG_VERSION"))
+        if request.clear_fabric_rekey {
+            store.publish_fleet_endpoints_clearing_fabric_rekey(
+                &request.mode, &request.endpoints, env!("CARGO_PKG_VERSION"),
+            )
+        } else {
+            store.publish_fleet_endpoints(&request.mode, &request.endpoints, env!("CARGO_PKG_VERSION"))
+        }
     })
     .await?;
     if written {
@@ -15,6 +21,20 @@ pub async fn fleet_publish_endpoints(
     Ok(Json(json!({ "published": written })))
 }
 
+pub async fn fleet_fabric_rekey(
+    State(state): State<AppState>,
+    Json(request): Json<FabricRekeyRequest>,
+) -> Result<Json<Value>, ApiError> {
+    let store = state.store.clone();
+    let written = blocking_store(move || {
+        store.fabric_rekey(&request, env!("CARGO_PKG_VERSION"))
+    }).await?;
+    if written {
+        signal_changed(&state);
+    }
+    Ok(Json(json!({ "published": written })))
+}
+
 /// Publish the configured trust root, not an exporter's proposed replacement key.
 pub async fn fleet_anchor(State(state): State<AppState>) -> Result<Json<Value>, ApiError> {
     let store = state.store.clone();
@@ -380,3 +400,58 @@ pub async fn admission_override(
     }
     Ok(Json(claim))
 }
+
+#[cfg(test)]
+mod rekey_tests {
+    use super::*;
+    use crate::api::tests_shared::state;
+    use crate::api::tests_shared_client_v0_tests::json_request;
+
+    #[tokio::test]
+    async fn signed_rekey_route_commits_preserves_refresh_and_rolls_back() {
+        let root = tempfile::tempdir().unwrap();
+        let state = state(root.path());
+        let (key, _) = smallclaims::fleet::MemberKey::generate().unwrap();
+        let public = key.public().to_owned();
+        state.store.bind_fleet("rekey-api-test").unwrap();
+        state.store.pin_fleet_anchor(&public).unwrap();
+        state.store.set_member_key(Some(Arc::new(key))).unwrap();
+        state.store.admit_fleet_anchor("rekey-api-test", &public, "listening").unwrap();
+        state.store.publish_fleet_endpoints("listening", &[json!({
+            "transport":"fabric","node":"a".repeat(64),"protocol":"st3/fleet/rekey-api-test"
+        })], "test").unwrap();
+        let deadline = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs() + 500;
+        let mut request = json!({"action":"prepare","old_node":"a".repeat(64),
+            "next_node":"b".repeat(64),"deadline":deadline});
+        let path = "/v1/internal/fleet/fabric-rekey";
+        let (status, body) = json_request(router(state.clone()), path, request.clone()).await;
+        assert_eq!(status, StatusCode::OK, "{body}");
+        assert_eq!(body["published"], true);
+        let (status, body) = json_request(router(state.clone()), path, request.clone()).await;
+        assert_eq!(status, StatusCode::OK, "{body}");
+        assert_eq!(body["published"], false);
+        request["action"] = json!("commit");
+        let (status, body) = json_request(router(state.clone()), path, request.clone()).await;
+        assert_eq!(status, StatusCode::OK, "{body}");
+        let (status, body) = json_request(router(state.clone()), "/v1/internal/fleet/endpoints", json!({
+            "mode":"listening", "endpoints":[{"transport":"fabric","node":"a".repeat(64),
+                "protocol":"st3/fleet/rekey-api-test"}]
+        })).await;
+        assert_eq!(status, StatusCode::OK, "{body}");
+        let view = state.store.fleet_view_sealed().unwrap();
+        assert!(view.members[0].endpoints.iter().any(|entry| entry["transport"] == "fabric" && entry["node"] == "b".repeat(64)));
+        request["action"] = json!("abort");
+        let (status, body) = json_request(router(state.clone()), path, request.clone()).await;
+        assert_eq!(status, StatusCode::OK, "{body}");
+        let view = state.store.fleet_view_sealed().unwrap();
+        assert!(view.members[0].endpoints.iter().any(|entry| entry["state"] == "aborted"));
+        let (status, body) = json_request(router(state.clone()), "/v1/internal/fleet/endpoints", json!({
+            "mode":"listening", "endpoints":[], "clear_fabric_rekey":true
+        })).await;
+        assert_eq!(status, StatusCode::OK, "{body}");
+        assert!(state.store.fleet_view_sealed().unwrap().members[0].endpoints.is_empty());
+        request["action"] = json!("commit");
+        let (status, _) = json_request(router(state), path, request).await;
+        assert!(!status.is_success());
+    }
+}

crates/fractal-kernel/src/model.rs

@@ -96,6 +96,9 @@ pub const FIRST_READINESS_FAULT_SCOPE: &str = "scheduler/first-readiness";
 pub struct FleetEndpointsRequest {
     pub mode: String,
     pub endpoints: Vec<Value>,
+    /// Explicit settings cutover; ordinary worker refreshes preserve signed rekey state.
+    #[serde(default)]
+    pub clear_fabric_rekey: bool,
 }
 
 /// The error code of a send or reply addressed to a person.

crates/fractal-peer/src/peer.rs

@@ -1977,6 +1977,7 @@ impl Backend for MainBackend {
                 &crate::model::FleetEndpointsRequest {
                     mode: mode.into(),
                     endpoints: endpoints.to_vec(),
+                    clear_fabric_rekey: false,
                 },
             )
             .await?;

crates/fractal/src/fleet_cli.rs

@@ -1126,6 +1126,8 @@ async fn save_and_publish_fleet_settings(
             &fractal_daemon::api::FleetEndpointsRequest {
                 mode: file.mode.as_str().into(),
                 endpoints,
+                clear_fabric_rekey: file.mode == fractal_daemon::config::FleetMode::DialOut
+                    || !file.transports.iter().any(|transport| transport == "fabric"),
             },
         )
         .await

crates/fractal/tests/daemon_cost.rs

@@ -166,6 +166,10 @@ const NOT_MEASURED: &[(&str, &str)] = &[
         "GET /v1/internal/fleet/leave-state",
         "proxies a live replication worker's in-memory membership/receipt snapshot over its private socket; no claim-store scan; worker snapshot and API incarnation controls are unit-covered",
     ),
+    (
+        "POST /v1/internal/fleet/fabric-rekey",
+        "requires a genuinely signed local fleet member and exact live transition metadata, absent from generated cost stores; an operator transition or adapter receipt folds current fleet membership and publishes one signed endpoint announcement; membership-history growth is not measured here; api::fleet::tests::signed_rekey_route_commits_preserves_refresh_and_rolls_back covers the real route",
+    ),
     (
         "POST /v1/internal/fleet/pairing-admission",
         "explicit one-use fleet admission proxies the bounded signed request to the live worker; uses the existing invite redemption and verified membership fold, not a polling read path",

crates/smallclaims/src/store/fabric_rekey.rs

@@ -293,4 +293,164 @@ mod tests {
         assert!(aborted.iter().any(|entry| entry["transport"] == "fabric" && entry["node"] == request.old_node));
         assert!(aborted.iter().any(|entry| entry["state"] == "aborted"));
     }
+    #[test]
+    fn prepare_refuses_wrong_identity_conflicts_extended_retries_and_expired_commit() {
+        let store = Store::open_memory("anchor", Arc::new(Plain)).unwrap();
+        configure(&store);
+        let mut invalid = request(FabricRekeyAction::Prepare);
+        invalid.old_node = "c".repeat(64);
+        assert!(store.fabric_rekey_at(&invalid, "test", 1000).is_err());
+        invalid = request(FabricRekeyAction::Prepare);
+        invalid.deadline = 1601;
+        assert!(store.fabric_rekey_at(&invalid, "test", 1000).is_err());
+        let prepared = request(FabricRekeyAction::Prepare);
+        store.fabric_rekey_at(&prepared, "test", 1000).unwrap();
+        invalid.deadline = 1600;
+        invalid.next_node = "c".repeat(64);
+        assert!(store.fabric_rekey_at(&invalid, "test", 1000).is_err());
+        invalid = prepared.clone();
+        invalid.deadline = 1601;
+        assert!(store.fabric_rekey_at(&invalid, "test", 1001).is_err());
+        assert!(store.fabric_rekey_at(&invalid, "test", 1600).is_err());
+        invalid = prepared;
+        invalid.action = FabricRekeyAction::Commit;
+        assert!(store.fabric_rekey_at(&invalid, "test", 1600).is_err());
+    }
+
+    #[test]
+    fn crashed_prepare_resumes_original_signed_deadline() {
+        let root = tempfile::tempdir().unwrap();
+        let path = root.path().join("claims.db");
+        let store = Store::open(&path, "anchor", Arc::new(Plain)).unwrap();
+        let key = configure(&store);
+        let prepared = request(FabricRekeyAction::Prepare);
+        store.fabric_rekey_at(&prepared, "test", 1000).unwrap();
+        drop(store);
+        let store = Store::open(&path, "anchor", Arc::new(Plain)).unwrap();
+        store.set_member_key(Some(key)).unwrap();
+        assert!(!store.fabric_rekey_at(&prepared, "test", 1100).unwrap());
+        let mut extended = prepared.clone();
+        extended.deadline += 1;
+        assert!(store.fabric_rekey_at(&extended, "test", 1100).is_err());
+        assert!(current_endpoints(&store).iter().any(|entry| entry["deadline"] == 1600));
+    }
+
+    #[test]
+    fn abort_before_first_prepare_is_safe_and_receipts_refuse_own_or_missing_target() {
+        let store = Store::open_memory("anchor", Arc::new(Plain)).unwrap();
+        let key = configure(&store);
+        assert!(!store.fabric_rekey_at(&request(FabricRekeyAction::Abort), "test", 1601).unwrap());
+        let mut receipt = request(FabricRekeyAction::Receipt);
+        receipt.member_key = Some(key.public().to_owned());
+        receipt.state = Some(FabricRekeyState::Prepared);
+        assert!(store.fabric_rekey_at(&receipt, "test", 1000).is_err());
+        receipt.member_key = Some("missing".into());
+        assert!(store.fabric_rekey_at(&receipt, "test", 1000).is_err());
+    }
+
+    #[test]
+    fn receipt_is_signed_and_bound_to_remote_live_state_and_all_selectors() {
+        let anchor = Store::open_memory("anchor", Arc::new(Plain)).unwrap();
+        let anchor_key = configure(&anchor);
+        let peer = Store::open_memory("peer", Arc::new(Plain)).unwrap();
+        let (peer_key, _) = MemberKey::generate().unwrap();
+        let peer_key = Arc::new(peer_key);
+        peer.bind_fleet("rekey-test").unwrap();
+        peer.pin_fleet_anchor(anchor_key.public()).unwrap();
+        peer.set_member_key(Some(peer_key.clone())).unwrap();
+        anchor.append_claim(&ClaimInput {
+            subject: "host/peer".into(), kind: "fleet.member-admitted".into(), actor: None,
+            fields: BTreeMap::from([
+                ("fleet_id".into(), json!("rekey-test")),
+                ("member_key".into(), json!(peer_key.public())),
+                ("via".into(), json!("invite")), ("mode".into(), json!("listening")),
+            ]),
+            evidence: Vec::new(), expected_subject: None, idempotency_key: None,
+        }).unwrap();
+        let mut rotation = request(FabricRekeyAction::Prepare);
+        anchor.fabric_rekey_at(&rotation, "test", 1000).unwrap();
+        let exchange = anchor.export_replication_exchange("rekey-test", &ReplicationInventory::default()).unwrap();
+        peer.receive_replication_exchange("anchor", "rekey-test", &exchange).unwrap();
+        peer.validate_replication_backlog().unwrap();
+        peer.apply_replication_repairs().unwrap();
+        peer.project_replication_backlog().unwrap();
+        let mut receipt = request(FabricRekeyAction::Receipt);
+        receipt.member_key = Some(anchor_key.public().to_owned());
+        receipt.state = Some(FabricRekeyState::Committed);
+        assert!(peer.fabric_rekey_at(&receipt, "test", 1001).is_err());
+        receipt.state = Some(FabricRekeyState::Prepared);
+        receipt.deadline -= 1;
+        assert!(peer.fabric_rekey_at(&receipt, "test", 1001).is_err());
+        receipt.deadline += 1;
+        assert!(peer.fabric_rekey_at(&receipt, "test", 1001).unwrap());
+        assert!(!peer.fabric_rekey_at(&receipt, "test", 1002).unwrap());
+        peer.publish_fleet_endpoints("listening", &[], "test").unwrap();
+        let view = peer.fleet_view_sealed().unwrap();
+        let own = view.members.iter().find(|member| member.name == "peer").unwrap();
+        assert!(own.endpoints.iter().any(|entry| entry["transport"] == "fabric-rekey-receipt"
+            && entry["state"] == "prepared" && entry["member_key"] == anchor_key.public()));
+        rotation.action = FabricRekeyAction::Commit;
+        anchor.fabric_rekey_at(&rotation, "test", 1003).unwrap();
+        let exchange = anchor.export_replication_exchange("rekey-test", &ReplicationInventory::default()).unwrap();
+        peer.receive_replication_exchange("anchor", "rekey-test", &exchange).unwrap();
+        peer.validate_replication_backlog().unwrap();
+        peer.apply_replication_repairs().unwrap();
+        peer.project_replication_backlog().unwrap();
+        assert!(peer.fabric_rekey_at(&receipt, "test", 1004).is_err());
+        receipt.state = Some(FabricRekeyState::Committed);
+        assert!(peer.fabric_rekey_at(&receipt, "test", 1004).unwrap());
+        assert!(peer.fabric_rekey_at(&receipt, "test", 1600).is_err());
+        peer.publish_fleet_endpoints_clearing_fabric_rekey("listening", &[], "test").unwrap();
+        let view = peer.fleet_view_sealed().unwrap();
+        let own = view.members.iter().find(|member| member.name == "peer").unwrap();
+        assert!(own.endpoints.iter().any(|entry| entry["transport"] == "fabric-rekey-receipt"
+            && entry["state"] == "committed"), "settings clear must preserve remote receipts");
+    }
+
+    #[test]
+    fn explicit_clear_withdraws_committed_route_without_resurrection() {
+        let store = Store::open_memory("anchor", Arc::new(Plain)).unwrap();
+        configure(&store);
+        let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
+        let mut rotation = request(FabricRekeyAction::Prepare);
+        rotation.deadline = now + 500;
+        store.fabric_rekey_at(&rotation, "test", now).unwrap();
+        rotation.action = FabricRekeyAction::Commit;
+        store.fabric_rekey_at(&rotation, "test", now).unwrap();
+        assert!(store.publish_fleet_endpoints_clearing_fabric_rekey("listening", &[], "test").unwrap());
+        assert!(current_endpoints(&store).is_empty());
+        assert!(!store.publish_fleet_endpoints("listening", &[], "test").unwrap());
+        assert!(current_endpoints(&store).is_empty());
+    }
+
+    #[test]
+    fn dial_out_publication_clears_completed_rotation_and_never_advertises_fabric() {
+        let store = Store::open_memory("anchor", Arc::new(Plain)).unwrap();
+        configure(&store);
+        let rotation = request(FabricRekeyAction::Prepare);
+        store.fabric_rekey_at(&rotation, "test", 1000).unwrap();
+        // Its historical deadline is expired: changing mode can withdraw the route.
+        store.publish_fleet_endpoints("dial-out", &[json!({
+            "transport":"fabric","node":rotation.old_node,"protocol":"fractal/fleet/rekey-test"
+        })], "test").unwrap();
+        assert!(current_endpoints(&store).is_empty());
+        assert_eq!(store.fleet_view_sealed().unwrap().members[0].mode, "dial-out");
+    }
+
+    #[test]
+    fn expired_aborted_operation_can_start_again_with_same_pending_key() {
+        let store = Store::open_memory("anchor", Arc::new(Plain)).unwrap();
+        configure(&store);
+        let mut rotation = request(FabricRekeyAction::Prepare);
+        store.fabric_rekey_at(&rotation, "test", 1000).unwrap();
+        rotation.action = FabricRekeyAction::Abort;
+        store.fabric_rekey_at(&rotation, "test", 1001).unwrap();
+        rotation.action = FabricRekeyAction::Prepare;
+        rotation.deadline = 1700;
+        assert!(store.fabric_rekey_at(&rotation, "test", 1100).is_err());
+        assert!(store.fabric_rekey_at(&rotation, "test", 1600).unwrap());
+        assert!(current_endpoints(&store).iter().any(|entry|
+            entry["transport"] == "fabric-rekey" && entry["state"] == "prepared"
+                && entry["deadline"] == 1700));
+    }
 }

release-notes/fabric-rekey-internal-api.json

@@ -0,0 +1,32 @@
+{
+  "version": 1,
+  "summary": "Expose signed Fabric rekey publication and exact peer receipts through the internal fleet API",
+  "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": "Reject changed deadlines and mismatched receipts; resume signed state after a restart and preserve explicit transport withdrawal."
+  }
+}