lfg2025/archy
1
0
Fork 0
Code Issues 23 Pull Requests 1 Actions Packages Projects Releases 63 Wiki Activity

fix(01-01): serialize federation node-store writes, close remove-vs-sync race (FED-01)

Browse Source 
- Add FEDERATION_STORE_LOCK (tokio::sync::Mutex<()>) guarding every
  load-mutate-save cycle in federation/storage.rs, closing the race where
  the 90s auto-sync loop's stale pre-removal snapshot could silently
  re-save a peer the operator just removed (no error logged anywhere).
- Split load_nodes/save_nodes/tombstone_did/untombstone_did into thin
  locked outer wrappers + lock-free *_inner bodies so remove_node and
  add_node can hold the guard across their whole tombstone+save critical
  section without self-deadlocking (Mutex is not re-entrant).
- Route load_nodes, save_nodes, remove_node, add_node, tombstone_did,
  untombstone_did, set_trust_level, and update_node_state through the
  lock (set_trust_level pulled forward from Task 2's scope — required for
  test_concurrent_writes_do_not_lose_updates, part of Task 1's own
  required-green test suite, to pass; documented in SUMMARY).
- Convert save_nodes_inner to an atomic write: serialize to a sibling
  nodes.json.tmp in the same directory, then tokio::fs::rename onto the
  real path, so a crash mid-write never leaves a partially-written
  nodes.json for a concurrent reader.
- Add 3 new regression tests, two of which are fail-first proven: heavy
  tokio::spawn-based concurrency (not just tokio::join!, since
  remove_node's extra tombstone I/O hop structurally biased a simple
  2-task race toward the safe ordering) reliably reproduced both the lost
  concurrent write and the removed-node-reappears bug pre-fix; both are
  green post-fix along with the existing suite (13/13).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
archipelago 2026-07-30 04:04:08 -04:00
parent 2d9625ca74
commit 2f99db5e6b
1 changed files with 233 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

244
core/archipelago/src/federation/storage.rs
View File

@ -46,9 +46,36 @@ pub(crate) async fn ensure_dir(data_dir: &Path) -> Result<std::path::PathBuf> {
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() {
@ -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<Vec<FederatedNode>> {
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<Vec<Federa
// 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(data_dir, &node.did)
untombstone_did_inner(data_dir, &node.did)
.await
.context("clear removal tombstone")?;
nodes.push(node);
save_nodes(data_dir, &nodes).await?;
save_nodes_inner(data_dir, &nodes).await?;
Ok(nodes)
}
pub async fn remove_node(data_dir: &Path, did: &str) -> Result<Vec<FederatedNode>> {
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<Vec<FederatedNode
// 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(data_dir, did)
tombstone_did_inner(data_dir, did)
.await
.context("persist removal tombstone")?;
save_nodes(data_dir, &nodes).await?;
save_nodes_inner(data_dir, &nodes).await?;
Ok(nodes)
}
@ -211,7 +258,23 @@ pub async fn load_removed_dids(data_dir: &Path) -> Result<std::collections::Hash
}
/// 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() {
@ -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<Vec<FederatedNode>> {
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();