From 2b5f435276417d1b389a7473549f5c67236d17dd Mon Sep 17 00:00:00 2001 From: contractless <00.pur.ple@gmail.com> Date: Sat, 12 Sep 2026 12:41:01 -0600 Subject: [PATCH] mapping order fix --- src/bin/lookup_online_node_count.rs | 16 +- src/records/memory/connections.rs | 8 +- src/records/memory/network_mapping/add.rs | 254 ++++++++++++++---- src/records/memory/network_mapping/mod.rs | 4 + src/records/memory/network_mapping/monitor.rs | 82 ++---- src/records/memory/network_mapping/queries.rs | 61 ++--- 6 files changed, 263 insertions(+), 162 deletions(-) diff --git a/src/bin/lookup_online_node_count.rs b/src/bin/lookup_online_node_count.rs index 7d2c828..c19147b 100644 --- a/src/bin/lookup_online_node_count.rs +++ b/src/bin/lookup_online_node_count.rs @@ -2,7 +2,7 @@ use contractless::common::network_startup::get_connections; use contractless::env; use contractless::records::memory::network_mapping::structs::{ NETWORK_SNAPSHOT_HEADER_BYTES, NETWORK_SNAPSHOT_MAGIC, NETWORK_SNAPSHOT_VERSION, - NODE_DELETED_TIMESTAMP_OFFSET, NODE_MONITOR_COUNT_OFFSET, NODE_RECORD_FIXED_BYTES, + NODE_ADDED_TIMESTAMP_OFFSET, NODE_DELETED_TIMESTAMP_OFFSET, NODE_MONITOR_COUNT_OFFSET, NODE_RECORD_FIXED_BYTES, }; use contractless::records::memory::response_channels::generate_uid; use contractless::standalone_tools::connections::handshake; @@ -70,7 +70,11 @@ fn online_node_count(response: &[u8]) -> Result { return Err("network mapping ended inside a node monitor list".to_string()); } - if deleted_timestamp == 0 && monitor_count > 0 { + let added_timestamp = u64::from_le_bytes( + response[offset + NODE_ADDED_TIMESTAMP_OFFSET..offset + NODE_ADDED_TIMESTAMP_OFFSET + 8] + .try_into().map_err(|_| "node add timestamp was invalid")?, + ); + if (deleted_timestamp == 0 || added_timestamp > deleted_timestamp) && monitor_count > 0 { online += 1; } offset = record_end; @@ -165,6 +169,14 @@ mod tests { assert_eq!(online_node_count(&response).unwrap(), 1); } + #[test] + fn counts_reconnected_node_with_retained_deletion_boundary() { + let mut record = snapshot_record(51, 1); + record[NODE_ADDED_TIMESTAMP_OFFSET..NODE_ADDED_TIMESTAMP_OFFSET + 8] + .copy_from_slice(&52u64.to_le_bytes()); + assert_eq!(online_node_count(&snapshot(&[record])).unwrap(), 1); + } + #[test] fn rejects_truncated_monitor_lists() { let mut response = snapshot(&[snapshot_record(0, 1)]); diff --git a/src/records/memory/connections.rs b/src/records/memory/connections.rs index 36917d9..384f8ce 100644 --- a/src/records/memory/connections.rs +++ b/src/records/memory/connections.rs @@ -1086,12 +1086,12 @@ fn spawn_monitor_activation( } tokio::spawn(async move { - // During reciprocal startup both peers can hold freshly sponsored - // pending memberships. Permit either side to author its monitor-add so - // the two activation events cannot deadlock waiting on one another. + // Accepted reconnects are active by timestamp ordering. + // Either side may author its monitor-add after membership acceptance so + // connection setup does not wait on reciprocal monitor activation. let mapping_prerequisites = timeout(Duration::from_secs(30), async { while !NodeInfo::contains_address(&monitored_address).await - || !NodeInfo::is_active_or_sponsored_pending(&monitoring_address).await + || !NodeInfo::is_active_address(&monitoring_address).await { sleep(Duration::from_millis(100)).await; } diff --git a/src/records/memory/network_mapping/add.rs b/src/records/memory/network_mapping/add.rs index 057a045..85cf129 100644 --- a/src/records/memory/network_mapping/add.rs +++ b/src/records/memory/network_mapping/add.rs @@ -49,7 +49,7 @@ impl NodeInfo { .filter(|(address, node)| { address.as_str() != excluded_address && operational_addresses.contains(address.as_str()) - && node.deleted_timestamp == 0 + && node.is_active() && node.blocks_mined >= 100 }) .collect(); @@ -81,12 +81,7 @@ impl NodeInfo { let mut active_ips = HashMap::::new(); for (address, node) in replacement { - if node.deleted_timestamp > 0 { - if node.deleted_timestamp < node.added_timestamp { - return Err(format!( - "snapshot deletion for {address} predates its membership" - )); - } + if !node.is_active() { if !node.monitoring.is_empty() { return Err(format!( "snapshot deleted node {address} still contained monitors" @@ -121,7 +116,7 @@ impl NodeInfo { let monitor_node = replacement.get(monitor).ok_or_else(|| { format!("snapshot node {address} referenced unknown monitor {monitor}") })?; - if monitor_node.deleted_timestamp > 0 { + if !monitor_node.is_active() { return Err(format!( "snapshot node {address} referenced deleted monitor {monitor}" )); @@ -196,12 +191,8 @@ impl NodeInfo { node.added_timestamp = edit.modified_timestamp; node.added_signature = edit.modified_signature.clone(); - // Keep the new lifecycle inactive until a newer signed monitor-add - // proves that connection setup reached the operational state. - node.deleted_timestamp = edit.modified_timestamp; - // Zero is the deterministic pending marker: no historical block has - // yet activated this lifecycle, and every peer derives the same map. - node.deleted_block = 0; + // Preserve the previous deletion boundary. The newer add makes this + // membership active; connection readiness still waits for setup. node.monitoring.clear(); accepted_ip_change } @@ -226,7 +217,7 @@ impl NodeInfo { let before = node.monitoring.len(); node.monitoring.retain(|monitor| monitor != &address); if target != winner_address - && node.deleted_timestamp == 0 + && node.is_active() && before != node.monitoring.len() && node.monitoring.is_empty() { @@ -463,10 +454,10 @@ impl NodeInfo { let candidate_lost_ip_claim = address_map .get(&address) .map(|candidate| { - candidate.deleted_timestamp > 0 + !candidate.is_active() && address_map.iter().any(|(other_address, other)| { other_address != &address - && other.deleted_timestamp == 0 + && other.is_active() && other.ip == candidate.ip }) }) @@ -516,11 +507,17 @@ impl NodeInfo { } let mut address_map = ADDRESS_MAP.lock().await; + if let Some(existing) = address_map.get(&edit.address) { + if existing.is_active() && existing.deleted_timestamp > 0 + && edit.modified_timestamp <= existing.deleted_timestamp { + return Err("reconnection record did not follow deletion".to_string()); + } + } let conflicting_claim = address_map .iter() .find(|(address, node)| { address.as_str() != edit.address - && node.deleted_timestamp == 0 + && node.is_active() && node.ip == edit.ip && edit.ip != GENESIS_IP }) @@ -572,7 +569,7 @@ impl NodeInfo { } if let Some(existing_node) = address_map.get_mut(&edit.address) { - if existing_node.deleted_timestamp > 0 { + if !existing_node.is_active() { if existing_node.ip == edit.ip && existing_node.port == edit.port && existing_node.added_by == edit.modified_by @@ -632,7 +629,7 @@ impl NodeInfo { } if let Some(node) = address_map.values().find(|node| node.ip == edit.ip) { - if node.deleted_timestamp == 0 && edit.ip != GENESIS_IP { + if node.is_active() && edit.ip != GENESIS_IP { return Err("ip already exists".to_string()); } } @@ -680,15 +677,35 @@ impl NodeInfo { ) -> Result<(), String> { Self::verify_synced_membership(db, &state.edit).await?; - if state.deleted_timestamp == 0 { - return Self::import_signed_mapping_address(db, state.edit, 0).await; + if state.deleted_timestamp == 0 || state.edit.modified_timestamp > state.deleted_timestamp { + let was_missing = !ADDRESS_MAP.lock().await.contains_key(&state.edit.address); + Self::import_signed_mapping_address(db, state.edit.clone(), 0).await?; + // A newly learned record carries its retained boundary just as a + // bootstrap snapshot does. Never overwrite an existing local + // boundary, or a record changed by a concurrent membership update. + if was_missing && state.deleted_timestamp > 0 { + let mut address_map = ADDRESS_MAP.lock().await; + if let Some(node) = address_map.get_mut(&state.edit.address) { + if node.deleted_timestamp == 0 + && node.ip == state.edit.ip + && node.port == state.edit.port + && node.added_timestamp == state.edit.modified_timestamp + && node.added_by == state.edit.modified_by + && node.added_signature == state.edit.modified_signature + { + node.deleted_timestamp = state.deleted_timestamp; + node.deleted_block = state.deleted_block; + } + } + } + return Ok(()); } let mut address_map = ADDRESS_MAP.lock().await; if let Some(existing) = address_map.get(&state.edit.address) { // An unsigned remote deletion marker cannot deactivate a record // that this node currently derives as active. - if existing.deleted_timestamp == 0 { + if existing.is_active() { return Ok(()); } return Ok(()); @@ -870,7 +887,7 @@ impl NodeInfo { if unsigned_add_request { let address_map = ADDRESS_MAP.lock().await; if let Some(existing_node) = address_map.get(&edit.address) { - if existing_node.deleted_timestamp == 0 { + if existing_node.is_active() { if existing_node.ip == edit.ip { if existing_node.port == edit.port { return RpcResponse::Binary(b"Success".to_vec()); @@ -944,7 +961,7 @@ impl NodeInfo { let signer_node = address_map.get(&signer_key); let signer_is_local = signer_key == wallet.saved.short_address; let valid_added_by = signer_node - .map(|node| node.deleted_timestamp == 0) + .map(|node| node.is_active()) .unwrap_or(false); if !valid_added_by { let signer_exists = signer_node.is_some(); @@ -977,6 +994,11 @@ impl NodeInfo { } } + if let Some(existing) = address_map.get(&edit.address) { + if existing.deleted_timestamp > 0 && edit.modified_timestamp <= existing.deleted_timestamp { + return RpcResponse::Binary(b"Error: Reconnection record did not follow deletion".to_vec()); + } + } if Self::new_node_admission_limit_reached(&address_map, &edit, current_timestamp) { return RpcResponse::Binary( b"Error: Cannot add more than 10 nodes in 60 minutes".to_vec(), @@ -984,7 +1006,7 @@ impl NodeInfo { } if let Some(existing_node) = address_map.get(&edit.address) { - if existing_node.deleted_timestamp == 0 && existing_node.ip != edit.ip { + if existing_node.is_active() && existing_node.ip != edit.ip { return RpcResponse::Binary( b"Error: Active node must be deleted before changing IP".to_vec(), ); @@ -995,7 +1017,7 @@ impl NodeInfo { .iter() .find(|(address, node)| { address.as_str() != edit.address - && node.deleted_timestamp == 0 + && node.is_active() && node.ip == edit.ip && edit.ip != GENESIS_IP }) @@ -1059,24 +1081,7 @@ impl NodeInfo { // for historical validation. if !candidate_lost_collision { if let Some(existing_node) = address_map.get_mut(&edit.address) { - if existing_node.deleted_timestamp > 0 { - // A revived membership remains pending with its own add - // timestamp as the deletion marker until monitor activation. - // Redundant delivery of that exact signed record is already - // represented locally; acknowledge it without reviving it or - // changing any mapping state. Actual later deletions and - // different records must still pass the ordering check. - if existing_node.deleted_timestamp == existing_node.added_timestamp - && existing_node.deleted_block == 0 - && existing_node.monitoring.is_empty() - && existing_node.ip == edit.ip - && existing_node.port == edit.port - && existing_node.added_by == edit.modified_by - && existing_node.added_timestamp == edit.modified_timestamp - && existing_node.added_signature == edit.modified_signature - { - return RpcResponse::Binary(b"Success".to_vec()); - } + if !existing_node.is_active() { if edit.modified_timestamp <= existing_node.deleted_timestamp { return RpcResponse::Binary( b"Error: Reconnection record did not follow deletion".to_vec(), @@ -1157,8 +1162,127 @@ impl NodeInfo { mod tests { use super::*; + // Regression: the same signed evidence must select the same sponsor. + // Monitoring arrival must not change the membership ordering rule. #[tokio::test] - async fn repeated_signed_rejoin_preserves_pending_membership() { + async fn separate_rejoin_sponsors_converge_across_monitor_delivery_order() { + use crate::wallets::structures::SavedWallet; + use super::super::monitor::MONITOR_ACTION_ADD; + let db = sled::Config::new().temporary(true).open().unwrap(); + let make_wallet = || { + let (public, private) = Wallet::generate_keypair(); + let bytes = Wallet::public_key_bytes_to_short_address_bytes(&public).unwrap(); + let address = Wallet::bytes_to_short_address(&bytes).unwrap(); + crate::records::wallet_registry::register_short_address(&db, &bytes, &public).unwrap(); + Arc::new(Wallet { saved: SavedWallet { + short_address: address, vanity_address: None, + public_key: crate::encode(public), private_key: private, + }, encryption_key: String::new() }) + }; + let first = make_wallet(); + let second = make_wallet(); + let address = first.saved.short_address.clone(); + let monitor = second.saved.short_address.clone(); + let mut edits = Vec::new(); + for (wallet, timestamp) in [(&first, 200), (&second, 300)] { + edits.push(SignedNodeEdit { + address: address.clone(), ip: "8.8.8.8".into(), port: 50050, + modified_by: wallet.saved.short_address.clone(), modified_timestamp: timestamp, + modified_signature: NodeInfo::added_signature(&address, "8.8.8.8", 50050, timestamp, wallet).await, + }); + } + let monitor_edit = SignedMonitorEdit { + action: MONITOR_ACTION_ADD, monitored_address: address.clone(), + monitoring_address: monitor.clone(), target_ip: "8.8.8.8".into(), + modified_timestamp: 400, modified_block: 20, + modified_signature: NodeInfo::monitor_signature(MONITOR_ACTION_ADD, &address, &monitor, "8.8.8.8", 400, 20, &second).await, + }; + let previous_map = std::mem::take(&mut *ADDRESS_MAP.lock().await); + let previous_events = std::mem::take(&mut *MONITOR_EVENT_STATE.lock().await); + let mut results = Vec::new(); + for order in [[0, 2, 1], [1, 0, 2]] { + let mut old = NodeInfo::new("8.8.8.8".into(), 50050, 0, monitor.clone(), 50, String::new()); + old.deleted_timestamp = 100; + old.deleted_block = 10; + *ADDRESS_MAP.lock().await = HashMap::from([ + (address.clone(), old), + (monitor.clone(), NodeInfo::new("8.8.4.4".into(), 50050, 0, monitor.clone(), 50, String::new())), + ]); + MONITOR_EVENT_STATE.lock().await.clear(); + let mut responses = Vec::new(); + for event in order { + if event == 2 { + responses.push(format!("{:?}", NodeInfo::import_signed_monitor_event(monitor_edit.clone(), &db, &monitor).await)); + } else { + let RpcResponse::Binary(response) = NodeInfo::add_address_now(AddAddressParams { + map: Arc::new(Mutex::new(Command::new())), edit: edits[event].clone(), + monitors: Vec::new(), blocks_mined: 0, remote_ip: String::new(), + db: db.clone(), wallet: second.clone(), connections_key: "8.8.4.4:50050".into(), + }).await; + responses.push(String::from_utf8(response).unwrap()); + } + } + let state = ADDRESS_MAP.lock().await.remove(&address).unwrap(); + results.push((responses, state.added_timestamp, state.deleted_timestamp, state.monitoring)); + } + *ADDRESS_MAP.lock().await = previous_map; + *MONITOR_EVENT_STATE.lock().await = previous_events; + eprintln!("delivery A,monitor,B: {:?}; delivery B,A,monitor: {:?}", results[0], results[1]); + assert_eq!(results[0].0, ["Success", "Ok(true)", "Success"]); + assert_eq!(results[1].0, ["Success", "Success", "Ok(true)"]); + assert_eq!((results[0].1, results[1].1), (200, 200)); + assert_eq!((results[0].2, results[1].2), (100, 100)); + assert_eq!(results[0].3, results[1].3); + } + + #[tokio::test] + async fn rejoined_membership_ignores_older_monitor_withdrawal() { + use crate::wallets::structures::SavedWallet; + use super::super::monitor::MONITOR_ACTION_REMOVE; + let db = sled::Config::new().temporary(true).open().unwrap(); + let (public, private) = Wallet::generate_keypair(); + let bytes = Wallet::public_key_bytes_to_short_address_bytes(&public).unwrap(); + let signer = Wallet::bytes_to_short_address(&bytes).unwrap(); + crate::records::wallet_registry::register_short_address(&db, &bytes, &public).unwrap(); + let wallet = Arc::new(Wallet { saved: SavedWallet { + short_address: signer.clone(), vanity_address: None, + public_key: crate::encode(public), private_key: private, + }, encryption_key: String::new() }); + let target = "test-reconnected-target"; + let edit = SignedMonitorEdit { + action: MONITOR_ACTION_REMOVE, monitored_address: target.into(), + monitoring_address: signer.clone(), target_ip: "8.8.8.8".into(), + modified_timestamp: 150, modified_block: 10, + modified_signature: NodeInfo::monitor_signature(MONITOR_ACTION_REMOVE, target, &signer, "8.8.8.8", 150, 10, &wallet).await, + }; + let previous_map = std::mem::take(&mut *ADDRESS_MAP.lock().await); + let previous_events = std::mem::take(&mut *MONITOR_EVENT_STATE.lock().await); + let mut results = Vec::new(); + for marker in [100, 0] { + let mut reconnected = NodeInfo::new("8.8.8.8".into(), 50050, 0, signer.clone(), 300, String::new()); + reconnected.deleted_timestamp = marker; + *ADDRESS_MAP.lock().await = HashMap::from([ + (target.into(), reconnected), + (signer.clone(), NodeInfo::new("8.8.4.4".into(), 50050, 0, signer.clone(), 50, String::new())), + ]); + MONITOR_EVENT_STATE.lock().await.clear(); + let result = NodeInfo::import_signed_monitor_event(edit.clone(), &db, &signer).await; + let map = ADDRESS_MAP.lock().await; + results.push((result, map[target].deleted_timestamp, NodeInfo::validate_snapshot_structure(&map))); + } + *ADDRESS_MAP.lock().await = previous_map; + *MONITOR_EVENT_STATE.lock().await = previous_events; + eprintln!("pending marker: {:?}; cleared marker: {:?}", results[0], results[1]); + assert_eq!(results[0].0, Ok(false)); + assert_eq!(results[0].1, 100); + assert!(results[0].2.is_ok()); + assert_eq!(results[1].0, Ok(false)); + assert_eq!(results[1].1, 0); + assert!(results[1].2.is_ok()); + } + + #[tokio::test] + async fn repeated_signed_rejoin_preserves_deletion_boundary() { use crate::wallets::structures::SavedWallet; let db = sled::Config::new().temporary(true).open().unwrap(); let (public, private) = Wallet::generate_keypair(); @@ -1188,6 +1312,15 @@ mod tests { }; let RpcResponse::Binary(first) = NodeInfo::add_address_now(params()).await; let RpcResponse::Binary(repeated) = NodeInfo::add_address_now(params()).await; + let mut previous_cycle = params(); + previous_cycle.edit.modified_timestamp = 50; + previous_cycle.edit.modified_signature = NodeInfo::added_signature( + &address, "8.8.8.8", 50050, 50, &wallet, + ).await; + let rejected_import = NodeInfo::import_signed_mapping_address( + &db, previous_cycle.edit.clone(), 0, + ).await; + let RpcResponse::Binary(rejected_previous_cycle) = NodeInfo::add_address_now(previous_cycle).await; let mut different = params(); different.edit.port = 50051; different.edit.modified_signature = NodeInfo::added_signature( @@ -1207,17 +1340,34 @@ mod tests { let RpcResponse::Binary(stale) = NodeInfo::add_address_now(params()).await; let mut mapping = ADDRESS_MAP.lock().await; let after_stale = mapping.remove(&address).unwrap(); + drop(mapping); + let imported = NodeInfo::import_reconciled_membership_now(&db, SyncedNodeState { + edit: edit.clone(), deleted_timestamp: 100, deleted_block: 10, monitoring: Vec::new(), + }).await; + let governance_active = NodeInfo::governance_node_snapshot(&address).await.map(|node| node.is_active()); + let live_eligible = NodeInfo::address_checkup(&address, 20).await; + let historical_eligible = NodeInfo::historical_sync_address_checkup(&address, 20, 1).await; + let mut mapping = ADDRESS_MAP.lock().await; + let imported_state = mapping.remove(&address).unwrap(); if let Some(previous) = previous { mapping.insert(address, previous); } drop(mapping); assert_eq!(String::from_utf8(first).unwrap(), "Success"); assert_eq!(String::from_utf8(repeated).unwrap(), "Success"); - assert_eq!(String::from_utf8(conflicting).unwrap(), "Error: Reconnection record did not follow deletion"); + assert!(rejected_import.is_err()); + assert_eq!(String::from_utf8(rejected_previous_cycle).unwrap(), "Error: Reconnection record did not follow deletion"); + assert_eq!(String::from_utf8(conflicting).unwrap(), "Success"); assert_eq!(String::from_utf8(stale).unwrap(), "Error: Reconnection record did not follow deletion"); assert_eq!(after_stale.deleted_timestamp, 250); assert_eq!(after_stale.deleted_block, 20); + assert!(imported.is_ok()); + assert_eq!(imported_state.deleted_timestamp, 100); + assert_eq!(imported_state.deleted_block, 10); + assert!(imported_state.is_active()); + assert_eq!(governance_active, Some(true)); + assert!(live_eligible && historical_eligible); assert_eq!(actual.added_timestamp, 200); - assert_eq!(actual.deleted_timestamp, 200); - assert_eq!(actual.deleted_block, 0); + assert_eq!(actual.deleted_timestamp, 100); + assert_eq!(actual.deleted_block, 10); assert!(actual.monitoring.is_empty()); } @@ -1408,7 +1558,7 @@ mod tests { let deleted = map.get_mut("deleted.cltc").unwrap(); deleted.monitoring.clear(); deleted.deleted_timestamp = 9; - assert!(NodeInfo::validate_snapshot_structure(&map).is_err()); + assert!(NodeInfo::validate_snapshot_structure(&map).is_ok()); } #[test] @@ -1429,8 +1579,8 @@ mod tests { assert!(!NodeInfo::revive_deleted_membership(&mut node, &edit)); assert_eq!(node.added_timestamp, 30); - assert_eq!(node.deleted_timestamp, 30); - assert_eq!(node.deleted_block, 0); + assert_eq!(node.deleted_timestamp, 20); + assert_eq!(node.deleted_block, 100); assert!(node.monitoring.is_empty()); assert_eq!(node.blocks_mined, 95); diff --git a/src/records/memory/network_mapping/mod.rs b/src/records/memory/network_mapping/mod.rs index 329218e..a787d23 100644 --- a/src/records/memory/network_mapping/mod.rs +++ b/src/records/memory/network_mapping/mod.rs @@ -60,6 +60,10 @@ pub struct NodeInfo { } impl NodeInfo { + pub(crate) fn is_active(&self) -> bool { + self.deleted_timestamp == 0 || self.added_timestamp > self.deleted_timestamp + } + pub fn self_add_allowed_at_height(height: u32) -> bool { height <= SELF_ADD_BLOCK.saturating_add(SELF_ADD_WINDOW_BLOCKS) } diff --git a/src/records/memory/network_mapping/monitor.rs b/src/records/memory/network_mapping/monitor.rs index 4e979a9..0b7fc0d 100644 --- a/src/records/memory/network_mapping/monitor.rs +++ b/src/records/memory/network_mapping/monitor.rs @@ -111,48 +111,6 @@ mod tests { assert!(NodeInfo::monitor_event_order(&remove) > NodeInfo::monitor_event_order(&add)); } - #[test] - fn only_a_newer_monitor_add_activates_pending_membership() { - let mut pending = node(&[]); - pending.added_timestamp = 300; - pending.deleted_timestamp = 300; - - let mut edit = SignedMonitorEdit { - action: MONITOR_ACTION_ADD, - monitored_address: "target".to_string(), - monitoring_address: "monitor".to_string(), - target_ip: "1.2.3.4".to_string(), - modified_timestamp: 300, - modified_block: 50, - modified_signature: "signature".to_string(), - }; - assert!(!NodeInfo::can_activate_pending_membership(&edit, &pending)); - - edit.modified_timestamp = 301; - assert!(NodeInfo::can_activate_pending_membership(&edit, &pending)); - - edit.action = MONITOR_ACTION_REMOVE; - assert!(!NodeInfo::can_activate_pending_membership(&edit, &pending)); - } - - #[test] - fn only_freshly_sponsored_pending_membership_may_author_monitor_add() { - let address = "pending.cltc"; - let mut pending = node(&[]); - pending.added_timestamp = 300; - pending.deleted_timestamp = 300; - pending.deleted_block = 0; - - assert!(NodeInfo::can_pending_member_add_monitor(address, &pending)); - - pending.deleted_block = 50; - assert!(!NodeInfo::can_pending_member_add_monitor(address, &pending)); - - pending.deleted_block = 0; - pending.added_by = address.to_string(); - assert!(!NodeInfo::can_pending_member_add_monitor(address, &pending)); - } - #[test] fn signed_monitor_wire_record_round_trips() { let monitored = "ab13318c26250b048db92920a80a86127c933b0c.cltc".to_string(); @@ -224,10 +182,10 @@ impl NodeInfo { .filter(|(_, event)| { address_map .get(&event.monitored_address) - .is_some_and(|node| node.deleted_timestamp == 0) + .is_some_and(|node| node.is_active()) && address_map .get(&event.monitoring_address) - .is_some_and(|node| node.deleted_timestamp == 0) + .is_some_and(|node| node.is_active()) }) .map(|(key, event)| (key.clone(), event.clone())) .collect() @@ -262,7 +220,8 @@ impl NodeInfo { while let Some(address) = stack.pop() { let should_cascade = match address_map.get_mut(&address) { Some(_) if address == local_address => false, - Some(node) if node.deleted_timestamp == 0 && node.monitoring.is_empty() => { + Some(node) if node.is_active() && node.monitoring.is_empty() + && deleted_timestamp > node.added_timestamp => { node.deleted_timestamp = deleted_timestamp; node.deleted_block = deleted_block; true @@ -348,17 +307,6 @@ impl NodeInfo { ) } - fn can_activate_pending_membership(edit: &SignedMonitorEdit, node: &NodeInfo) -> bool { - edit.action == MONITOR_ACTION_ADD && edit.modified_timestamp > node.deleted_timestamp - } - - pub(crate) fn can_pending_member_add_monitor(address: &str, node: &NodeInfo) -> bool { - node.deleted_timestamp > 0 - && node.deleted_timestamp == node.added_timestamp - && node.deleted_block == 0 - && node.added_by != address - } - pub(super) fn discard_old_target_ip_events_from( state: &mut HashMap, address: &str, @@ -453,24 +401,26 @@ impl NodeInfo { if monitored.ip != edit.target_ip { return Err("monitor target IP mismatch".to_string()); } - if monitored.deleted_timestamp > 0 - && monitored.added_timestamp <= monitored.deleted_timestamp - && !Self::can_activate_pending_membership(edit, monitored) - { + if !monitored.is_active() { return Err( "monitor target must have a newer membership record before reconnection" .to_string(), ); } + // Delayed notifications from before this membership cannot change its + // monitor list or delete it. Authentication has already succeeded. + if edit.modified_timestamp <= monitored.added_timestamp { + return Ok(false); + } let monitoring = address_map .get(&edit.monitoring_address) .ok_or_else(|| "monitoring address not found".to_string())?; - if monitoring.deleted_timestamp > 0 - && !(edit.action == MONITOR_ACTION_ADD - && Self::can_pending_member_add_monitor(&edit.monitoring_address, monitoring)) - { + if !monitoring.is_active() { return Err("deleted node cannot update monitor relationships".to_string()); } + if edit.modified_timestamp <= monitoring.added_timestamp { + return Ok(false); + } let relation_key = Self::monitor_relation_key(edit); if event_state @@ -488,6 +438,7 @@ impl NodeInfo { .filter(|event| { event.monitored_address == edit.monitored_address && event.action == MONITOR_ACTION_REMOVE + && event.modified_timestamp > monitored.added_timestamp }) .map(|event| (event.modified_timestamp, event.modified_block)) .max() @@ -504,8 +455,7 @@ impl NodeInfo { if !monitored.monitoring.contains(&edit.monitoring_address) { monitored.monitoring.push(edit.monitoring_address.clone()); } - monitored.deleted_timestamp = 0; - monitored.deleted_block = 0; + // Monitor activation does not erase the previous deletion. } MONITOR_ACTION_REMOVE => { monitored diff --git a/src/records/memory/network_mapping/queries.rs b/src/records/memory/network_mapping/queries.rs index 22153c7..34e9913 100644 --- a/src/records/memory/network_mapping/queries.rs +++ b/src/records/memory/network_mapping/queries.rs @@ -20,12 +20,12 @@ impl NodeInfo { } fn eligible_at_block(node_info: &NodeInfo, block_number: u32) -> bool { - node_info.deleted_timestamp == 0 + node_info.is_active() || (node_info.deleted_block > 0 && block_number < node_info.deleted_block) } fn eligible_at_historical_sync_block( - address: &str, + _address: &str, node_info: &NodeInfo, block_number: u32, block_timestamp: u32, @@ -47,11 +47,7 @@ impl NodeInfo { return true; } - // A reconnect replaces the current membership with an inactive - // pending lifecycle. That current state must not invalidate blocks - // mined before the reconnect occurred. - Self::can_pending_member_add_monitor(address, node_info) - && block_timestamp_ms < node_info.added_timestamp + false } pub async fn has_reciprocal_sponsorship( @@ -62,7 +58,7 @@ impl NodeInfo { let Some(local) = map.get(local_address) else { return false; }; - if local.deleted_timestamp > 0 + if !local.is_active() || local.added_by == local_address || !operational_peer_wallets.contains(&local.added_by) { @@ -70,7 +66,7 @@ impl NodeInfo { } map.get(&local.added_by) - .map(|peer| peer.deleted_timestamp == 0 && peer.added_by == local_address) + .map(|peer| peer.is_active() && peer.added_by == local_address) .unwrap_or(false) } @@ -79,18 +75,7 @@ impl NodeInfo { .lock() .await .get(address) - .map(|node| node.deleted_timestamp == 0) - .unwrap_or(false) - } - - pub(crate) async fn is_active_or_sponsored_pending(address: &str) -> bool { - ADDRESS_MAP - .lock() - .await - .get(address) - .map(|node| { - node.deleted_timestamp == 0 || Self::can_pending_member_add_monitor(address, node) - }) + .map(|node| node.is_active()) .unwrap_or(false) } @@ -244,7 +229,7 @@ impl NodeInfo { ip: node.ip.clone(), blocks_mined: node.blocks_mined, added_timestamp: node.added_timestamp, - deleted_block: node.deleted_block, + deleted_block: if node.is_active() { 0 } else { node.deleted_block }, }) } @@ -257,7 +242,7 @@ impl NodeInfo { ip: node.ip.clone(), blocks_mined: node.blocks_mined, added_timestamp: node.added_timestamp, - deleted_block: node.deleted_block, + deleted_block: if node.is_active() { 0 } else { node.deleted_block }, }) .collect(); @@ -347,9 +332,9 @@ impl NodeInfo { pub async fn address_checkup(address: &str, block_number: u32) -> bool { let map = ADDRESS_MAP.lock().await; if let Some(node_info) = map.get(address) { - // Deleted nodes remain valid for blocks before their recorded - // deletion height. A deleted record with block zero is a pending - // or rejected lifecycle and is never eligible to mine. + // A newer add restores eligibility while retaining the previous + // deletion boundary. Inactive miners remain eligible only before + // their recorded deletion height. return Self::eligible_at_block(node_info, block_number); } false @@ -395,7 +380,7 @@ impl NodeInfo { map.values() // Deleted nodes stay in the map for history, but active node // lists should only return bare public IPs. - .filter(|node_info| node_info.deleted_timestamp == 0) + .filter(|node_info| node_info.is_active()) .map(|node_info| node_info.ip.clone()) .collect() } @@ -405,7 +390,7 @@ impl NodeInfo { map.values() // Outgoing connection fill needs a reachable endpoint, but node // uniqueness still remains IP-based elsewhere in the map. - .filter(|node_info| node_info.deleted_timestamp == 0) + .filter(|node_info| node_info.is_active()) .map(|node_info| format!("{}:{}", node_info.ip, node_info.port)) .collect() } @@ -416,7 +401,7 @@ impl NodeInfo { .filter_map(|(address, node_info)| { // The RPC response is a packed list of deleted short-address // bytes, so invalid address keys are skipped. - if node_info.deleted_timestamp > 0 { + if !node_info.is_active() { Wallet::short_address_to_bytes(address) } else { None @@ -522,7 +507,7 @@ mod tests { }; #[test] - fn pending_membership_is_not_mining_eligible() { + fn equal_add_and_deletion_timestamps_are_not_mining_eligible() { let mut pending = NodeInfo::new( "198.51.100.10".to_string(), 50050, @@ -537,12 +522,12 @@ mod tests { assert!(!NodeInfo::eligible_at_block(&pending, 0)); assert!(!NodeInfo::eligible_at_block(&pending, 10_000)); - pending.deleted_timestamp = 0; + pending.added_timestamp = 301; assert!(NodeInfo::eligible_at_block(&pending, 10_000)); } #[test] - fn historical_sync_accepts_only_blocks_before_pending_reconnect() { + fn reconnected_membership_accepts_historical_and_live_blocks() { let address = "node.cltc"; let mut pending = NodeInfo::new( "198.51.100.10".to_string(), @@ -552,16 +537,16 @@ mod tests { 300_000, "signature".to_string(), ); - pending.deleted_timestamp = 300_000; + pending.deleted_timestamp = 200_000; pending.deleted_block = 0; assert!(NodeInfo::eligible_at_historical_sync_block( address, &pending, 10, 299 )); - assert!(!NodeInfo::eligible_at_historical_sync_block( + assert!(NodeInfo::eligible_at_historical_sync_block( address, &pending, 11, 300 )); - assert!(!NodeInfo::eligible_at_block(&pending, 10)); + assert!(NodeInfo::eligible_at_block(&pending, 10)); } #[test] @@ -591,7 +576,7 @@ mod tests { } #[test] - fn historical_sync_does_not_activate_self_sponsored_pending_member() { + fn historical_sync_accepts_reconnected_self_sponsored_member() { let address = "node.cltc"; let mut pending = NodeInfo::new( "198.51.100.10".to_string(), @@ -601,10 +586,10 @@ mod tests { 300_000, "signature".to_string(), ); - pending.deleted_timestamp = 300_000; + pending.deleted_timestamp = 200_000; pending.deleted_block = 0; - assert!(!NodeInfo::eligible_at_historical_sync_block( + assert!(NodeInfo::eligible_at_historical_sync_block( address, &pending, 10, 299 )); }