diff --git a/src/orphans/undo_transactions/restore_mempool.rs b/src/orphans/undo_transactions/restore_mempool.rs index f813539..4fca29b 100644 --- a/src/orphans/undo_transactions/restore_mempool.rs +++ b/src/orphans/undo_transactions/restore_mempool.rs @@ -9,31 +9,64 @@ use crate::blocks::swap::SwapTransaction; use crate::blocks::token::CreateTokenTransaction; use crate::blocks::transfer::TransferTransaction; use crate::blocks::vanity::VanityAddressTransaction; -use crate::common::asset_names::loan_collateral_asset_name_or_base; use crate::common::nft_assets::nft_asset_name; use crate::common::types::Transaction; use crate::decode; -use crate::records::memory::mempool::{restore_processed_by_signatures, BASECOIN}; +use crate::log::warn; +use crate::records::memory::mempool::{db_client, restore_processed_by_signatures, BASECOIN}; use crate::rpc::commands::transaction_by_txid::request_transaction_by_txid; use crate::rpc::responses::RpcResponse; use crate::sled::Db; use crate::verifications::async_funcs::checks::balance_check::balance_checkup; -async fn restore_if_spendable(signatures: &[String], spendable: bool, insert: F) -where +async fn restore_if_spendable( + tx_label: &str, + txid: &str, + signatures: &[String], + spendable: bool, + insert: F, +) where F: FnOnce() -> Fut, - Fut: std::future::Future, + Fut: std::future::Future>>, { if !spendable { + warn!("[mempool_restore] skipped rolled-back tx: type={tx_label} txid={txid} reason=not_spendable_after_rollback"); return; } match restore_processed_by_signatures(signatures).await { - Ok(false) | Err(_) => insert().await, Ok(true) => {} + Ok(false) => { + if let Err(err) = insert().await { + warn!("[mempool_restore] failed to reinsert rolled-back tx: type={tx_label} txid={txid} err={err}"); + } + } + Err(err) => { + warn!("[mempool_restore] failed to unmark processed rolled-back tx: type={tx_label} txid={txid} err={err}"); + if let Err(insert_err) = insert().await { + warn!("[mempool_restore] failed to reinsert rolled-back tx after unmark error: type={tx_label} txid={txid} err={insert_err}"); + } + } } } +async fn restore_if_spendable_void( + tx_label: &str, + txid: &str, + signatures: &[String], + spendable: bool, + insert: F, +) where + F: FnOnce() -> Fut, + Fut: std::future::Future, +{ + restore_if_spendable(tx_label, txid, signatures, spendable, || async { + insert().await; + Ok(()) + }) + .await; +} + async fn loan_coin_for_contract(db: &Db, contract_hash: &str) -> Option { let contract_key = decode(contract_hash).ok()?; let RpcResponse::Binary(bytes) = request_transaction_by_txid(db, contract_key).await; @@ -48,6 +81,54 @@ async fn loan_coin_for_contract(db: &Db, contract_hash: &str) -> Option .map(|contract| contract.unsigned_loan_contract.loan_coin) } +async fn restore_loan_contract_row( + transaction: &LoanContractTransaction, +) -> Result<(), Box> { + let client_handle = db_client().await?; + let client = client_handle.as_ref(); + let hash = &transaction.hash; + let signature1 = &transaction.signature1; + let signature2 = &transaction.signature2; + + client + .execute( + r#" + UPDATE loan_contract + SET processed=false, + processed_block_number=NULL + WHERE signature1 = $1 + OR signature2 = $2 + OR hash = $3 + OR txid = $3 + "#, + &[signature1, signature2, hash], + ) + .await?; + + let row = client + .query_one( + r#" + SELECT COUNT(*) + FROM loan_contract + WHERE processed=false + AND ( + signature1 = $1 + OR signature2 = $2 + OR hash = $3 + OR txid = $3 + ) + "#, + &[signature1, signature2, hash], + ) + .await?; + let pending_count: i64 = row.get(0); + if pending_count == 0 { + transaction.add_to_memory().await?; + } + + Ok(()) +} + pub async fn restore_transfer(transaction: &TransferTransaction, db: &Db) { let transfer = &transaction.unsigned_transfer; let asset = nft_asset_name(&transfer.coin, transfer.nft_series); @@ -55,7 +136,7 @@ pub async fn restore_transfer(transaction: &TransferTransaction, db: &Db) { balance_checkup(db, transfer.value, transfer.txfee, asset, &transfer.sender).await; let signature = transaction.signature.clone(); - restore_if_spendable(&[signature], spendable, || async { + restore_if_spendable_void("transfer", "unknown", &[signature], spendable, || async { let _ = transaction.add_to_memory().await; }) .await; @@ -67,7 +148,7 @@ pub async fn restore_burn(transaction: &BurnTransaction, db: &Db) { let spendable = balance_checkup(db, burn.value, burn.txfee, asset, &burn.address).await; let signature = transaction.signature.clone(); - restore_if_spendable(&[signature], spendable, || async { + restore_if_spendable_void("burn", "unknown", &[signature], spendable, || async { let _ = transaction.add_to_memory().await; }) .await; @@ -78,9 +159,15 @@ pub async fn restore_create_token(transaction: &CreateTokenTransaction, db: &Db) let spendable = balance_checkup(db, 0, token.txfee, BASECOIN.clone(), &token.creator).await; let signature = transaction.signature.clone(); - restore_if_spendable(&[signature], spendable, || async { - let _ = transaction.add_to_memory().await; - }) + restore_if_spendable_void( + "create_token", + "unknown", + &[signature], + spendable, + || async { + let _ = transaction.add_to_memory().await; + }, + ) .await; } @@ -89,9 +176,15 @@ pub async fn restore_issue_token(transaction: &IssueTokenTransaction, db: &Db) { let spendable = balance_checkup(db, 0, token.txfee, BASECOIN.clone(), &token.creator).await; let signature = transaction.signature.clone(); - restore_if_spendable(&[signature], spendable, || async { - let _ = transaction.add_to_memory().await; - }) + restore_if_spendable_void( + "issue_token", + "unknown", + &[signature], + spendable, + || async { + let _ = transaction.add_to_memory().await; + }, + ) .await; } @@ -100,7 +193,7 @@ pub async fn restore_create_nft(transaction: &CreateNftTransaction, db: &Db) { let spendable = balance_checkup(db, 0, nft.txfee, BASECOIN.clone(), &nft.creator).await; let signature = transaction.signature.clone(); - restore_if_spendable(&[signature], spendable, || async { + restore_if_spendable_void("create_nft", "unknown", &[signature], spendable, || async { let _ = transaction.add_to_memory().await; }) .await; @@ -118,7 +211,7 @@ pub async fn restore_marketing(transaction: &MarketingTransaction, db: &Db) { .await; let signature = transaction.signature.clone(); - restore_if_spendable(&[signature], spendable, || async { + restore_if_spendable_void("marketing", "unknown", &[signature], spendable, || async { let _ = transaction.add_to_memory().await; }) .await; @@ -137,7 +230,9 @@ pub async fn restore_swap(transaction: &SwapTransaction, db: &Db) { transaction.signature2.clone(), ]; - restore_if_spendable( + restore_if_spendable_void( + "swap", + "unknown", &signatures, sender1_spendable && sender2_spendable, || async { @@ -147,34 +242,13 @@ pub async fn restore_swap(transaction: &SwapTransaction, db: &Db) { .await; } -pub async fn restore_loan_creation(transaction: &LoanContractTransaction, db: &Db) { - let loan = &transaction.unsigned_loan_contract; - let Some(collateral) = loan_collateral_asset_name_or_base(&loan.collateral, &BASECOIN) else { - return; - }; - let lender_spendable = balance_checkup( - db, - loan.loan_amount, - loan.txfee, - loan.loan_coin.clone(), - &loan.lender, - ) - .await; - let borrower_spendable = - balance_checkup(db, loan.collateral_amount, 0, collateral, &loan.borrower).await; - let signatures = vec![ - transaction.signature1.clone(), - transaction.signature2.clone(), - ]; - - restore_if_spendable( - &signatures, - lender_spendable && borrower_spendable, - || async { - let _ = transaction.add_to_memory().await; - }, - ) - .await; +pub async fn restore_loan_creation(transaction: &LoanContractTransaction, _db: &Db) { + if let Err(err) = restore_loan_contract_row(transaction).await { + warn!( + "[mempool_restore] failed to restore rolled-back loan_contract txid={} err={err}", + transaction.hash + ); + } } pub async fn restore_borrower(transaction: &ContractPaymentTransaction, db: &Db) { @@ -186,9 +260,15 @@ pub async fn restore_borrower(transaction: &ContractPaymentTransaction, db: &Db) let spendable = balance_checkup(db, value, payment.txfee, loan_coin, &payment.address).await; let signature = transaction.signature.clone(); - restore_if_spendable(&[signature], spendable, || async { - let _ = transaction.add_to_memory().await; - }) + restore_if_spendable_void( + "loan_payment", + "unknown", + &[signature], + spendable, + || async { + let _ = transaction.add_to_memory().await; + }, + ) .await; } @@ -204,9 +284,15 @@ pub async fn restore_collateral(transaction: &CollateralClaimTransaction, db: &D .await; let signature = transaction.signature.clone(); - restore_if_spendable(&[signature], spendable, || async { - let _ = transaction.add_to_memory().await; - }) + restore_if_spendable_void( + "collateral_claim", + "unknown", + &[signature], + spendable, + || async { + let _ = transaction.add_to_memory().await; + }, + ) .await; } @@ -215,7 +301,7 @@ pub async fn restore_vanity(transaction: &VanityAddressTransaction, db: &Db) { let spendable = balance_checkup(db, 0, vanity.txfee, BASECOIN.clone(), &vanity.address).await; let signature = transaction.signature.clone(); - restore_if_spendable(&[signature], spendable, || async { + restore_if_spendable_void("vanity", "unknown", &[signature], spendable, || async { let _ = transaction.add_to_memory().await; }) .await; diff --git a/src/records/memory/network_mapping/add.rs b/src/records/memory/network_mapping/add.rs index 323a961..9180f9b 100644 --- a/src/records/memory/network_mapping/add.rs +++ b/src/records/memory/network_mapping/add.rs @@ -264,21 +264,35 @@ impl NodeInfo { // Once the chain is mature, adding nodes is restricted to older // active participants with sufficient mined history. - if get_height(&db) > 10000 { + let local_height = get_height(&db); + if !Self::self_add_allowed_at_height(local_height) { let signer_key = Wallet::normalize_to_short_address(&edit.modified_by) .unwrap_or_else(|| edit.modified_by.clone()); 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| { current_timestamp.saturating_sub(node.added_timestamp) >= ONE_HOUR_MILLIS && node.deleted_timestamp == 0 + && (!signer_is_local + || Self::local_runtime_is_mature(current_timestamp)) }) .unwrap_or(false); if !valid_added_by { - return RpcResponse::Binary( - b"Error: This address cannot add nodes. It must exist for at least 60 minutes and not be marked for deletion" - .to_vec(), - ); + let signer_exists = signer_node.is_some(); + let signer_deleted_timestamp = + signer_node.map(|node| node.deleted_timestamp).unwrap_or(0); + let signer_age_ms = signer_node + .map(|node| current_timestamp.saturating_sub(node.added_timestamp)) + .unwrap_or(0); + let signer_blocks_mined = + signer_node.map(|node| node.blocks_mined).unwrap_or(0); + return RpcResponse::Binary(format!( + "Error: This address cannot add nodes. It must exist for at least 60 minutes and not be marked for deletion local_height={local_height} self_add_block={} self_add_limit={} signer={signer_key} signer_exists={signer_exists} signer_is_local={signer_is_local} signer_deleted_timestamp={signer_deleted_timestamp} signer_age_ms={signer_age_ms} signer_blocks_mined={signer_blocks_mined}", + super::SELF_ADD_BLOCK, + Self::self_add_limit_height(), + ) + .into_bytes()); } let mined_count = signer_node.map(|node| node.blocks_mined).unwrap_or(0); if mined_count < 100 { diff --git a/src/records/memory/network_mapping/mod.rs b/src/records/memory/network_mapping/mod.rs index 4268e1f..1df9f73 100644 --- a/src/records/memory/network_mapping/mod.rs +++ b/src/records/memory/network_mapping/mod.rs @@ -21,12 +21,18 @@ use crate::wallets::structures::Wallet; use crate::Arc; use crate::HashMap; use crate::Mutex; +use crate::OnceLock; use crate::Utc; lazy_static! { static ref ADDRESS_MAP: Mutex> = Mutex::new(HashMap::new()); } +pub const SELF_ADD_BLOCK: u32 = 34_000; +pub const SELF_ADD_WINDOW_BLOCKS: u32 = 10_000; + +static NODE_RUNTIME_STARTED_MILLIS: OnceLock = OnceLock::new(); + #[derive(Debug)] pub struct NodeInfo { ip: String, @@ -40,6 +46,26 @@ pub struct NodeInfo { } impl NodeInfo { + pub fn initialize_runtime_start() { + let _ = NODE_RUNTIME_STARTED_MILLIS.set(Utc::now().timestamp_millis() as u64); + } + + fn runtime_started_millis() -> u64 { + *NODE_RUNTIME_STARTED_MILLIS.get_or_init(|| Utc::now().timestamp_millis() as u64) + } + + fn local_runtime_is_mature(current_timestamp: u64) -> bool { + current_timestamp.saturating_sub(Self::runtime_started_millis()) >= 3_600_000 + } + + pub fn self_add_allowed_at_height(height: u32) -> bool { + height <= SELF_ADD_BLOCK.saturating_add(SELF_ADD_WINDOW_BLOCKS) + } + + pub fn self_add_limit_height() -> u32 { + SELF_ADD_BLOCK.saturating_add(SELF_ADD_WINDOW_BLOCKS) + } + fn new( ip: String, blocks_mined: u8, diff --git a/src/records/memory/network_mapping/monitor.rs b/src/records/memory/network_mapping/monitor.rs index d482f14..f169a02 100644 --- a/src/records/memory/network_mapping/monitor.rs +++ b/src/records/memory/network_mapping/monitor.rs @@ -24,6 +24,7 @@ impl NodeInfo { fn mark_deleted_and_cascade( address_map: &mut HashMap, deleted_address: &str, + local_address: &str, current_timestamp: u64, deleted_block: u32, ) { @@ -31,6 +32,7 @@ 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() => { node.deleted_timestamp = current_timestamp; node.deleted_block = deleted_block; @@ -155,6 +157,7 @@ impl NodeInfo { Self::mark_deleted_and_cascade( &mut address_map, &edit.monitored_address, + &local_short, current_timestamp, deleted_block, ); diff --git a/src/rpc/client/handshake_processing.rs b/src/rpc/client/handshake_processing.rs index 0eb8515..03cde20 100644 --- a/src/rpc/client/handshake_processing.rs +++ b/src/rpc/client/handshake_processing.rs @@ -375,24 +375,53 @@ pub async fn process_handshake_response( ))); } + let remote_height = request_remote_height( + broadcast_stream.clone(), + params.map.clone(), + connections_key.clone(), + ) + .await + .map_err(|err| { + let connections_key = connections_key.clone(); + tokio::spawn(async move { + remove_key_from_memory(&connections_key).await; + }); + io::Error::other(format!( + "Remote height lookup failed before network join: {err}" + )) + })?; + let self_add_block = crate::records::memory::network_mapping::SELF_ADD_BLOCK; + let self_add_limit = crate::records::memory::network_mapping::NodeInfo::self_add_limit_height(); + let join_mode = if crate::records::memory::network_mapping::NodeInfo::self_add_allowed_at_height( + remote_height, + ) { + "bootstrap_self_add" + } else { + "sponsored_add" + }; + info!( + "[handshake_state] peer={connections_key} remote_height={remote_height} self_add_block={self_add_block} self_add_limit={self_add_limit} join_mode={join_mode}" + ); + + announce_self_to_network( + broadcast_stream.clone(), + &wallet.saved.short_address.clone(), + params.map.clone(), + ¶ms.db.clone(), + params.wallet.clone(), + &connections_key, + ) + .await + .map_err(|err| { + let connections_key = connections_key.clone(); + tokio::spawn(async move { + remove_key_from_memory(&connections_key).await; + }); + io::Error::other(format!("Network join/map sync failed: {err}")) + })?; + mark_peer_network_map_synced(&connections_key).await; + if params.first { - announce_self_to_network( - broadcast_stream.clone(), - &wallet.saved.short_address.clone(), - params.map.clone(), - ¶ms.db.clone(), - params.wallet.clone(), - &connections_key, - ) - .await - .map_err(|err| { - let connections_key = connections_key.clone(); - tokio::spawn(async move { - remove_key_from_memory(&connections_key).await; - }); - io::Error::other(format!("Network map sync failed: {err}")) - })?; - mark_peer_network_map_synced(&connections_key).await; let bsparams = BootstrapParams { stream: Arc::clone(&stream), connections_key: connections_key.clone(), @@ -405,23 +434,6 @@ pub async fn process_handshake_response( spawn_bootstrap_peer_discovery(bsparams); } else { - announce_self_to_network( - broadcast_stream.clone(), - &wallet.saved.short_address.clone(), - params.map.clone(), - ¶ms.db.clone(), - params.wallet.clone(), - &connections_key, - ) - .await - .map_err(|err| { - let connections_key = connections_key.clone(); - tokio::spawn(async move { - remove_key_from_memory(&connections_key).await; - }); - io::Error::other(format!("Network map sync failed: {err}")) - })?; - mark_peer_network_map_synced(&connections_key).await; if crate::miner::flag::is_normal_mode() { mark_peer_operational(&connections_key, params.map.clone()).await; } diff --git a/src/rpc/commands/contract.rs b/src/rpc/commands/contract.rs index abd0862..80c2e39 100644 --- a/src/rpc/commands/contract.rs +++ b/src/rpc/commands/contract.rs @@ -4,11 +4,14 @@ use crate::blocks::loans::LoanContractTransaction; use crate::common::loan_schedule::overdue_payment_count; use crate::encode; use crate::records::memory::mempool::get_pending_payments_for_contract; +use crate::records::record_chain::wallet_tx_index::WALLET_TX_INDEX_TREE; use crate::rpc::commands::transaction_by_txid::request_transaction_by_txid; use crate::rpc::responses::RpcResponse; use crate::sled::Db; use crate::wallets::structures::Wallet; use crate::{Local, TimeZone, Utc}; +use std::collections::{BTreeMap, BTreeSet}; +use std::ops::Bound; fn format_amount(value: u64) -> f64 { // Contract RPC output presents coin amounts as user-facing decimal values. @@ -52,9 +55,10 @@ async fn load_contract(db: &Db, hash: &[u8]) -> Result>, ) -> Result< ( Vec, @@ -62,16 +66,12 @@ async fn collect_contract_activity( ), String, > { - // Scan saved txids to gather all payments and any collateral-claim - // transaction associated with the requested contract. - let tree = db - .open_tree("txid") - .map_err(|e| format!("error: Failed to open txid tree: {e}"))?; + // Inspect only txids already indexed to the contract parties. This keeps + // loan lookups bounded by wallet activity instead of total chain size. let mut payments = Vec::new(); let mut collateral_claim: Option = None; - for entry in tree.iter() { - let (txid_bytes, _) = entry.map_err(|e| format!("error: Failed to read txid tree: {e}"))?; + for txid_bytes in txids { let RpcResponse::Binary(bytes) = request_transaction_by_txid(db, txid_bytes.to_vec()).await; if bytes.is_empty() { continue; @@ -112,19 +112,24 @@ async fn collect_contract_activity( Ok((payments, collateral_claim)) } -async fn build_contract_summary(hash: Vec, db: &Db) -> Result { +async fn build_contract_summary_from_parts( + contract_hash: String, + contract: LoanContractTransaction, + mut payments: Vec, + collateral_claim: Option, + db: &Db, +) -> Result { // Build the text view of a contract by combining the original loan, // its payments, and any collateral-claim activity into one summary. - let contract_hash = encode(&hash); - let contract = load_contract(db, &hash).await?; - - let (payments, collateral_claim) = collect_contract_activity(db, &contract_hash).await?; + payments.sort_by_key(|payment| payment.unsigned_contract_payment.timestamp); let loan_tree = db .open_tree("loan") .map_err(|err| format!("error: Failed to open loan tree: {err}"))?; // The loan tree stores whether collateral remains claimable; if the // tree does not have the row, fall back to discovered claim history. - let collateral_already_claimed = match loan_tree.get(&hash) { + let contract_hash_bytes = crate::decode(&contract_hash) + .map_err(|err| format!("error: Invalid contract hash: {err}"))?; + let collateral_already_claimed = match loan_tree.get(&contract_hash_bytes) { Ok(Some(value)) => value == b"false", _ => collateral_claim.is_some(), }; @@ -216,6 +221,16 @@ async fn build_contract_summary(hash: Vec, db: &Db) -> Result, db: &Db) -> Result { + let contract_hash = encode(&hash); + let contract = load_contract(db, &hash).await?; + let loan = &contract.unsigned_loan_contract; + let txids = indexed_txids_for_addresses(db, &[loan.lender.trim(), loan.borrower.trim()])?; + let (payments, collateral_claim) = + collect_contract_activity_from_txids(db, &contract_hash, txids).await?; + build_contract_summary_from_parts(contract_hash, contract, payments, collateral_claim, db).await +} + pub async fn contract_details(hash: Vec, db: &Db) -> RpcResponse { // Return the text summary for one contract hash. let output = match build_contract_summary(hash, db).await { @@ -225,6 +240,57 @@ pub async fn contract_details(hash: Vec, db: &Db) -> RpcResponse { RpcResponse::Binary(output.into_bytes()) } +fn prefix_successor(prefix: &[u8]) -> Option> { + let mut next = prefix.to_vec(); + for index in (0..next.len()).rev() { + if next[index] != u8::MAX { + next[index] += 1; + next.truncate(index + 1); + return Some(next); + } + } + None +} + +fn indexed_txids_for_address(db: &Db, address: &str) -> Result>, String> { + let address_bytes = Wallet::short_address_to_bytes(address) + .ok_or_else(|| "error: Invalid wallet address".to_string())?; + let tree = db + .open_tree(WALLET_TX_INDEX_TREE) + .map_err(|err| format!("error: Failed to open wallet transaction index: {err}"))?; + let start = Bound::Included(address_bytes.clone()); + let end = prefix_successor(&address_bytes) + .map(Bound::Excluded) + .unwrap_or(Bound::Unbounded); + let mut txids = Vec::new(); + let mut seen = BTreeSet::new(); + + for entry in tree.range((start, end)) { + let (_key, value) = entry + .map_err(|err| format!("error: Failed to read wallet transaction index: {err}"))?; + if value.len() == 32 && seen.insert(value.to_vec()) { + txids.push(value.to_vec()); + } + } + + Ok(txids) +} + +fn indexed_txids_for_addresses(db: &Db, addresses: &[&str]) -> Result>, String> { + let mut seen = BTreeSet::new(); + let mut txids = Vec::new(); + + for address in addresses { + for txid in indexed_txids_for_address(db, address.trim())? { + if seen.insert(txid.clone()) { + txids.push(txid); + } + } + } + + Ok(txids) +} + pub async fn contract_details_by_address(address: String, db: &Db) -> RpcResponse { // Return every saved contract where the address appears as either the // lender or borrower, each expanded into the same text summary view. @@ -232,27 +298,13 @@ pub async fn contract_details_by_address(address: String, db: &Db) -> RpcRespons return RpcResponse::Binary(b"error: Invalid wallet address".to_vec()); }; - let tree = match db.open_tree("txid") { - Ok(tree) => tree, - Err(err) => { - return RpcResponse::Binary( - format!("error: Failed to open txid tree: {err}").into_bytes(), - ) - } + let txids = match indexed_txids_for_address(db, &address) { + Ok(txids) => txids, + Err(err) => return RpcResponse::Binary(err.into_bytes()), }; + let mut contracts = BTreeMap::::new(); - let mut contracts = Vec::new(); - - for entry in tree.iter() { - let (txid_bytes, _) = match entry { - Ok(entry) => entry, - Err(err) => { - return RpcResponse::Binary( - format!("error: Failed to read txid tree: {err}").into_bytes(), - ) - } - }; - + for txid_bytes in txids { let RpcResponse::Binary(bytes) = request_transaction_by_txid(db, txid_bytes.to_vec()).await; if bytes.is_empty() { continue; @@ -263,28 +315,47 @@ pub async fn contract_details_by_address(address: String, db: &Db) -> RpcRespons continue; } - // Only loan-contract transactions can be expanded into the - // contract summary returned by this RPC. let contract = match LoanContractTransaction::from_bytes(txtype, &bytes[1..]).await { Ok(contract) => contract, Err(_) => continue, }; - - if contract.unsigned_loan_contract.lender != address - && contract.unsigned_loan_contract.borrower != address + if contract.unsigned_loan_contract.lender == address + || contract.unsigned_loan_contract.borrower == address { - continue; + contracts.insert(encode(&txid_bytes), contract); } + } - let summary = match build_contract_summary(txid_bytes.to_vec(), db).await { + let mut summaries = Vec::new(); + for (contract_hash, contract) in contracts { + let loan = &contract.unsigned_loan_contract; + let txids = + match indexed_txids_for_addresses(db, &[loan.lender.trim(), loan.borrower.trim()]) { + Ok(txids) => txids, + Err(err) => return RpcResponse::Binary(err.into_bytes()), + }; + let (payments, collateral_claim) = + match collect_contract_activity_from_txids(db, &contract_hash, txids).await { + Ok(activity) => activity, + Err(err) => return RpcResponse::Binary(err.into_bytes()), + }; + let summary = match build_contract_summary_from_parts( + contract_hash, + contract, + payments, + collateral_claim, + db, + ) + .await + { Ok(summary) => summary, Err(err) => return RpcResponse::Binary(err.into_bytes()), }; - contracts.push(summary); + summaries.push(summary); } - let mut output = format!("address={}\ncontracts={}\n", address, contracts.len()); - for (index, contract) in contracts.iter().enumerate() { + let mut output = format!("address={}\ncontracts={}\n", address, summaries.len()); + for (index, contract) in summaries.iter().enumerate() { output.push_str(&format!("contract_{}:\n{}\n", index + 1, contract)); } diff --git a/src/rpc/commands/tx_submit.rs b/src/rpc/commands/tx_submit.rs index 723e952..49ea940 100644 --- a/src/rpc/commands/tx_submit.rs +++ b/src/rpc/commands/tx_submit.rs @@ -19,7 +19,7 @@ use crate::records::memory::response_channels::{reserve_transient_entry_with_con use crate::rpc::command_maps::RPC_SUBMIT_TRANSACTION; use crate::rpc::responses::RpcResponse; use crate::sled::Db; -use crate::torrent::torrenting_system::get_nodes::get_nodes_from_memory; +use crate::torrent::torrenting_system::get_nodes::get_live_node_broadcast_peers; use crate::Arc; use crate::Mutex; use crate::OnceLock; @@ -33,7 +33,7 @@ fn loan_payment_submit_lock() -> &'static Mutex<()> { async fn broadcast_tx(tx_bytes: Vec, map: Arc>) { // Broadcast newly accepted mempool transactions only to miner peers, // since those are the nodes that need transaction fan-out most. - let nodes = get_nodes_from_memory().await; + let nodes = get_live_node_broadcast_peers().await; for (key, stream) in nodes { let client_type = get_client_type_from_memory(&key) diff --git a/src/rpc/server/handshake.rs b/src/rpc/server/handshake.rs index 5645a2f..8745e5e 100644 --- a/src/rpc/server/handshake.rs +++ b/src/rpc/server/handshake.rs @@ -25,7 +25,7 @@ use crate::rpc::server::structs::{CombineAndSendDataParams, HandshakeTestParams} use crate::rpc::server::tests::{endpoint_port, is_port_open}; use crate::sled::Db; use crate::sleep; -use crate::startup::network_broadcast::announce_self_to_network; +use crate::startup::network_broadcast::{announce_self_to_network, get_network_mapping}; use crate::startup::remote_height::request_remote_height; use crate::wallets::structures::Wallet; use crate::Arc; @@ -210,11 +210,27 @@ async fn complete_incoming_miner_setup( return; } - if get_height(db) <= 10_000 { + let remote_height = + match request_remote_height(stream.clone(), map.clone(), connections_key.to_string()).await + { + Ok(height) => height, + Err(err) => { + error!( + "[startup] incoming peer remote height lookup failed before network join: {err}" + ); + remove_key_from_memory(connections_key).await; + return; + } + }; + let self_add_allowed = + crate::records::memory::network_mapping::NodeInfo::self_add_allowed_at_height( + remote_height, + ); + + if self_add_allowed { // Before the sponsored-add gate, reverse announcement lets tiny early // networks bootstrap from one live peer. After the gate activates, the - // incoming peer may have no local history yet, so only its outgoing - // self-announcement should be sponsored by this established node. + // incoming peer's outgoing join request is the only add path. let short_address = wallet.saved.short_address.clone(); if let Err(err) = announce_self_to_network( stream.clone(), @@ -230,6 +246,18 @@ async fn complete_incoming_miner_setup( remove_key_from_memory(connections_key).await; return; } + } else if let Err(err) = get_network_mapping( + stream.clone(), + map.clone(), + db, + wallet.clone(), + connections_key, + ) + .await + { + error!("[startup] incoming peer network map import failed: {err}"); + remove_key_from_memory(connections_key).await; + return; } mark_peer_network_map_synced(connections_key).await; diff --git a/src/startup/initialize_startup.rs b/src/startup/initialize_startup.rs index 164ee4d..039c872 100644 --- a/src/startup/initialize_startup.rs +++ b/src/startup/initialize_startup.rs @@ -7,6 +7,7 @@ use crate::miner::flag::{ request_mining_stop, set_mining_state, set_node_mode, MiningState, NodeMode, }; use crate::records::memory::connections::initialize_connection; +use crate::records::memory::network_mapping::NodeInfo; use crate::wallets::structures::Wallet; use crate::Path; @@ -68,6 +69,7 @@ pub async fn prepare_pre_wallet_startup() { set_node_mode(NodeMode::Startup); request_mining_stop(); set_mining_state(MiningState::Idle); + NodeInfo::initialize_runtime_start(); initialize_connection().await; create_file_paths().await; } diff --git a/src/torrent/create_metadata.rs b/src/torrent/create_metadata.rs index daebe8a..3fefe03 100644 --- a/src/torrent/create_metadata.rs +++ b/src/torrent/create_metadata.rs @@ -7,7 +7,7 @@ use crate::records::memory::response_channels::{reserve_transient_entry_with_con use crate::rpc::command_maps::RPC_SUBMIT_TORRENT; use crate::rpc::responses::RpcResponse; use crate::torrent::structs::{Info, Torrent}; -use crate::torrent::torrenting_system::get_nodes::get_torrent_broadcast_nodes_from_memory; +use crate::torrent::torrenting_system::get_nodes::get_live_node_broadcast_peers; use crate::torrent::torrenting_system::torrent_cache::should_broadcast_torrent; use crate::Arc; use crate::File; @@ -158,8 +158,8 @@ pub async fn broadcast_new_torrent_to_peers( } let torrent_len = 4 + torrent_bytes.len() as u32; - // Send the torrent to all currently connected miner peers. - let peers = get_torrent_broadcast_nodes_from_memory().await; + // Send the torrent to the same live node-peer set used by transaction relay. + let peers = get_live_node_broadcast_peers().await; for (connections_key, stream) in peers { // Each peer gets a short-lived reply mapping so the ack can be drained. let uid_bytes = reserve_transient_entry_with_context( diff --git a/src/torrent/torrenting_system/get_nodes.rs b/src/torrent/torrenting_system/get_nodes.rs index 02722da..b50a573 100644 --- a/src/torrent/torrenting_system/get_nodes.rs +++ b/src/torrent/torrenting_system/get_nodes.rs @@ -56,11 +56,11 @@ pub async fn get_sync_nodes_from_memory() -> Vec<(String, Arc>) nodes } -pub async fn get_torrent_broadcast_nodes_from_memory() -> Vec<(String, Arc>)> { - // Torrent announcements are allowed to reach miner peers that are - // still starting/syncing so they can stage new candidates while - // catching up. Consensus/piece-selection paths must keep using - // get_nodes_from_memory(), which requires ready peers. +pub async fn get_live_node_broadcast_peers() -> Vec<(String, Arc>)> { + // Live node broadcasts go to long-lived node peers that completed the + // wallet-registry and network-map phases. They do not require the peer to + // be fully operational/ready, because syncing peers still need live + // transactions, torrents, and network events staged while catching up. let connection_storage = CONNECTIONS.read().await; let mut nodes = Vec::new(); if let Some(connection) = &*connection_storage { @@ -68,6 +68,9 @@ pub async fn get_torrent_broadcast_nodes_from_memory() -> Vec<(String, Arc