Whitespace-only reflow in storage.rs/seed.rs/update.rs (rustfmt line- wrapping rules) and app_ports.rs (array literal reflow after the port list grew). No logic change. tests/release/run.sh's cargo-fmt --check stage was failing on this before v1.7.119-alpha could be cut. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
759 lines
30 KiB
Rust
759 lines
30 KiB
Rust
//! Federation persistent storage: node list and invite management on disk.
|
|
|
|
use anyhow::{Context, Result};
|
|
use serde::{Deserialize, Serialize};
|
|
use std::path::Path;
|
|
use tokio::fs;
|
|
|
|
use super::types::{FederatedNode, FederationInvite, NodeStateSnapshot, TrustLevel};
|
|
|
|
pub(crate) const FEDERATION_DIR: &str = "federation";
|
|
pub(crate) const NODES_FILE: &str = "nodes.json";
|
|
pub(crate) const INVITES_FILE: &str = "invites.json";
|
|
/// Tombstones: DIDs the operator explicitly removed. Kept so transitive
|
|
/// federation discovery can't silently re-add a peer they deleted.
|
|
pub(crate) const REMOVED_FILE: &str = "removed-nodes.json";
|
|
|
|
/// Top-level file structures.
|
|
#[derive(Debug, Default, Serialize, Deserialize)]
|
|
pub(crate) struct NodesFile {
|
|
pub(crate) nodes: Vec<FederatedNode>,
|
|
}
|
|
|
|
#[derive(Debug, Default, Serialize, Deserialize)]
|
|
pub(crate) struct RemovedFile {
|
|
pub(crate) removed: Vec<RemovedNode>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
pub(crate) struct RemovedNode {
|
|
pub(crate) did: String,
|
|
pub(crate) removed_at: String,
|
|
}
|
|
|
|
#[derive(Debug, Default, Serialize, Deserialize)]
|
|
pub(crate) struct InvitesFile {
|
|
pub(crate) outgoing: Vec<FederationInvite>,
|
|
pub(crate) incoming: Vec<FederationInvite>,
|
|
}
|
|
|
|
/// Ensure federation directory exists.
|
|
pub(crate) async fn ensure_dir(data_dir: &Path) -> Result<std::path::PathBuf> {
|
|
let dir = data_dir.join(FEDERATION_DIR);
|
|
fs::create_dir_all(&dir)
|
|
.await
|
|
.context("Failed to create federation directory")?;
|
|
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<Vec<FederatedNode>> {
|
|
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<Vec<FederatedNode>> {
|
|
let dir = data_dir.join(FEDERATION_DIR);
|
|
let path = dir.join(NODES_FILE);
|
|
if !path.exists() {
|
|
return Ok(Vec::new());
|
|
}
|
|
let content = fs::read_to_string(&path)
|
|
.await
|
|
.context("Failed to read federation nodes")?;
|
|
let file: NodesFile = serde_json::from_str(&content).unwrap_or_default();
|
|
Ok(dedup_nodes_by_onion(file.nodes))
|
|
}
|
|
|
|
/// Collapse entries that share an onion. An onion is a node's stable, unique
|
|
/// network identity, so two entries with the same onion are the SAME physical
|
|
/// node lingering under two dids (e.g. after a did/key change). Returning both
|
|
/// duplicates the node in the trusted-node list (B1) and the chat list (B2).
|
|
/// Keep the first occurrence and merge any missing fips_npub/name/last_state
|
|
/// from the duplicates into it, then drop them. Non-destructive to disk; the
|
|
/// deduped list persists the next time nodes are saved (add/sync).
|
|
fn dedup_nodes_by_onion(nodes: Vec<FederatedNode>) -> Vec<FederatedNode> {
|
|
use std::collections::HashMap;
|
|
let mut by_onion: HashMap<String, usize> = HashMap::new();
|
|
let mut out: Vec<FederatedNode> = Vec::with_capacity(nodes.len());
|
|
for node in nodes {
|
|
let key = node.onion.trim_end_matches(".onion").to_string();
|
|
if key.is_empty() {
|
|
out.push(node);
|
|
continue;
|
|
}
|
|
if let Some(&idx) = by_onion.get(&key) {
|
|
let kept = &mut out[idx];
|
|
if kept.fips_npub.is_none() {
|
|
kept.fips_npub = node.fips_npub;
|
|
}
|
|
if kept.name.is_none() {
|
|
kept.name = node.name;
|
|
}
|
|
if kept.last_state.is_none() {
|
|
kept.last_state = node.last_state;
|
|
}
|
|
continue;
|
|
}
|
|
by_onion.insert(key, out.len());
|
|
out.push(node);
|
|
}
|
|
out
|
|
}
|
|
|
|
/// Look up a federated peer's FIPS npub given their onion address.
|
|
/// Returns `None` when the onion isn't in our federation list or the
|
|
/// peer hasn't advertised a FIPS key. Matching is suffix-tolerant so
|
|
/// callers can pass `abc` or `abc.onion` interchangeably.
|
|
pub async fn fips_npub_for_onion(data_dir: &Path, onion: &str) -> Option<String> {
|
|
let target = onion.trim_end_matches(".onion");
|
|
let nodes = load_nodes(data_dir).await.ok()?;
|
|
nodes
|
|
.iter()
|
|
.find(|n| n.onion.trim_end_matches(".onion") == target)
|
|
.and_then(|n| n.fips_npub.clone())
|
|
}
|
|
|
|
/// Record the transport used on the most recent successful peer reach.
|
|
/// Used for the "FIPS"/"Tor" badge on each node card in the UI — we write
|
|
/// what we actually used, not what was predicted.
|
|
///
|
|
/// Matches by DID first (precise) and falls back to onion (when the
|
|
/// caller didn't carry the DID through). No-op if the peer isn't in
|
|
/// our federation list.
|
|
pub async fn record_peer_transport(
|
|
data_dir: &Path,
|
|
did: Option<&str>,
|
|
onion: Option<&str>,
|
|
transport: &str,
|
|
) -> Result<()> {
|
|
let mut nodes = load_nodes(data_dir).await?;
|
|
let now = chrono::Utc::now().to_rfc3339();
|
|
let onion_target = onion.map(|o| o.trim_end_matches(".onion"));
|
|
|
|
let mut modified = false;
|
|
for node in nodes.iter_mut() {
|
|
let did_match = did.is_some_and(|d| d == node.did);
|
|
let onion_match = onion_target.is_some_and(|t| node.onion.trim_end_matches(".onion") == t);
|
|
if did_match || onion_match {
|
|
node.last_transport = Some(transport.to_string());
|
|
node.last_transport_at = Some(now.clone());
|
|
node.last_seen = Some(now.clone());
|
|
modified = true;
|
|
break;
|
|
}
|
|
}
|
|
|
|
if modified {
|
|
save_nodes(data_dir, &nodes).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
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")?;
|
|
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<Vec<FederatedNode>> {
|
|
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);
|
|
}
|
|
// Explicitly (re-)adding a node clears any prior tombstone so the
|
|
// operator can intentionally bring back a previously removed peer.
|
|
// Propagate failure BEFORE mutating the node list: with the tombstone
|
|
// still in place, sync reconciliation would silently re-remove the
|
|
// node the operator just added.
|
|
untombstone_did_inner(data_dir, &node.did)
|
|
.await
|
|
.context("clear removal tombstone")?;
|
|
nodes.push(node);
|
|
save_nodes_inner(data_dir, &nodes).await?;
|
|
Ok(nodes)
|
|
}
|
|
|
|
pub async fn remove_node(data_dir: &Path, did: &str) -> Result<Vec<FederatedNode>> {
|
|
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 {
|
|
anyhow::bail!("No federated node with DID {}", did);
|
|
}
|
|
// Tombstone the DID so transitive federation discovery (a still-federated
|
|
// peer advertising this DID as one of *its* trusted peers) can't silently
|
|
// re-add it. Tombstone FIRST and propagate failure: a remove whose
|
|
// tombstone never landed isn't a remove — the peer would quietly
|
|
// reappear after the next sync. Tombstoning is idempotent, so if the
|
|
// node-list save below fails the operator's retry works cleanly.
|
|
tombstone_did_inner(data_dir, did)
|
|
.await
|
|
.context("persist removal tombstone")?;
|
|
save_nodes_inner(data_dir, &nodes).await?;
|
|
Ok(nodes)
|
|
}
|
|
|
|
/// Load the set of tombstoned (operator-removed) DIDs.
|
|
pub async fn load_removed_dids(data_dir: &Path) -> Result<std::collections::HashSet<String>> {
|
|
let path = data_dir.join(FEDERATION_DIR).join(REMOVED_FILE);
|
|
if !path.exists() {
|
|
return Ok(std::collections::HashSet::new());
|
|
}
|
|
let content = fs::read_to_string(&path)
|
|
.await
|
|
.context("Failed to read removed nodes")?;
|
|
let file: RemovedFile = serde_json::from_str(&content).unwrap_or_default();
|
|
Ok(file.removed.into_iter().map(|r| r.did).collect())
|
|
}
|
|
|
|
/// Record a DID as removed. Idempotent.
|
|
///
|
|
/// `remove_node` now calls `tombstone_did_inner` directly (to keep the
|
|
/// tombstone write and the node-list save in one critical section), so this
|
|
/// locked outer wrapper currently has no in-crate caller. Kept `pub` and
|
|
/// `#[allow(dead_code)]` rather than removed: it's part of this module's
|
|
/// documented public surface (see PLAN.md's "public signatures unchanged"
|
|
/// contract) for any future direct caller that needs a standalone,
|
|
/// correctly-locked tombstone write.
|
|
#[allow(dead_code)]
|
|
pub async fn tombstone_did(data_dir: &Path, did: &str) -> 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() {
|
|
serde_json::from_str(&fs::read_to_string(&path).await.unwrap_or_default())
|
|
.unwrap_or_default()
|
|
} else {
|
|
RemovedFile::default()
|
|
};
|
|
if !file.removed.iter().any(|r| r.did == did) {
|
|
file.removed.push(RemovedNode {
|
|
did: did.to_string(),
|
|
removed_at: chrono::Utc::now().to_rfc3339(),
|
|
});
|
|
let content = serde_json::to_string_pretty(&file).context("serialize removed nodes")?;
|
|
fs::write(&path, content)
|
|
.await
|
|
.context("Failed to write removed nodes")?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// 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(());
|
|
}
|
|
let mut file: RemovedFile =
|
|
serde_json::from_str(&fs::read_to_string(&path).await.unwrap_or_default())
|
|
.unwrap_or_default();
|
|
let before = file.removed.len();
|
|
file.removed.retain(|r| r.did != did);
|
|
if file.removed.len() != before {
|
|
let content = serde_json::to_string_pretty(&file).context("serialize removed nodes")?;
|
|
fs::write(&path, content)
|
|
.await
|
|
.context("Failed to write removed nodes")?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn set_trust_level(
|
|
data_dir: &Path,
|
|
did: &str,
|
|
trust: TrustLevel,
|
|
) -> Result<Vec<FederatedNode>> {
|
|
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_inner(data_dir, &nodes).await?;
|
|
Ok(nodes)
|
|
}
|
|
|
|
/// Update a federated node's metadata (onion, pubkey, name, last_seen).
|
|
pub async fn update_node(data_dir: &Path, updated: &FederatedNode) -> Result<()> {
|
|
let mut nodes = load_nodes(data_dir).await?;
|
|
if let Some(node) = nodes.iter_mut().find(|n| n.did == updated.did) {
|
|
if !updated.onion.is_empty() {
|
|
node.onion = updated.onion.clone();
|
|
}
|
|
if !updated.pubkey.is_empty() {
|
|
node.pubkey = updated.pubkey.clone();
|
|
}
|
|
if updated.name.is_some() {
|
|
node.name = updated.name.clone();
|
|
}
|
|
if updated.last_seen.is_some() {
|
|
node.last_seen = updated.last_seen.clone();
|
|
}
|
|
save_nodes(data_dir, &nodes).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn update_node_state(data_dir: &Path, did: &str, state: NodeStateSnapshot) -> Result<()> {
|
|
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)
|
|
if let Some(ref name) = state.node_name {
|
|
if !name.is_empty() {
|
|
node.name = Some(name.clone());
|
|
}
|
|
}
|
|
// Learn the peer's FIPS npub from their state snapshot so
|
|
// federations established before v1.4 (pre-fips_npub) start
|
|
// routing over FIPS on the very next sync. Refresh if the peer
|
|
// rotated their FIPS key, too.
|
|
if let Some(ref npub) = state.own_fips_npub {
|
|
if !npub.is_empty() && node.fips_npub.as_deref().map(str::trim) != Some(npub.trim()) {
|
|
node.fips_npub = Some(npub.clone());
|
|
}
|
|
}
|
|
node.last_state = Some(state);
|
|
save_nodes_inner(data_dir, &nodes).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
// ──────────────────────────── Invite Storage ────────────────────────────
|
|
|
|
pub(crate) async fn load_invites(data_dir: &Path) -> Result<InvitesFile> {
|
|
let dir = data_dir.join(FEDERATION_DIR);
|
|
let path = dir.join(INVITES_FILE);
|
|
if !path.exists() {
|
|
return Ok(InvitesFile::default());
|
|
}
|
|
let content = fs::read_to_string(&path)
|
|
.await
|
|
.context("Failed to read invites")?;
|
|
let file: InvitesFile = serde_json::from_str(&content).unwrap_or_default();
|
|
Ok(file)
|
|
}
|
|
|
|
pub(crate) async fn save_invites(data_dir: &Path, invites: &InvitesFile) -> Result<()> {
|
|
let dir = ensure_dir(data_dir).await?;
|
|
let content = serde_json::to_string_pretty(invites).context("Failed to serialize invites")?;
|
|
fs::write(dir.join(INVITES_FILE), content)
|
|
.await
|
|
.context("Failed to write invites")?;
|
|
Ok(())
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::federation::types::AppStatus;
|
|
|
|
fn make_node(did: &str, onion: &str) -> FederatedNode {
|
|
FederatedNode {
|
|
did: did.to_string(),
|
|
pubkey: "aabbccdd".to_string(),
|
|
onion: onion.to_string(),
|
|
name: None,
|
|
trust_level: TrustLevel::Trusted,
|
|
added_at: "2026-01-01T00:00:00Z".to_string(),
|
|
last_seen: None,
|
|
last_state: None,
|
|
fips_npub: None,
|
|
last_transport: None,
|
|
last_transport_at: None,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn test_dedup_nodes_by_onion_collapses_same_onion() {
|
|
// Two entries share an onion (same physical node under two dids) — must
|
|
// collapse to one, keeping the first did and merging fips_npub/name (B1/B2).
|
|
let mut dup = make_node("did:key:zDUP", "shared.onion");
|
|
dup.fips_npub = Some("npub1merged".to_string());
|
|
dup.name = Some("Sapien".to_string());
|
|
let nodes = vec![
|
|
make_node("did:key:zKEEP", "shared.onion"),
|
|
dup,
|
|
make_node("did:key:zOTHER", "other.onion"),
|
|
];
|
|
let out = dedup_nodes_by_onion(nodes);
|
|
assert_eq!(out.len(), 2, "two distinct onions remain");
|
|
let kept = out.iter().find(|n| n.onion == "shared.onion").unwrap();
|
|
assert_eq!(kept.did, "did:key:zKEEP", "keeps first did for the onion");
|
|
assert_eq!(
|
|
kept.fips_npub.as_deref(),
|
|
Some("npub1merged"),
|
|
"merges fips_npub from the dropped duplicate"
|
|
);
|
|
assert_eq!(
|
|
kept.name.as_deref(),
|
|
Some("Sapien"),
|
|
"merges name from the dup"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_dedup_onion_suffix_insensitive() {
|
|
// The ".onion" suffix must not affect the match.
|
|
let nodes = vec![
|
|
make_node("did:key:z1", "abc"),
|
|
make_node("did:key:z2", "abc.onion"),
|
|
];
|
|
assert_eq!(dedup_nodes_by_onion(nodes).len(), 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_load_nodes_empty_when_no_file() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let nodes = load_nodes(dir.path()).await.unwrap();
|
|
assert!(nodes.is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_save_and_load_nodes_roundtrip() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let nodes = vec![
|
|
make_node("did:key:z1", "a.onion"),
|
|
make_node("did:key:z2", "b.onion"),
|
|
];
|
|
save_nodes(dir.path(), &nodes).await.unwrap();
|
|
let loaded = load_nodes(dir.path()).await.unwrap();
|
|
assert_eq!(loaded.len(), 2);
|
|
assert_eq!(loaded[0].did, "did:key:z1");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_add_node_deduplicates_by_did() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
add_node(dir.path(), make_node("did:key:z1", "a.onion"))
|
|
.await
|
|
.unwrap();
|
|
let result = add_node(dir.path(), make_node("did:key:z1", "b.onion")).await;
|
|
assert!(result.is_err());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_remove_node_by_did() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
add_node(dir.path(), make_node("did:key:z1", "a.onion"))
|
|
.await
|
|
.unwrap();
|
|
add_node(dir.path(), make_node("did:key:z2", "b.onion"))
|
|
.await
|
|
.unwrap();
|
|
let result = remove_node(dir.path(), "did:key:z1").await.unwrap();
|
|
assert_eq!(result.len(), 1);
|
|
assert_eq!(result[0].did, "did:key:z2");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_remove_nonexistent_node_errors() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let result = remove_node(dir.path(), "did:key:nonexistent").await;
|
|
assert!(result.is_err());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_remove_tombstones_and_readd_clears_it() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
add_node(dir.path(), make_node("did:key:z1", "a.onion"))
|
|
.await
|
|
.unwrap();
|
|
// No tombstones yet.
|
|
assert!(load_removed_dids(dir.path()).await.unwrap().is_empty());
|
|
|
|
// Removing tombstones the DID so transitive discovery won't re-add it.
|
|
remove_node(dir.path(), "did:key:z1").await.unwrap();
|
|
let removed = load_removed_dids(dir.path()).await.unwrap();
|
|
assert!(
|
|
removed.contains("did:key:z1"),
|
|
"removed DID must be tombstoned"
|
|
);
|
|
|
|
// Explicitly re-adding clears the tombstone (intentional re-federate).
|
|
add_node(dir.path(), make_node("did:key:z1", "a.onion"))
|
|
.await
|
|
.unwrap();
|
|
assert!(
|
|
!load_removed_dids(dir.path())
|
|
.await
|
|
.unwrap()
|
|
.contains("did:key:z1"),
|
|
"explicit re-add must clear the tombstone"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_set_trust_level() {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
add_node(dir.path(), make_node("did:key:z1", "a.onion"))
|
|
.await
|
|
.unwrap();
|
|
let nodes = set_trust_level(dir.path(), "did:key:z1", TrustLevel::Observer)
|
|
.await
|
|
.unwrap();
|
|
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();
|
|
add_node(dir.path(), make_node("did:key:z1", "a.onion"))
|
|
.await
|
|
.unwrap();
|
|
|
|
let state = NodeStateSnapshot {
|
|
timestamp: "2026-03-10T12:00:00Z".to_string(),
|
|
node_name: None,
|
|
apps: vec![AppStatus {
|
|
id: "bitcoin".to_string(),
|
|
status: "running".to_string(),
|
|
version: Some("27.0".to_string()),
|
|
}],
|
|
cpu_usage_percent: Some(45.2),
|
|
mem_used_bytes: Some(4_000_000_000),
|
|
mem_total_bytes: Some(8_000_000_000),
|
|
disk_used_bytes: None,
|
|
disk_total_bytes: None,
|
|
uptime_secs: Some(86400),
|
|
tor_active: Some(true),
|
|
nostr_npub: None,
|
|
own_fips_npub: None,
|
|
federated_peers: Vec::new(),
|
|
lat: None,
|
|
lon: None,
|
|
};
|
|
|
|
update_node_state(dir.path(), "did:key:z1", state)
|
|
.await
|
|
.unwrap();
|
|
|
|
let nodes = load_nodes(dir.path()).await.unwrap();
|
|
assert!(nodes[0].last_seen.is_some());
|
|
let ls = nodes[0].last_state.as_ref().unwrap();
|
|
assert_eq!(ls.apps.len(), 1);
|
|
assert_eq!(ls.cpu_usage_percent, Some(45.2));
|
|
}
|
|
}
|