Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions crates/fractal-api-core/src/api.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -9406,6 +9410,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.
Expand Down
77 changes: 76 additions & 1 deletion crates/fractal-api-core/src/api/fleet.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand All @@ -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();
Expand Down Expand Up @@ -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());
}
}
3 changes: 3 additions & 0 deletions crates/fractal-kernel/src/model.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
1 change: 1 addition & 0 deletions crates/fractal-peer/src/peer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1977,6 +1977,7 @@ impl Backend for MainBackend {
&crate::model::FleetEndpointsRequest {
mode: mode.into(),
endpoints: endpoints.to_vec(),
clear_fabric_rekey: false,
},
)
.await?;
Expand Down
2 changes: 2 additions & 0 deletions crates/fractal/src/fleet_cli.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
27 changes: 27 additions & 0 deletions crates/smallclaims/src/store.rs
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,7 @@ mod sealed_view_cache_tests;
mod repair_operation_tests;
#[cfg(test)]
mod schema_install_tests;
pub mod fabric_rekey;
#[cfg(any(test, feature = "test-support"))]
pub mod contents;
pub mod prepared_seal;
Expand Down Expand Up @@ -844,6 +845,8 @@ pub struct Store {
/// it instead of building their own.
pub replication_snapshot_build: Mutex<()>,
pub replication_sync: Mutex<BTreeMap<String, PeerSyncProgress>>,
/// Serializes endpoint refreshes and signed Fabric rekey transitions.
fleet_endpoints_publication: Mutex<()>,
/// Whether exchanges compare projections with a peer at all, keeping what a reproduction
/// needs when they differ. Off unless the daemon's divergence check is on: a default
/// install never compares, reports no divergence, and settles sync on claims alone.
Expand Down Expand Up @@ -1305,6 +1308,7 @@ impl Store {
replication_snapshot: Mutex::new(None),
replication_snapshot_build: Mutex::new(()),
replication_sync: Mutex::new(BTreeMap::new()),
fleet_endpoints_publication: Mutex::new(()),
divergence_capture: std::sync::atomic::AtomicBool::new(false),
trim_local: AtomicBool::new(false),
trim_daily_claims: AtomicU64::new(checkpoint_trim::LOCAL_TRIM_DAILY_CLAIMS),
Expand Down Expand Up @@ -2596,6 +2600,29 @@ impl Store {
mode: &str,
endpoints: &[Value],
build: &str,
) -> Result<bool> {
let _publication = self.fleet_endpoints_publication.lock().unwrap_or_else(PoisonError::into_inner);
let endpoints = self.preserve_fabric_rekey(mode, endpoints, mode != "listening")?;
self.publish_fleet_endpoints_locked(mode, &endpoints, build)
}

/// Explicitly withdraw rotation metadata when changing transport settings.
pub fn publish_fleet_endpoints_clearing_fabric_rekey(
&self,
mode: &str,
endpoints: &[Value],
build: &str,
) -> Result<bool> {
let _publication = self.fleet_endpoints_publication.lock().unwrap_or_else(PoisonError::into_inner);
let endpoints = self.preserve_fabric_rekey(mode, endpoints, true)?;
self.publish_fleet_endpoints_locked(mode, &endpoints, build)
}

fn publish_fleet_endpoints_locked(
&self,
mode: &str,
endpoints: &[Value],
build: &str,
) -> Result<bool> {
let Some(member_key) = self.member_public_key() else {
return Ok(false);
Expand Down
Loading
Loading