contractless/Contractless
contractless
/
Contractless
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

mapping order fix

This commit is contained in:
contractless 2026-09-12 12:41:01 -06:00
parent ca24d25f68
commit 2b5f435276
6 changed files with 263 additions and 162 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
src/bin/lookup_online_node_count.rs
View File

@ -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<usize, String> {
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)]);

8
src/records/memory/connections.rs
View File

@ -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;
}

254
src/records/memory/network_mapping/add.rs
View File

@ -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::<String, String>::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);

4
src/records/memory/network_mapping/mod.rs
View File

@ -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)
}

82
src/records/memory/network_mapping/monitor.rs
View File

@ -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<String, SignedMonitorEdit>,
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

61
src/records/memory/network_mapping/queries.rs
View File

@ -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
));
}