diff --git a/core/archipelago/src/federation/storage.rs b/core/archipelago/src/federation/storage.rs index 8ff3a525..cd707df8 100644 --- a/core/archipelago/src/federation/storage.rs +++ b/core/archipelago/src/federation/storage.rs @@ -46,9 +46,36 @@ pub(crate) async fn ensure_dir(data_dir: &Path) -> Result { Ok(dir) } +/// Serializes every load-mutate-save cycle against `federation/nodes.json` +/// (and its paired `removed-nodes.json` tombstone file). The concrete +/// failure this prevents: a `federation.remove-node` RPC racing the 90s +/// auto-sync loop's `update_node_state` call. Without this lock, the sync +/// task's `load_nodes` snapshot — taken *before* the removal lands — could +/// finish its own save *after* the removal's save, silently re-writing the +/// peer the operator just removed back into the node list, with no error +/// logged anywhere (the "removed nodes reappear" symptom). +/// +/// Acquire with `.lock().await`, never `try_lock`. Unlike `update.rs`'s +/// `UPDATE_OP_LOCK` (which deliberately rejects a concurrent caller with an +/// "already running" error — the right UX for update downloads), +/// reject-on-contention is wrong here: a rejected federation write would +/// reproduce the very lost-write symptom this lock exists to close, instead +/// of fixing it. Federation writes are infrequent, so callers queueing +/// briefly behind `.lock().await` is the correct trade-off. +static FEDERATION_STORE_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + // ──────────────────────────── Node Management ──────────────────────────── pub async fn load_nodes(data_dir: &Path) -> Result> { + let _guard = FEDERATION_STORE_LOCK.lock().await; + load_nodes_inner(data_dir).await +} + +/// Lock-free body of `load_nodes`. Callers that already hold +/// `FEDERATION_STORE_LOCK` (i.e. other functions in this module composing a +/// multi-step critical section) must call this instead of `load_nodes` to +/// avoid self-deadlock — `tokio::sync::Mutex` is not re-entrant. +async fn load_nodes_inner(data_dir: &Path) -> Result> { let dir = data_dir.join(FEDERATION_DIR); let path = dir.join(NODES_FILE); if !path.exists() { @@ -147,19 +174,38 @@ pub async fn record_peer_transport( } pub async fn save_nodes(data_dir: &Path, nodes: &[FederatedNode]) -> Result<()> { + let _guard = FEDERATION_STORE_LOCK.lock().await; + save_nodes_inner(data_dir, nodes).await +} + +/// Lock-free body of `save_nodes`. See `load_nodes_inner` for why callers +/// that already hold `FEDERATION_STORE_LOCK` must use this instead. +/// +/// Writes atomically: serialize to a sibling `.tmp` file in the same +/// directory, then `rename` it onto the real path. Same-directory rename is +/// required — it's atomic on the same filesystem (a crash mid-write leaves +/// either the old complete file or the new complete file, never a partial +/// one); a cross-filesystem rename would not have this guarantee. +async fn save_nodes_inner(data_dir: &Path, nodes: &[FederatedNode]) -> Result<()> { let dir = ensure_dir(data_dir).await?; let file = NodesFile { nodes: nodes.to_vec(), }; let content = serde_json::to_string_pretty(&file).context("Failed to serialize nodes")?; - fs::write(dir.join(NODES_FILE), content) + let final_path = dir.join(NODES_FILE); + let tmp_path = dir.join(format!("{NODES_FILE}.tmp")); + fs::write(&tmp_path, content) + .await + .context("Failed to write federation nodes")?; + fs::rename(&tmp_path, &final_path) .await .context("Failed to write federation nodes")?; Ok(()) } pub async fn add_node(data_dir: &Path, node: FederatedNode) -> Result> { - let mut nodes = load_nodes(data_dir).await?; + let _guard = FEDERATION_STORE_LOCK.lock().await; + let mut nodes = load_nodes_inner(data_dir).await?; let exists = nodes.iter().any(|n| n.did == node.did); if exists { anyhow::bail!("Node with DID {} is already federated", node.did); @@ -169,16 +215,17 @@ pub async fn add_node(data_dir: &Path, node: FederatedNode) -> Result Result> { - let mut nodes = load_nodes(data_dir).await?; + let _guard = FEDERATION_STORE_LOCK.lock().await; + let mut nodes = load_nodes_inner(data_dir).await?; let before = nodes.len(); nodes.retain(|n| n.did != did); if nodes.len() == before { @@ -190,10 +237,10 @@ pub async fn remove_node(data_dir: &Path, did: &str) -> Result Result Result<()> { + let _guard = FEDERATION_STORE_LOCK.lock().await; + tombstone_did_inner(data_dir, did).await +} + +/// Lock-free body of `tombstone_did`. See `load_nodes_inner` for why +/// callers that already hold `FEDERATION_STORE_LOCK` must use this instead. +async fn tombstone_did_inner(data_dir: &Path, did: &str) -> Result<()> { let dir = ensure_dir(data_dir).await?; let path = dir.join(REMOVED_FILE); let mut file: RemovedFile = if path.exists() { @@ -234,7 +297,19 @@ pub async fn tombstone_did(data_dir: &Path, did: &str) -> Result<()> { } /// Clear a DID's tombstone (operator explicitly re-added it). +/// +/// `add_node` now calls `untombstone_did_inner` directly for the same +/// single-critical-section reason as `tombstone_did` above; see that +/// doc comment. +#[allow(dead_code)] pub async fn untombstone_did(data_dir: &Path, did: &str) -> Result<()> { + let _guard = FEDERATION_STORE_LOCK.lock().await; + untombstone_did_inner(data_dir, did).await +} + +/// Lock-free body of `untombstone_did`. See `load_nodes_inner` for why +/// callers that already hold `FEDERATION_STORE_LOCK` must use this instead. +async fn untombstone_did_inner(data_dir: &Path, did: &str) -> Result<()> { let path = data_dir.join(FEDERATION_DIR).join(REMOVED_FILE); if !path.exists() { return Ok(()); @@ -258,13 +333,14 @@ pub async fn set_trust_level( did: &str, trust: TrustLevel, ) -> Result> { - let mut nodes = load_nodes(data_dir).await?; + let _guard = FEDERATION_STORE_LOCK.lock().await; + let mut nodes = load_nodes_inner(data_dir).await?; let node = nodes .iter_mut() .find(|n| n.did == did) .ok_or_else(|| anyhow::anyhow!("No federated node with DID {}", did))?; node.trust_level = trust; - save_nodes(data_dir, &nodes).await?; + save_nodes_inner(data_dir, &nodes).await?; Ok(nodes) } @@ -290,7 +366,8 @@ pub async fn update_node(data_dir: &Path, updated: &FederatedNode) -> Result<()> } pub async fn update_node_state(data_dir: &Path, did: &str, state: NodeStateSnapshot) -> Result<()> { - let mut nodes = load_nodes(data_dir).await?; + let _guard = FEDERATION_STORE_LOCK.lock().await; + let mut nodes = load_nodes_inner(data_dir).await?; if let Some(node) = nodes.iter_mut().find(|n| n.did == did) { node.last_seen = Some(state.timestamp.clone()); // Update node name from sync if provided (peer announced their name) @@ -309,7 +386,7 @@ pub async fn update_node_state(data_dir: &Path, did: &str, state: NodeStateSnaps } } node.last_state = Some(state); - save_nodes(data_dir, &nodes).await?; + save_nodes_inner(data_dir, &nodes).await?; } Ok(()) } @@ -490,6 +567,151 @@ mod tests { assert_eq!(nodes[0].trust_level, TrustLevel::Observer); } + /// The .198 v1.7.103 update-bricking race (see `update.rs`'s + /// `UPDATE_OP_LOCK`) had the same shape as this test: two concurrent + /// mutators sharing one on-disk file with no coordination. Here, + /// `add_node` and `set_trust_level` both do their own load-mutate-save + /// cycle against `nodes.json`; without `FEDERATION_STORE_LOCK` held for + /// the whole cycle, whichever writer's `save_nodes` lands second wins + /// and clobbers the other writer's update entirely. + /// + /// Uses real `tokio::spawn` tasks (not just `tokio::join!` polled within + /// one task) so the two writers are genuinely scheduled across the + /// multi-thread runtime's worker pool, and loops so OS scheduling + /// jitter gets many chances to interleave the two load-mutate-save + /// cycles. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_concurrent_writes_do_not_lose_updates() { + for i in 0..40 { + let dir = tempfile::tempdir().unwrap(); + add_node(dir.path(), make_node("did:key:zA", "a.onion")) + .await + .unwrap(); + + let dir_a = dir.path().to_path_buf(); + let dir_b = dir.path().to_path_buf(); + let add_task = tokio::spawn(async move { + add_node(&dir_a, make_node("did:key:zB", "b.onion")).await + }); + let trust_task = tokio::spawn(async move { + set_trust_level(&dir_b, "did:key:zA", TrustLevel::Observer).await + }); + add_task.await.unwrap().unwrap(); + trust_task.await.unwrap().unwrap(); + + let nodes = load_nodes(dir.path()).await.unwrap(); + assert_eq!( + nodes.len(), + 2, + "iteration {i}: both concurrent writes must persist — neither may be lost" + ); + let node_a = nodes + .iter() + .find(|n| n.did == "did:key:zA") + .unwrap_or_else(|| panic!("iteration {i}: did:key:zA must still be present")); + assert_eq!( + node_a.trust_level, + TrustLevel::Observer, + "iteration {i}: the concurrent set_trust_level write must not be lost" + ); + } + } + + /// Reproduces the FED-01 "removed nodes reappear" symptom: a + /// `federation.remove-node` RPC racing the 90s auto-sync loop's + /// `update_node_state` call. The sync task's `load_nodes` snapshot, + /// taken before the removal lands, must never be allowed to re-save + /// the peer the operator just removed. + /// + /// `remove_node`'s critical path does one more disk hop than + /// `update_node_state` (the tombstone write), which structurally + /// biases a single 1-vs-1 race toward the *safe* ordering (remove's + /// save landing last). To reliably exercise the unsafe ordering this + /// test races `remove_node` against a BURST of concurrent + /// `update_node_state` calls, each a genuine `tokio::spawn`ed task, so + /// OS scheduling jitter has many independent chances per iteration to + /// land at least one sync save after the removal's save. Looped so a + /// single unlucky iteration isn't required to catch it. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn test_remove_survives_concurrent_state_sync() { + const SYNC_RACERS: usize = 12; + for i in 0..30 { + let dir = tempfile::tempdir().unwrap(); + add_node(dir.path(), make_node("did:key:zA", "a.onion")) + .await + .unwrap(); + add_node(dir.path(), make_node("did:key:zB", "b.onion")) + .await + .unwrap(); + + let mut sync_tasks = Vec::with_capacity(SYNC_RACERS); + for _ in 0..SYNC_RACERS { + let dir_sync = dir.path().to_path_buf(); + let state = NodeStateSnapshot { + timestamp: "2026-03-10T12:00:00Z".to_string(), + node_name: None, + apps: Vec::new(), + cpu_usage_percent: None, + mem_used_bytes: None, + mem_total_bytes: None, + disk_used_bytes: None, + disk_total_bytes: None, + uptime_secs: None, + tor_active: None, + nostr_npub: None, + own_fips_npub: None, + federated_peers: Vec::new(), + lat: None, + lon: None, + }; + sync_tasks.push(tokio::spawn(async move { + update_node_state(&dir_sync, "did:key:zA", state).await + })); + } + let dir_remove = dir.path().to_path_buf(); + let remove_task = + tokio::spawn(async move { remove_node(&dir_remove, "did:key:zA").await }); + + for task in sync_tasks { + task.await.unwrap().unwrap(); + } + remove_task.await.unwrap().unwrap(); + + let nodes = load_nodes(dir.path()).await.unwrap(); + assert!( + !nodes.iter().any(|n| n.did == "did:key:zA"), + "iteration {i}: removed node reappeared after a concurrent sync" + ); + let removed = load_removed_dids(dir.path()).await.unwrap(); + assert!( + removed.contains("did:key:zA"), + "iteration {i}: removed DID must remain tombstoned" + ); + } + } + + /// FED-01 empty edge: removing the last federated node must succeed + /// cleanly, not error out on an "empty list" special case. + #[tokio::test] + async fn test_remove_last_node_leaves_empty_list() { + let dir = tempfile::tempdir().unwrap(); + add_node(dir.path(), make_node("did:key:zOnly", "only.onion")) + .await + .unwrap(); + + let result = remove_node(dir.path(), "did:key:zOnly").await.unwrap(); + assert!(result.is_empty(), "returned Vec must be empty"); + + let nodes = load_nodes(dir.path()).await.unwrap(); + assert!(nodes.is_empty(), "load_nodes must return an empty Vec, not an error"); + + let removed = load_removed_dids(dir.path()).await.unwrap(); + assert!( + removed.contains("did:key:zOnly"), + "the sole removed node must still be tombstoned" + ); + } + #[tokio::test] async fn test_update_node_state() { let dir = tempfile::tempdir().unwrap();