use crate::common::binary_conversions::binary_to_string; use crate::encode; use crate::log::warn; use crate::records::memory::enums::ClientType; use crate::records::memory::response_channels::Command; use crate::rpc::server::command_loop_state::next_incoming_command; use crate::rpc::server::connection_memory_manager::remove_stream_from_memory; use crate::rpc::*; use crate::sled::Db; use crate::wallets::structures::Wallet; use crate::Arc; use crate::AsyncWriteExt; use crate::Instant; use crate::Mutex; use crate::TcpStream; pub async fn start_loop( stream_locked: Arc>, db: Db, connections_key: String, wallet: Arc, map: Arc>, ) -> Result<(), String> { 'outer: loop { let Some(incoming_command) = next_incoming_command(stream_locked.clone(), &db, &connections_key, wallet.clone()) .await? else { break 'outer Ok(()); }; let command = incoming_command.command; let ip = incoming_command.ip; let client_type = incoming_command.client_type; let command_started = Instant::now(); match command { 1 => { // requst network info let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::network_info::request_network_info(&db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 2 => { // request the current block height let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::block_height::request_block_height(&db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 3 => { // request a random node let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::random_node::request_node(&db, &connections_key).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 4 => { // request the current time let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::time::request_time().await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 5 => { // request current block difficulty let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::difficulty::request_difficulty(&db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 6 => { // validate a torrent file let (uid, result) = commands::validate_torrent::validate( &connections_key, stream_locked.clone(), &db, wallet.clone(), ) .await?; result .send(&stream_locked, Some(&connections_key), uid) .await; } 7 => { // request a piece of a block let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let block_number = read_bytes_from_stream::read_u32_from_stream( &connections_key, stream_locked.clone(), ) .await?; let block_piece = read_bytes_from_stream::read_u8_from_stream( &connections_key, stream_locked.clone(), ) .await?; let info_hash = read_bytes_from_stream::read_u128_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::torrent::request_block_piece( &db, block_number, block_piece, info_hash, ) .await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 8 => { // submit a transaction let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let txtype = read_bytes_from_stream::read_u8_from_stream( &connections_key, stream_locked.clone(), ) .await?; let txsize = command_maps::get_bytes(txtype); if txsize == 0 { let result = responses::RpcResponse::Binary( "error: No such transaction type".as_bytes().to_vec(), ); result .send(&stream_locked, Some(&connections_key), uid) .await; continue; } let size = (txsize - 1) as usize; let tx = read_bytes_from_stream::read_usize_from_stream( &connections_key, size, stream_locked.clone(), ) .await?; let result = commands::tx_submit::save_and_submit(txtype, tx, &db, map.clone()).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 9 => { // request a block by block height let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let block_number = read_bytes_from_stream::read_u32_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::block_by_height::request_block(block_number).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 10 => { // request a block by block hash let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let hash = read_bytes_from_stream::read_usize_from_stream( &connections_key, 32, stream_locked.clone(), ) .await?; let hash = encode(hash); let result = commands::block_by_hash::request_block(&db, &hash).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 11 => { // request the latest block let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::latest_block::request_latest_block(&db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 12 => { // request the torrent of a given block let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let block_number = read_bytes_from_stream::read_u32_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::torrent_by_block::request_block_torrent(&block_number).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 13 => { // request the largest txfee let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::largest_tx_fee::request_largest_tx_fee().await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 14 => { // get transaction from mempool by signature let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let signature = read_bytes_from_stream::read_signature_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::memory_by_signature::request_transaction_by_signature(signature) .await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 15 => { // get total count of tranasctions in mempool let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::tx_count_from_mempool::request_tx_count_from_mempool().await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 16 => { // get all transactions of a given address from memory let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let address = read_bytes_from_stream::read_short_address_string_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::transactions_by_address::request_transactions_by_address( &db, &address, ) .await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 17 => { // get a transaction from given txid let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let hash = read_bytes_from_stream::read_usize_from_stream( &connections_key, 32, stream_locked.clone(), ) .await?; let result = commands::transaction_by_txid::request_transaction_by_txid_with_block( &db, hash, ) .await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 18 => { // get total confirmed transaction count let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::tx_count::request_tx_count(&db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 19 => { // get header of given height let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let block_number = read_bytes_from_stream::read_u32_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::block_header_by_height::lookup_by_block_number(&db, block_number) .await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 20 => { // get header of given block hash let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let hash = read_bytes_from_stream::read_usize_from_stream( &connections_key, 32, stream_locked.clone(), ) .await?; let hash = encode(hash); let result = commands::block_header_by_hash::lookup_by_hash(&db, &hash).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 21 => { // get all block headers let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::block_headers::get_all_headers(&db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 22 => { // get balance of address for given coin or token let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let coin = read_bytes_from_stream::read_usize_from_stream( &connections_key, 15, stream_locked.clone(), ) .await?; let coin = binary_to_string(coin); let address = read_bytes_from_stream::read_short_address_string_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::address_coin_lookup::lookup_wallet_coin(&db, address, coin).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 23 => { // get complete balance sheet for a wallet. let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let address = read_bytes_from_stream::read_short_address_string_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::address_complete_balance_sheet::get_token_balances(&db, address) .await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 24 => { // validate wallet address let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let address = read_bytes_from_stream::read_short_address_string_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::validate_address::validate(address, &db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 25 => { // validate signed message let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let message_size = read_bytes_from_stream::read_u16_from_stream( &connections_key, stream_locked.clone(), ) .await?; let message_bytes = read_bytes_from_stream::read_usize_from_stream( &connections_key, message_size as usize, stream_locked.clone(), ) .await?; let message = binary_to_string(message_bytes); let address = read_bytes_from_stream::read_short_address_string_from_stream( &connections_key, stream_locked.clone(), ) .await?; let signature = read_bytes_from_stream::read_signature_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::validate_message::validate(message, signature, address, &db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 26 => { // unblock and ip address let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let ip = read_bytes_from_stream::read_ip_from_stream( &connections_key, stream_locked.clone(), ) .await?; let signature = read_bytes_from_stream::read_signature_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::block_peer_ip::block_peer(&db, ip, signature, wallet.clone()).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 27 => { // block an ip address let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let ip = read_bytes_from_stream::read_ip_from_stream( &connections_key, stream_locked.clone(), ) .await?; let signature = read_bytes_from_stream::read_signature_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::unblock_peer_ip::unblock_peer(&db, ip, signature, wallet.clone()) .await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 28 => { // add a new network node let (uid, result) = commands::add_network_node::add_network_node( &connections_key, stream_locked.clone(), &db, wallet.clone(), map.clone(), ) .await?; result .send(&stream_locked, Some(&connections_key), uid) .await; } 29 => { // delete a new network node let (uid, result) = commands::delete_network_node::delete_network_node( &connections_key, stream_locked.clone(), &db, wallet.clone(), map.clone(), ) .await?; result .send(&stream_locked, Some(&connections_key), uid) .await; } 30 => { // request node list let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::request_valid_nodes::request_valid_nodes().await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 31 => { // request token list let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::token_list::get_tokens(&db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 32 => { // request nft list let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::nft_list::get_nfts(&db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 33 => { // request contract and all payments let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let contract_hash = read_bytes_from_stream::read_usize_from_stream( &connections_key, 32, stream_locked.clone(), ) .await?; let result = commands::contract::contract_details(contract_hash, &db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 34 => { // submit torrent file let (uid, result) = commands::receive_torrent::receive_torrent( &connections_key, stream_locked.clone(), &db, wallet.clone(), map.clone(), ) .await?; result .send(&stream_locked, Some(&connections_key), uid) .await; } 35 => { // request token details let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let token_name = read_bytes_from_stream::read_usize_from_stream( &connections_key, 15, stream_locked.clone(), ) .await?; let token_name = binary_to_string(token_name); let result = commands::token_lookup::lookup_token_details(&db, token_name).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 36 => { // request nft details let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let nft_name = read_bytes_from_stream::read_usize_from_stream( &connections_key, 15, stream_locked.clone(), ) .await?; let nft_name = binary_to_string(nft_name); let series = read_bytes_from_stream::read_u32_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::nft_lookup::lookup_nft_details(&db, nft_name, series).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 37 => { // request contract details by address let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let address = read_bytes_from_stream::read_short_address_string_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::contract::contract_details_by_address(address, &db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 38 => { // register a deterministic short wallet address let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let short_address = read_bytes_from_stream::read_short_address_from_stream( &connections_key, stream_locked.clone(), ) .await?; let public_key = read_bytes_from_stream::read_public_key_from_stream( &connections_key, stream_locked.clone(), ) .await?; let signature = read_bytes_from_stream::read_signature_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::wallet_register::register( short_address, public_key, signature, &db, map.clone(), ip.clone(), connections_key.clone(), ) .await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 39 => { // request the deterministic short wallet registry snapshot let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::wallet_registry_sync::request_registry(&db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 40 => { // request the vanity address registered to a canonical wallet let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let address = read_bytes_from_stream::read_short_address_string_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::wallet_vanity_lookup::lookup(address, &db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 41 => { // request canonical and staged torrent candidates from snapshot onward let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::torrent_candidates::request_torrent_candidates(&db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 42 => { // request the canonical block hash at a specific height let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let block_number = read_bytes_from_stream::read_u32_from_stream( &connections_key, stream_locked.clone(), ) .await?; let result = commands::block_hash_at_height::request_block_hash_at_height(block_number) .await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 45 => { // request the canonical short address that owns a registered vanity address let (uid, _) = read_bytes_from_stream::read_uid_from_stream( &connections_key, stream_locked.clone(), ) .await?; let vanity_bytes = read_bytes_from_stream::read_short_address_from_stream( &connections_key, stream_locked.clone(), ) .await?; let vanity_address = match crate::wallets::structures::Wallet::bytes_to_vanity_address(&vanity_bytes) { Some(address) => address, None => { let result = crate::rpc::responses::RpcResponse::Binary( b"error: Invalid vanity address bytes".to_vec(), ); result .send(&stream_locked, Some(&connections_key), uid) .await; continue; } }; let result = commands::wallet_vanity_owner_lookup::lookup(vanity_address, &db).await; result .send(&stream_locked, Some(&connections_key), uid) .await; } 255 => { commands::route_reply::route_reply( &connections_key, stream_locked.clone(), &db, wallet.clone(), map.clone(), &ip, client_type, ) .await?; } _ => { // Unknown commands are ignored at the protocol level but // still count as bad RPC behavior for scoring purposes. commands::bad_rpc_call::record(&ip, client_type, &db, wallet.clone()).await; } } let elapsed_ms = command_started.elapsed().as_millis(); if elapsed_ms >= 1_000 { warn!( "[rpc_trace] slow handler: cmd={command} peer={connections_key} elapsed_ms={elapsed_ms}" ); } // Wallet-backed client sessions are short-lived request/response // connections, so close them after each non-reply command. if client_type == ClientType::Client && command != command_maps::RPC_REPLY { let mut stream = stream_locked.lock().await; let _ = stream.shutdown().await; drop(stream); remove_stream_from_memory(&stream_locked).await; break 'outer Ok(()); } } }