diff --git a/cli/src/modules/rpc.rs b/cli/src/modules/rpc.rs index c84915480..1f6b0053e 100644 --- a/cli/src/modules/rpc.rs +++ b/cli/src/modules/rpc.rs @@ -42,15 +42,15 @@ impl Rpc { self.println(&ctx, result); } RpcApiOps::GetServerInfo => { - let result = rpc.get_server_info_call(GetServerInfoRequest {}).await?; + let result = rpc.get_server_info_call(None, GetServerInfoRequest {}).await?; self.println(&ctx, result); } RpcApiOps::GetSyncStatus => { - let result = rpc.get_sync_status_call(GetSyncStatusRequest {}).await?; + let result = rpc.get_sync_status_call(None, GetSyncStatusRequest {}).await?; self.println(&ctx, result); } RpcApiOps::GetCurrentNetwork => { - let result = rpc.get_current_network_call(GetCurrentNetworkRequest {}).await?; + let result = rpc.get_current_network_call(None, GetCurrentNetworkRequest {}).await?; self.println(&ctx, result); } // RpcApiOps::SubmitBlock => { @@ -62,11 +62,11 @@ impl Rpc { // self.println(&ctx, result); // } RpcApiOps::GetPeerAddresses => { - let result = rpc.get_peer_addresses_call(GetPeerAddressesRequest {}).await?; + let result = rpc.get_peer_addresses_call(None, GetPeerAddressesRequest {}).await?; self.println(&ctx, result); } RpcApiOps::GetSink => { - let result = rpc.get_sink_call(GetSinkRequest {}).await?; + let result = rpc.get_sink_call(None, GetSinkRequest {}).await?; self.println(&ctx, result); } // RpcApiOps::GetMempoolEntry => { @@ -76,12 +76,15 @@ impl Rpc { RpcApiOps::GetMempoolEntries => { // TODO let result = rpc - .get_mempool_entries_call(GetMempoolEntriesRequest { include_orphan_pool: true, filter_transaction_pool: true }) + .get_mempool_entries_call( + None, + GetMempoolEntriesRequest { include_orphan_pool: true, filter_transaction_pool: true }, + ) .await?; self.println(&ctx, result); } RpcApiOps::GetConnectedPeerInfo => { - let result = rpc.get_connected_peer_info_call(GetConnectedPeerInfoRequest {}).await?; + let result = rpc.get_connected_peer_info_call(None, GetConnectedPeerInfoRequest {}).await?; self.println(&ctx, result); } RpcApiOps::AddPeer => { @@ -90,7 +93,7 @@ impl Rpc { } let peer_address = argv.remove(0).parse::()?; let is_permanent = argv.remove(0).parse::().unwrap_or(false); - let result = rpc.add_peer_call(AddPeerRequest { peer_address, is_permanent }).await?; + let result = rpc.add_peer_call(None, AddPeerRequest { peer_address, is_permanent }).await?; self.println(&ctx, result); } // RpcApiOps::SubmitTransaction => { @@ -103,7 +106,7 @@ impl Rpc { } let hash = argv.remove(0); let hash = RpcHash::from_hex(hash.as_str())?; - let result = rpc.get_block_call(GetBlockRequest { hash, include_transactions: true }).await?; + let result = rpc.get_block_call(None, GetBlockRequest { hash, include_transactions: true }).await?; self.println(&ctx, result); } // RpcApiOps::GetSubnetwork => { @@ -119,11 +122,11 @@ impl Rpc { // self.println(&ctx, result); // } RpcApiOps::GetBlockCount => { - let result = rpc.get_block_count_call(GetBlockCountRequest {}).await?; + let result = rpc.get_block_count_call(None, GetBlockCountRequest {}).await?; self.println(&ctx, result); } RpcApiOps::GetBlockDagInfo => { - let result = rpc.get_block_dag_info_call(GetBlockDagInfoRequest {}).await?; + let result = rpc.get_block_dag_info_call(None, GetBlockDagInfoRequest {}).await?; self.println(&ctx, result); } // RpcApiOps::ResolveFinalityConflict => { @@ -131,7 +134,7 @@ impl Rpc { // self.println(&ctx, result); // } RpcApiOps::Shutdown => { - let result = rpc.shutdown_call(ShutdownRequest {}).await?; + let result = rpc.shutdown_call(None, ShutdownRequest {}).await?; self.println(&ctx, result); } // RpcApiOps::GetHeaders => { @@ -143,7 +146,7 @@ impl Rpc { return Err(Error::custom("Please specify at least one address")); } let addresses = argv.iter().map(|s| Address::try_from(s.as_str())).collect::, _>>()?; - let result = rpc.get_utxos_by_addresses_call(GetUtxosByAddressesRequest { addresses }).await?; + let result = rpc.get_utxos_by_addresses_call(None, GetUtxosByAddressesRequest { addresses }).await?; self.println(&ctx, result); } RpcApiOps::GetBalanceByAddress => { @@ -152,7 +155,7 @@ impl Rpc { } let addresses = argv.iter().map(|s| Address::try_from(s.as_str())).collect::, _>>()?; for address in addresses { - let result = rpc.get_balance_by_address_call(GetBalanceByAddressRequest { address }).await?; + let result = rpc.get_balance_by_address_call(None, GetBalanceByAddressRequest { address }).await?; self.println(&ctx, sompi_to_kaspa(result.balance)); } } @@ -161,11 +164,11 @@ impl Rpc { return Err(Error::custom("Please specify at least one address")); } let addresses = argv.iter().map(|s| Address::try_from(s.as_str())).collect::, _>>()?; - let result = rpc.get_balances_by_addresses_call(GetBalancesByAddressesRequest { addresses }).await?; + let result = rpc.get_balances_by_addresses_call(None, GetBalancesByAddressesRequest { addresses }).await?; self.println(&ctx, result); } RpcApiOps::GetSinkBlueScore => { - let result = rpc.get_sink_blue_score_call(GetSinkBlueScoreRequest {}).await?; + let result = rpc.get_sink_blue_score_call(None, GetSinkBlueScoreRequest {}).await?; self.println(&ctx, result); } RpcApiOps::Ban => { @@ -173,7 +176,7 @@ impl Rpc { return Err(Error::custom("Please specify peer IP address")); } let ip: RpcIpAddress = argv.remove(0).parse()?; - let result = rpc.ban_call(BanRequest { ip }).await?; + let result = rpc.ban_call(None, BanRequest { ip }).await?; self.println(&ctx, result); } RpcApiOps::Unban => { @@ -181,11 +184,11 @@ impl Rpc { return Err(Error::custom("Please specify peer IP address")); } let ip: RpcIpAddress = argv.remove(0).parse()?; - let result = rpc.unban_call(UnbanRequest { ip }).await?; + let result = rpc.unban_call(None, UnbanRequest { ip }).await?; self.println(&ctx, result); } RpcApiOps::GetInfo => { - let result = rpc.get_info_call(GetInfoRequest {}).await?; + let result = rpc.get_info_call(None, GetInfoRequest {}).await?; self.println(&ctx, result); } // RpcApiOps::EstimateNetworkHashesPerSecond => { @@ -200,16 +203,15 @@ impl Rpc { let include_orphan_pool = true; let filter_transaction_pool = true; let result = rpc - .get_mempool_entries_by_addresses_call(GetMempoolEntriesByAddressesRequest { - addresses, - include_orphan_pool, - filter_transaction_pool, - }) + .get_mempool_entries_by_addresses_call( + None, + GetMempoolEntriesByAddressesRequest { addresses, include_orphan_pool, filter_transaction_pool }, + ) .await?; self.println(&ctx, result); } RpcApiOps::GetCoinSupply => { - let result = rpc.get_coin_supply_call(GetCoinSupplyRequest {}).await?; + let result = rpc.get_coin_supply_call(None, GetCoinSupplyRequest {}).await?; self.println(&ctx, result); } RpcApiOps::GetDaaScoreTimestampEstimate => { @@ -220,8 +222,9 @@ impl Rpc { match daa_score_result { Ok(daa_scores) => { - let result = - rpc.get_daa_score_timestamp_estimate_call(GetDaaScoreTimestampEstimateRequest { daa_scores }).await?; + let result = rpc + .get_daa_score_timestamp_estimate_call(None, GetDaaScoreTimestampEstimateRequest { daa_scores }) + .await?; self.println(&ctx, result); } Err(_err) => { diff --git a/rpc/core/src/api/connection.rs b/rpc/core/src/api/connection.rs new file mode 100644 index 000000000..5b4254288 --- /dev/null +++ b/rpc/core/src/api/connection.rs @@ -0,0 +1,7 @@ +use std::sync::Arc; + +pub trait RpcConnection: Send + Sync { + fn id(&self) -> u64; +} + +pub type DynRpcConnection = Arc; diff --git a/rpc/core/src/api/mod.rs b/rpc/core/src/api/mod.rs index 6bc968b46..1373bd6e0 100644 --- a/rpc/core/src/api/mod.rs +++ b/rpc/core/src/api/mod.rs @@ -1,3 +1,4 @@ +pub mod connection; pub mod ctl; pub mod notifications; pub mod ops; diff --git a/rpc/core/src/api/rpc.rs b/rpc/core/src/api/rpc.rs index 36f8ef308..0096afcb5 100644 --- a/rpc/core/src/api/rpc.rs +++ b/rpc/core/src/api/rpc.rs @@ -4,6 +4,7 @@ //! All data provided by the RCP server can be trusted by the client //! No data submitted by the client to the server can be trusted +use crate::api::connection::DynRpcConnection; use crate::{model::*, notify::connection::ChannelConnection, RpcResult}; use async_trait::async_trait; use downcast::{downcast_sync, AnySync}; @@ -21,10 +22,10 @@ pub const MAX_SAFE_WINDOW_SIZE: u32 = 10_000; pub trait RpcApi: Sync + Send + AnySync { /// async fn ping(&self) -> RpcResult<()> { - self.ping_call(PingRequest {}).await?; + self.ping_call(None, PingRequest {}).await?; Ok(()) } - async fn ping_call(&self, request: PingRequest) -> RpcResult; + async fn ping_call(&self, connection: Option<&DynRpcConnection>, request: PingRequest) -> RpcResult; // --- @@ -35,58 +36,87 @@ pub trait RpcApi: Sync + Send + AnySync { bandwidth_metrics: bool, consensus_metrics: bool, ) -> RpcResult { - self.get_metrics_call(GetMetricsRequest { process_metrics, connection_metrics, bandwidth_metrics, consensus_metrics }).await + self.get_metrics_call(None, GetMetricsRequest { process_metrics, connection_metrics, bandwidth_metrics, consensus_metrics }) + .await } - async fn get_metrics_call(&self, request: GetMetricsRequest) -> RpcResult; + async fn get_metrics_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetMetricsRequest, + ) -> RpcResult; // get_info alternative that carries only version, network_id (full), is_synced, virtual_daa_score // these are the only variables needed to negotiate a wRPC connection (besides the wRPC handshake) async fn get_server_info(&self) -> RpcResult { - self.get_server_info_call(GetServerInfoRequest {}).await + self.get_server_info_call(None, GetServerInfoRequest {}).await } - async fn get_server_info_call(&self, request: GetServerInfoRequest) -> RpcResult; + async fn get_server_info_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetServerInfoRequest, + ) -> RpcResult; // Get current sync status of the node (should be converted to a notification subscription) async fn get_sync_status(&self) -> RpcResult { - Ok(self.get_sync_status_call(GetSyncStatusRequest {}).await?.is_synced) + Ok(self.get_sync_status_call(None, GetSyncStatusRequest {}).await?.is_synced) } - async fn get_sync_status_call(&self, request: GetSyncStatusRequest) -> RpcResult; + async fn get_sync_status_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetSyncStatusRequest, + ) -> RpcResult; // --- /// Requests the network the node is currently running against. async fn get_current_network(&self) -> RpcResult { - Ok(self.get_current_network_call(GetCurrentNetworkRequest {}).await?.network) + Ok(self.get_current_network_call(None, GetCurrentNetworkRequest {}).await?.network) } - async fn get_current_network_call(&self, request: GetCurrentNetworkRequest) -> RpcResult; + async fn get_current_network_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetCurrentNetworkRequest, + ) -> RpcResult; /// Submit a block into the DAG. /// /// Blocks are generally expected to have been generated using the get_block_template call. async fn submit_block(&self, block: RpcBlock, allow_non_daa_blocks: bool) -> RpcResult { - self.submit_block_call(SubmitBlockRequest::new(block, allow_non_daa_blocks)).await + self.submit_block_call(None, SubmitBlockRequest::new(block, allow_non_daa_blocks)).await } - async fn submit_block_call(&self, request: SubmitBlockRequest) -> RpcResult; + async fn submit_block_call( + &self, + connection: Option<&DynRpcConnection>, + request: SubmitBlockRequest, + ) -> RpcResult; /// Request a current block template. /// /// Callers are expected to solve the block template and submit it using the submit_block call. async fn get_block_template(&self, pay_address: RpcAddress, extra_data: RpcExtraData) -> RpcResult { - self.get_block_template_call(GetBlockTemplateRequest::new(pay_address, extra_data)).await + self.get_block_template_call(None, GetBlockTemplateRequest::new(pay_address, extra_data)).await } - async fn get_block_template_call(&self, request: GetBlockTemplateRequest) -> RpcResult; + async fn get_block_template_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetBlockTemplateRequest, + ) -> RpcResult; /// Requests the list of known kaspad addresses in the current network (mainnet, testnet, etc.) async fn get_peer_addresses(&self) -> RpcResult { - self.get_peer_addresses_call(GetPeerAddressesRequest {}).await + self.get_peer_addresses_call(None, GetPeerAddressesRequest {}).await } - async fn get_peer_addresses_call(&self, request: GetPeerAddressesRequest) -> RpcResult; + async fn get_peer_addresses_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetPeerAddressesRequest, + ) -> RpcResult; /// requests the hash of the current virtual's selected parent. async fn get_sink(&self) -> RpcResult { - self.get_sink_call(GetSinkRequest {}).await + self.get_sink_call(None, GetSinkRequest {}).await } - async fn get_sink_call(&self, request: GetSinkRequest) -> RpcResult; + async fn get_sink_call(&self, connection: Option<&DynRpcConnection>, request: GetSinkRequest) -> RpcResult; /// Requests information about a specific transaction in the mempool. async fn get_mempool_entry( @@ -96,53 +126,73 @@ pub trait RpcApi: Sync + Send + AnySync { filter_transaction_pool: bool, ) -> RpcResult { Ok(self - .get_mempool_entry_call(GetMempoolEntryRequest::new(transaction_id, include_orphan_pool, filter_transaction_pool)) + .get_mempool_entry_call(None, GetMempoolEntryRequest::new(transaction_id, include_orphan_pool, filter_transaction_pool)) .await? .mempool_entry) } - async fn get_mempool_entry_call(&self, request: GetMempoolEntryRequest) -> RpcResult; + async fn get_mempool_entry_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetMempoolEntryRequest, + ) -> RpcResult; /// Requests information about all the transactions currently in the mempool. async fn get_mempool_entries(&self, include_orphan_pool: bool, filter_transaction_pool: bool) -> RpcResult> { Ok(self - .get_mempool_entries_call(GetMempoolEntriesRequest::new(include_orphan_pool, filter_transaction_pool)) + .get_mempool_entries_call(None, GetMempoolEntriesRequest::new(include_orphan_pool, filter_transaction_pool)) .await? .mempool_entries) } - async fn get_mempool_entries_call(&self, request: GetMempoolEntriesRequest) -> RpcResult; + async fn get_mempool_entries_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetMempoolEntriesRequest, + ) -> RpcResult; /// requests information about all the p2p peers currently connected to this node. async fn get_connected_peer_info(&self) -> RpcResult { - self.get_connected_peer_info_call(GetConnectedPeerInfoRequest {}).await + self.get_connected_peer_info_call(None, GetConnectedPeerInfoRequest {}).await } - async fn get_connected_peer_info_call(&self, request: GetConnectedPeerInfoRequest) -> RpcResult; + async fn get_connected_peer_info_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetConnectedPeerInfoRequest, + ) -> RpcResult; /// Adds a peer to the node's outgoing connection list. /// /// This will, in most cases, result in the node connecting to said peer. async fn add_peer(&self, peer_address: RpcContextualPeerAddress, is_permanent: bool) -> RpcResult<()> { - self.add_peer_call(AddPeerRequest::new(peer_address, is_permanent)).await?; + self.add_peer_call(None, AddPeerRequest::new(peer_address, is_permanent)).await?; Ok(()) } - async fn add_peer_call(&self, request: AddPeerRequest) -> RpcResult; + async fn add_peer_call(&self, connection: Option<&DynRpcConnection>, request: AddPeerRequest) -> RpcResult; /// Submits a transaction to the mempool. async fn submit_transaction(&self, transaction: RpcTransaction, allow_orphan: bool) -> RpcResult { - Ok(self.submit_transaction_call(SubmitTransactionRequest { transaction, allow_orphan }).await?.transaction_id) + Ok(self.submit_transaction_call(None, SubmitTransactionRequest { transaction, allow_orphan }).await?.transaction_id) } - async fn submit_transaction_call(&self, request: SubmitTransactionRequest) -> RpcResult; + async fn submit_transaction_call( + &self, + connection: Option<&DynRpcConnection>, + request: SubmitTransactionRequest, + ) -> RpcResult; /// Requests information about a specific block. async fn get_block(&self, hash: RpcHash, include_transactions: bool) -> RpcResult { - Ok(self.get_block_call(GetBlockRequest::new(hash, include_transactions)).await?.block) + Ok(self.get_block_call(None, GetBlockRequest::new(hash, include_transactions)).await?.block) } - async fn get_block_call(&self, request: GetBlockRequest) -> RpcResult; + async fn get_block_call(&self, connection: Option<&DynRpcConnection>, request: GetBlockRequest) -> RpcResult; /// Requests information about a specific subnetwork. async fn get_subnetwork(&self, subnetwork_id: RpcSubnetworkId) -> RpcResult { - self.get_subnetwork_call(GetSubnetworkRequest::new(subnetwork_id)).await + self.get_subnetwork_call(None, GetSubnetworkRequest::new(subnetwork_id)).await } - async fn get_subnetwork_call(&self, request: GetSubnetworkRequest) -> RpcResult; + async fn get_subnetwork_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetSubnetworkRequest, + ) -> RpcResult; /// Requests the virtual selected parent chain from some `start_hash` to this node's current virtual. async fn get_virtual_chain_from_block( @@ -150,11 +200,15 @@ pub trait RpcApi: Sync + Send + AnySync { start_hash: RpcHash, include_accepted_transaction_ids: bool, ) -> RpcResult { - self.get_virtual_chain_from_block_call(GetVirtualChainFromBlockRequest::new(start_hash, include_accepted_transaction_ids)) - .await + self.get_virtual_chain_from_block_call( + None, + GetVirtualChainFromBlockRequest::new(start_hash, include_accepted_transaction_ids), + ) + .await } async fn get_virtual_chain_from_block_call( &self, + connection: Option<&DynRpcConnection>, request: GetVirtualChainFromBlockRequest, ) -> RpcResult; @@ -165,61 +219,79 @@ pub trait RpcApi: Sync + Send + AnySync { include_blocks: bool, include_transactions: bool, ) -> RpcResult { - self.get_blocks_call(GetBlocksRequest::new(low_hash, include_blocks, include_transactions)).await + self.get_blocks_call(None, GetBlocksRequest::new(low_hash, include_blocks, include_transactions)).await } - async fn get_blocks_call(&self, request: GetBlocksRequest) -> RpcResult; + async fn get_blocks_call(&self, connection: Option<&DynRpcConnection>, request: GetBlocksRequest) -> RpcResult; /// Requests the current number of blocks in this node. /// /// Note that this number may decrease as pruning occurs. async fn get_block_count(&self) -> RpcResult { - self.get_block_count_call(GetBlockCountRequest {}).await + self.get_block_count_call(None, GetBlockCountRequest {}).await } - async fn get_block_count_call(&self, request: GetBlockCountRequest) -> RpcResult; + async fn get_block_count_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetBlockCountRequest, + ) -> RpcResult; /// Requests general information about the current state of this node's DAG. async fn get_block_dag_info(&self) -> RpcResult { - self.get_block_dag_info_call(GetBlockDagInfoRequest {}).await + self.get_block_dag_info_call(None, GetBlockDagInfoRequest {}).await } - async fn get_block_dag_info_call(&self, request: GetBlockDagInfoRequest) -> RpcResult; + async fn get_block_dag_info_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetBlockDagInfoRequest, + ) -> RpcResult; /// async fn resolve_finality_conflict(&self, finality_block_hash: RpcHash) -> RpcResult<()> { - self.resolve_finality_conflict_call(ResolveFinalityConflictRequest::new(finality_block_hash)).await?; + self.resolve_finality_conflict_call(None, ResolveFinalityConflictRequest::new(finality_block_hash)).await?; Ok(()) } async fn resolve_finality_conflict_call( &self, + connection: Option<&DynRpcConnection>, request: ResolveFinalityConflictRequest, ) -> RpcResult; /// Shuts down this node. async fn shutdown(&self) -> RpcResult<()> { - self.shutdown_call(ShutdownRequest {}).await?; + self.shutdown_call(None, ShutdownRequest {}).await?; Ok(()) } - async fn shutdown_call(&self, request: ShutdownRequest) -> RpcResult; + async fn shutdown_call(&self, connection: Option<&DynRpcConnection>, request: ShutdownRequest) -> RpcResult; /// Requests headers between the given `start_hash` and the current virtual, up to the given limit. async fn get_headers(&self, start_hash: RpcHash, limit: u64, is_ascending: bool) -> RpcResult> { - Ok(self.get_headers_call(GetHeadersRequest::new(start_hash, limit, is_ascending)).await?.headers) + Ok(self.get_headers_call(None, GetHeadersRequest::new(start_hash, limit, is_ascending)).await?.headers) } - async fn get_headers_call(&self, request: GetHeadersRequest) -> RpcResult; + async fn get_headers_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetHeadersRequest, + ) -> RpcResult; /// Returns the total balance in unspent transactions towards a given address. /// /// This call is only available when this node was started with `--utxoindex`. async fn get_balance_by_address(&self, address: RpcAddress) -> RpcResult { - Ok(self.get_balance_by_address_call(GetBalanceByAddressRequest::new(address)).await?.balance) + Ok(self.get_balance_by_address_call(None, GetBalanceByAddressRequest::new(address)).await?.balance) } - async fn get_balance_by_address_call(&self, request: GetBalanceByAddressRequest) -> RpcResult; + async fn get_balance_by_address_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetBalanceByAddressRequest, + ) -> RpcResult; /// async fn get_balances_by_addresses(&self, addresses: Vec) -> RpcResult> { - Ok(self.get_balances_by_addresses_call(GetBalancesByAddressesRequest::new(addresses)).await?.entries) + Ok(self.get_balances_by_addresses_call(None, GetBalancesByAddressesRequest::new(addresses)).await?.entries) } async fn get_balances_by_addresses_call( &self, + connection: Option<&DynRpcConnection>, request: GetBalancesByAddressesRequest, ) -> RpcResult; @@ -227,45 +299,54 @@ pub trait RpcApi: Sync + Send + AnySync { /// /// This call is only available when this node was started with `--utxoindex`. async fn get_utxos_by_addresses(&self, addresses: Vec) -> RpcResult> { - Ok(self.get_utxos_by_addresses_call(GetUtxosByAddressesRequest::new(addresses)).await?.entries) + Ok(self.get_utxos_by_addresses_call(None, GetUtxosByAddressesRequest::new(addresses)).await?.entries) } - async fn get_utxos_by_addresses_call(&self, request: GetUtxosByAddressesRequest) -> RpcResult; + async fn get_utxos_by_addresses_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetUtxosByAddressesRequest, + ) -> RpcResult; /// Requests the blue score of the current selected parent of the virtual block. async fn get_sink_blue_score(&self) -> RpcResult { - Ok(self.get_sink_blue_score_call(GetSinkBlueScoreRequest {}).await?.blue_score) + Ok(self.get_sink_blue_score_call(None, GetSinkBlueScoreRequest {}).await?.blue_score) } - async fn get_sink_blue_score_call(&self, request: GetSinkBlueScoreRequest) -> RpcResult; + async fn get_sink_blue_score_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetSinkBlueScoreRequest, + ) -> RpcResult; /// Bans the given ip. async fn ban(&self, ip: RpcIpAddress) -> RpcResult<()> { - self.ban_call(BanRequest::new(ip)).await?; + self.ban_call(None, BanRequest::new(ip)).await?; Ok(()) } - async fn ban_call(&self, request: BanRequest) -> RpcResult; + async fn ban_call(&self, connection: Option<&DynRpcConnection>, request: BanRequest) -> RpcResult; /// Unbans the given ip. async fn unban(&self, ip: RpcIpAddress) -> RpcResult<()> { - self.unban_call(UnbanRequest::new(ip)).await?; + self.unban_call(None, UnbanRequest::new(ip)).await?; Ok(()) } - async fn unban_call(&self, request: UnbanRequest) -> RpcResult; + async fn unban_call(&self, connection: Option<&DynRpcConnection>, request: UnbanRequest) -> RpcResult; /// Returns info about the node. - async fn get_info_call(&self, request: GetInfoRequest) -> RpcResult; async fn get_info(&self) -> RpcResult { - self.get_info_call(GetInfoRequest {}).await + self.get_info_call(None, GetInfoRequest {}).await } + async fn get_info_call(&self, connection: Option<&DynRpcConnection>, request: GetInfoRequest) -> RpcResult; /// async fn estimate_network_hashes_per_second(&self, window_size: u32, start_hash: Option) -> RpcResult { Ok(self - .estimate_network_hashes_per_second_call(EstimateNetworkHashesPerSecondRequest::new(window_size, start_hash)) + .estimate_network_hashes_per_second_call(None, EstimateNetworkHashesPerSecondRequest::new(window_size, start_hash)) .await? .network_hashes_per_second) } async fn estimate_network_hashes_per_second_call( &self, + connection: Option<&DynRpcConnection>, request: EstimateNetworkHashesPerSecondRequest, ) -> RpcResult; @@ -277,30 +358,35 @@ pub trait RpcApi: Sync + Send + AnySync { filter_transaction_pool: bool, ) -> RpcResult> { Ok(self - .get_mempool_entries_by_addresses_call(GetMempoolEntriesByAddressesRequest::new( - addresses, - include_orphan_pool, - filter_transaction_pool, - )) + .get_mempool_entries_by_addresses_call( + None, + GetMempoolEntriesByAddressesRequest::new(addresses, include_orphan_pool, filter_transaction_pool), + ) .await? .entries) } async fn get_mempool_entries_by_addresses_call( &self, + connection: Option<&DynRpcConnection>, request: GetMempoolEntriesByAddressesRequest, ) -> RpcResult; /// async fn get_coin_supply(&self) -> RpcResult { - self.get_coin_supply_call(GetCoinSupplyRequest {}).await + self.get_coin_supply_call(None, GetCoinSupplyRequest {}).await } - async fn get_coin_supply_call(&self, request: GetCoinSupplyRequest) -> RpcResult; + async fn get_coin_supply_call( + &self, + connection: Option<&DynRpcConnection>, + request: GetCoinSupplyRequest, + ) -> RpcResult; async fn get_daa_score_timestamp_estimate(&self, daa_scores: Vec) -> RpcResult> { - Ok(self.get_daa_score_timestamp_estimate_call(GetDaaScoreTimestampEstimateRequest { daa_scores }).await?.timestamps) + Ok(self.get_daa_score_timestamp_estimate_call(None, GetDaaScoreTimestampEstimateRequest { daa_scores }).await?.timestamps) } async fn get_daa_score_timestamp_estimate_call( &self, + connection: Option<&DynRpcConnection>, request: GetDaaScoreTimestampEstimateRequest, ) -> RpcResult; diff --git a/rpc/grpc/client/src/route.rs b/rpc/grpc/client/src/route.rs index 5bb1bf395..bb5b5ce56 100644 --- a/rpc/grpc/client/src/route.rs +++ b/rpc/grpc/client/src/route.rs @@ -9,12 +9,14 @@ macro_rules! route { clippy::type_repetition_in_bounds, clippy::used_underscore_binding )] - fn $fn<'life0, 'async_trait>( + fn $fn<'life0, 'life1, 'async_trait>( &'life0 self, + _connection : ::core::option::Option<&'life1 Arc>, request: [<$name Request>], ) -> ::core::pin::Pin]>> + ::core::marker::Send + 'async_trait>> where 'life0: 'async_trait, + 'life1: 'async_trait, Self: 'async_trait, { Box::pin(async move { diff --git a/rpc/grpc/server/src/tests/rpc_core_mock.rs b/rpc/grpc/server/src/tests/rpc_core_mock.rs index ddf78ccbd..e6abf8aa3 100644 --- a/rpc/grpc/server/src/tests/rpc_core_mock.rs +++ b/rpc/grpc/server/src/tests/rpc_core_mock.rs @@ -6,7 +6,7 @@ use kaspa_notify::notifier::{Notifier, Notify}; use kaspa_notify::scope::Scope; use kaspa_notify::subscription::context::SubscriptionContext; use kaspa_notify::subscription::{MutationPolicies, UtxosChangedMutationPolicy}; -use kaspa_rpc_core::{api::rpc::RpcApi, *}; +use kaspa_rpc_core::{api::connection::DynRpcConnection, api::rpc::RpcApi, *}; use kaspa_rpc_core::{notify::connection::ChannelConnection, RpcResult}; use std::sync::Arc; @@ -66,7 +66,7 @@ impl RpcCoreMock { #[async_trait] impl RpcApi for RpcCoreMock { // This fn needs to succeed while the client connects - async fn get_info_call(&self, _request: GetInfoRequest) -> RpcResult { + async fn get_info_call(&self, _connection: Option<&DynRpcConnection>, _request: GetInfoRequest) -> RpcResult { Ok(GetInfoResponse { p2p_id: "p2p-mock".to_string(), mempool_size: 1234, @@ -78,133 +78,213 @@ impl RpcApi for RpcCoreMock { }) } - async fn ping_call(&self, _request: PingRequest) -> RpcResult { + async fn ping_call(&self, _connection: Option<&DynRpcConnection>, _request: PingRequest) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_metrics_call(&self, _request: GetMetricsRequest) -> RpcResult { + async fn get_metrics_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetMetricsRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_server_info_call(&self, _request: GetServerInfoRequest) -> RpcResult { + async fn get_server_info_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetServerInfoRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_sync_status_call(&self, _request: GetSyncStatusRequest) -> RpcResult { + async fn get_sync_status_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetSyncStatusRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_current_network_call(&self, _request: GetCurrentNetworkRequest) -> RpcResult { + async fn get_current_network_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetCurrentNetworkRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn submit_block_call(&self, _request: SubmitBlockRequest) -> RpcResult { + async fn submit_block_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: SubmitBlockRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_block_template_call(&self, _request: GetBlockTemplateRequest) -> RpcResult { + async fn get_block_template_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetBlockTemplateRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_peer_addresses_call(&self, _request: GetPeerAddressesRequest) -> RpcResult { + async fn get_peer_addresses_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetPeerAddressesRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_sink_call(&self, _request: GetSinkRequest) -> RpcResult { + async fn get_sink_call(&self, _connection: Option<&DynRpcConnection>, _request: GetSinkRequest) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_mempool_entry_call(&self, _request: GetMempoolEntryRequest) -> RpcResult { + async fn get_mempool_entry_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetMempoolEntryRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_mempool_entries_call(&self, _request: GetMempoolEntriesRequest) -> RpcResult { + async fn get_mempool_entries_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetMempoolEntriesRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_connected_peer_info_call(&self, _request: GetConnectedPeerInfoRequest) -> RpcResult { + async fn get_connected_peer_info_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetConnectedPeerInfoRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn add_peer_call(&self, _request: AddPeerRequest) -> RpcResult { + async fn add_peer_call(&self, _connection: Option<&DynRpcConnection>, _request: AddPeerRequest) -> RpcResult { Err(RpcError::NotImplemented) } - async fn submit_transaction_call(&self, _request: SubmitTransactionRequest) -> RpcResult { + async fn submit_transaction_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: SubmitTransactionRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_block_call(&self, _request: GetBlockRequest) -> RpcResult { + async fn get_block_call(&self, _connection: Option<&DynRpcConnection>, _request: GetBlockRequest) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_subnetwork_call(&self, _request: GetSubnetworkRequest) -> RpcResult { + async fn get_subnetwork_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetSubnetworkRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } async fn get_virtual_chain_from_block_call( &self, + _connection: Option<&DynRpcConnection>, _request: GetVirtualChainFromBlockRequest, ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_blocks_call(&self, _request: GetBlocksRequest) -> RpcResult { + async fn get_blocks_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetBlocksRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_block_count_call(&self, _request: GetBlockCountRequest) -> RpcResult { + async fn get_block_count_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetBlockCountRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_block_dag_info_call(&self, _request: GetBlockDagInfoRequest) -> RpcResult { + async fn get_block_dag_info_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetBlockDagInfoRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } async fn resolve_finality_conflict_call( &self, + _connection: Option<&DynRpcConnection>, _request: ResolveFinalityConflictRequest, ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn shutdown_call(&self, _request: ShutdownRequest) -> RpcResult { + async fn shutdown_call(&self, _connection: Option<&DynRpcConnection>, _request: ShutdownRequest) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_headers_call(&self, _request: GetHeadersRequest) -> RpcResult { + async fn get_headers_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetHeadersRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_balance_by_address_call(&self, _request: GetBalanceByAddressRequest) -> RpcResult { + async fn get_balance_by_address_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetBalanceByAddressRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } async fn get_balances_by_addresses_call( &self, + _connection: Option<&DynRpcConnection>, _request: GetBalancesByAddressesRequest, ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_utxos_by_addresses_call(&self, _request: GetUtxosByAddressesRequest) -> RpcResult { + async fn get_utxos_by_addresses_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetUtxosByAddressesRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_sink_blue_score_call(&self, _request: GetSinkBlueScoreRequest) -> RpcResult { + async fn get_sink_blue_score_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetSinkBlueScoreRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn ban_call(&self, _request: BanRequest) -> RpcResult { + async fn ban_call(&self, _connection: Option<&DynRpcConnection>, _request: BanRequest) -> RpcResult { Err(RpcError::NotImplemented) } - async fn unban_call(&self, _request: UnbanRequest) -> RpcResult { + async fn unban_call(&self, _connection: Option<&DynRpcConnection>, _request: UnbanRequest) -> RpcResult { Err(RpcError::NotImplemented) } async fn estimate_network_hashes_per_second_call( &self, + _connection: Option<&DynRpcConnection>, _request: EstimateNetworkHashesPerSecondRequest, ) -> RpcResult { Err(RpcError::NotImplemented) @@ -212,17 +292,23 @@ impl RpcApi for RpcCoreMock { async fn get_mempool_entries_by_addresses_call( &self, + _connection: Option<&DynRpcConnection>, _request: GetMempoolEntriesByAddressesRequest, ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_coin_supply_call(&self, _request: GetCoinSupplyRequest) -> RpcResult { + async fn get_coin_supply_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetCoinSupplyRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } async fn get_daa_score_timestamp_estimate_call( &self, + _connection: Option<&DynRpcConnection>, _request: GetDaaScoreTimestampEstimateRequest, ) -> RpcResult { Err(RpcError::NotImplemented) diff --git a/rpc/macros/src/grpc/server.rs b/rpc/macros/src/grpc/server.rs index 91dea1dd9..f6f3f33b9 100644 --- a/rpc/macros/src/grpc/server.rs +++ b/rpc/macros/src/grpc/server.rs @@ -72,7 +72,8 @@ impl ToTokens for RpcTable { Box::pin(async move { let mut response: #kaspad_response_type = match request.payload { Some(Payload::#request_type(ref request)) => match request.try_into() { - Ok(request) => server_ctx.core_service.#fn_call(request).await.into(), + // TODO: RPC-CONNECTION + Ok(request) => server_ctx.core_service.#fn_call(None,request).await.into(), Err(err) => #response_message_type::from(err).into(), }, _ => { diff --git a/rpc/macros/src/wrpc/client.rs b/rpc/macros/src/wrpc/client.rs index 49c09c491..12f41687a 100644 --- a/rpc/macros/src/wrpc/client.rs +++ b/rpc/macros/src/wrpc/client.rs @@ -52,12 +52,14 @@ impl ToTokens for RpcTable { // the async implementation of the RPC caller is inlined targets.push(quote! { - fn #fn_call<'life0, 'async_trait>( + fn #fn_call<'life0, 'life1, 'async_trait>( &'life0 self, + _connection : ::core::option::Option<&'life1 Arc>, request: #request_type, ) -> ::core::pin::Pin> + ::core::marker::Send + 'async_trait>> where 'life0: 'async_trait, + 'life1: 'async_trait, Self: 'async_trait, { use workflow_serializer::prelude::*; diff --git a/rpc/macros/src/wrpc/server.rs b/rpc/macros/src/wrpc/server.rs index dfb1491d8..ee2afc551 100644 --- a/rpc/macros/src/wrpc/server.rs +++ b/rpc/macros/src/wrpc/server.rs @@ -53,7 +53,8 @@ impl ToTokens for RpcTable { interface.method(#rpc_api_ops::#handler, method!(|server_ctx: #server_ctx_type, connection_ctx: #connection_ctx_type, request: Serializable<#request_type>| async move { let verbose = server_ctx.verbose(); if verbose { workflow_log::log_info!("request: {:?}",request); } - let response: #response_type = server_ctx.rpc_service(&connection_ctx).#fn_call(request.into_inner()).await + // TODO: RPC-CONNECT + let response: #response_type = server_ctx.rpc_service(&connection_ctx).#fn_call(None, request.into_inner()).await .map_err(|e|ServerError::Text(e.to_string()))?; if verbose { workflow_log::log_info!("response: {:?}",response); } Ok(Serializable(response)) diff --git a/rpc/macros/src/wrpc/wasm.rs b/rpc/macros/src/wrpc/wasm.rs index 011822019..30af3e74a 100644 --- a/rpc/macros/src/wrpc/wasm.rs +++ b/rpc/macros/src/wrpc/wasm.rs @@ -59,7 +59,7 @@ impl ToTokens for RpcHandlers { pub async fn #fn_no_suffix(&self, request : Option<#ts_request_type>) -> Result<#ts_response_type> { let request: #request_type = request.unwrap_or_default().try_into()?; // log_info!("request: {:#?}",request); - let result: RpcResult<#response_type> = self.inner.client.#fn_call(request).await; + let result: RpcResult<#response_type> = self.inner.client.#fn_call(None, request).await; // log_info!("result: {:#?}",result); let response: #response_type = result.map_err(|err|wasm_bindgen::JsError::new(&err.to_string()))?; //log_info!("response: {:#?}",response); @@ -83,7 +83,7 @@ impl ToTokens for RpcHandlers { #[wasm_bindgen(js_name = #fn_camel)] pub async fn #fn_no_suffix(&self, request: #ts_request_type) -> Result<#ts_response_type> { let request: #request_type = request.try_into()?; - let result: RpcResult<#response_type> = self.inner.client.#fn_call(request).await; + let result: RpcResult<#response_type> = self.inner.client.#fn_call(None, request).await; let response: #response_type = result.map_err(|err|wasm_bindgen::JsError::new(&err.to_string()))?; Ok(response.try_into()?) } diff --git a/rpc/service/src/service.rs b/rpc/service/src/service.rs index 00cc0d082..c1beff244 100644 --- a/rpc/service/src/service.rs +++ b/rpc/service/src/service.rs @@ -53,6 +53,7 @@ use kaspa_p2p_lib::common::ProtocolError; use kaspa_perf_monitor::{counters::CountersSnapshot, Monitor as PerfMonitor}; use kaspa_rpc_core::{ api::{ + connection::DynRpcConnection, ops::RPC_API_VERSION, rpc::{RpcApi, MAX_SAFE_WINDOW_SIZE}, }, @@ -275,7 +276,11 @@ impl RpcCoreService { #[async_trait] impl RpcApi for RpcCoreService { - async fn submit_block_call(&self, request: SubmitBlockRequest) -> RpcResult { + async fn submit_block_call( + &self, + _connection: Option<&DynRpcConnection>, + request: SubmitBlockRequest, + ) -> RpcResult { let session = self.consensus_manager.consensus().unguarded_session(); // TODO: consider adding an error field to SubmitBlockReport to document both the report and error fields @@ -333,7 +338,11 @@ NOTE: This error usually indicates an RPC conversion error between the node and } } - async fn get_block_template_call(&self, request: GetBlockTemplateRequest) -> RpcResult { + async fn get_block_template_call( + &self, + _connection: Option<&DynRpcConnection>, + request: GetBlockTemplateRequest, + ) -> RpcResult { trace!("incoming GetBlockTemplate request"); if *self.config.net == NetworkType::Mainnet && !self.config.enable_mainnet_mining { @@ -365,7 +374,7 @@ NOTE: This error usually indicates an RPC conversion error between the node and }) } - async fn get_block_call(&self, request: GetBlockRequest) -> RpcResult { + async fn get_block_call(&self, _connection: Option<&DynRpcConnection>, request: GetBlockRequest) -> RpcResult { // TODO: test let session = self.consensus_manager.consensus().session().await; let block = session.async_get_block_even_if_header_only(request.hash).await?; @@ -377,7 +386,11 @@ NOTE: This error usually indicates an RPC conversion error between the node and }) } - async fn get_blocks_call(&self, request: GetBlocksRequest) -> RpcResult { + async fn get_blocks_call( + &self, + _connection: Option<&DynRpcConnection>, + request: GetBlocksRequest, + ) -> RpcResult { // Validate that user didn't set include_transactions without setting include_blocks if !request.include_blocks && request.include_transactions { return Err(RpcError::InvalidGetBlocksRequest); @@ -426,7 +439,7 @@ NOTE: This error usually indicates an RPC conversion error between the node and Ok(GetBlocksResponse { block_hashes, blocks }) } - async fn get_info_call(&self, _request: GetInfoRequest) -> RpcResult { + async fn get_info_call(&self, _connection: Option<&DynRpcConnection>, _request: GetInfoRequest) -> RpcResult { let is_nearly_synced = self.consensus_manager.consensus().unguarded_session().async_is_nearly_synced().await; Ok(GetInfoResponse { p2p_id: self.flow_context.node_id.to_string(), @@ -439,7 +452,11 @@ NOTE: This error usually indicates an RPC conversion error between the node and }) } - async fn get_mempool_entry_call(&self, request: GetMempoolEntryRequest) -> RpcResult { + async fn get_mempool_entry_call( + &self, + _connection: Option<&DynRpcConnection>, + request: GetMempoolEntryRequest, + ) -> RpcResult { let query = self.extract_tx_query(request.filter_transaction_pool, request.include_orphan_pool)?; let Some(transaction) = self.mining_manager.clone().get_transaction(request.transaction_id, query).await else { return Err(RpcError::TransactionNotFound(request.transaction_id)); @@ -448,7 +465,11 @@ NOTE: This error usually indicates an RPC conversion error between the node and Ok(GetMempoolEntryResponse::new(self.consensus_converter.get_mempool_entry(&session, &transaction))) } - async fn get_mempool_entries_call(&self, request: GetMempoolEntriesRequest) -> RpcResult { + async fn get_mempool_entries_call( + &self, + _connection: Option<&DynRpcConnection>, + request: GetMempoolEntriesRequest, + ) -> RpcResult { let query = self.extract_tx_query(request.filter_transaction_pool, request.include_orphan_pool)?; let session = self.consensus_manager.consensus().unguarded_session(); let (transactions, orphans) = self.mining_manager.clone().get_all_transactions(query).await; @@ -462,6 +483,7 @@ NOTE: This error usually indicates an RPC conversion error between the node and async fn get_mempool_entries_by_addresses_call( &self, + _connection: Option<&DynRpcConnection>, request: GetMempoolEntriesByAddressesRequest, ) -> RpcResult { let query = self.extract_tx_query(request.filter_transaction_pool, request.include_orphan_pool)?; @@ -485,7 +507,11 @@ NOTE: This error usually indicates an RPC conversion error between the node and Ok(GetMempoolEntriesByAddressesResponse::new(mempool_entries)) } - async fn submit_transaction_call(&self, request: SubmitTransactionRequest) -> RpcResult { + async fn submit_transaction_call( + &self, + _connection: Option<&DynRpcConnection>, + request: SubmitTransactionRequest, + ) -> RpcResult { let allow_orphan = self.config.unsafe_rpc && request.allow_orphan; if !self.config.unsafe_rpc && request.allow_orphan { warn!("SubmitTransaction RPC command called with AllowOrphan enabled while node in safe RPC mode -- switching to ForbidOrphan."); @@ -506,25 +532,38 @@ NOTE: This error usually indicates an RPC conversion error between the node and Ok(SubmitTransactionResponse::new(transaction_id)) } - async fn get_current_network_call(&self, _: GetCurrentNetworkRequest) -> RpcResult { + async fn get_current_network_call( + &self, + _connection: Option<&DynRpcConnection>, + _: GetCurrentNetworkRequest, + ) -> RpcResult { Ok(GetCurrentNetworkResponse::new(*self.config.net)) } - async fn get_subnetwork_call(&self, _: GetSubnetworkRequest) -> RpcResult { + async fn get_subnetwork_call( + &self, + _connection: Option<&DynRpcConnection>, + _: GetSubnetworkRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_sink_call(&self, _: GetSinkRequest) -> RpcResult { + async fn get_sink_call(&self, _connection: Option<&DynRpcConnection>, _: GetSinkRequest) -> RpcResult { Ok(GetSinkResponse::new(self.consensus_manager.consensus().unguarded_session().async_get_sink().await)) } - async fn get_sink_blue_score_call(&self, _: GetSinkBlueScoreRequest) -> RpcResult { + async fn get_sink_blue_score_call( + &self, + _connection: Option<&DynRpcConnection>, + _: GetSinkBlueScoreRequest, + ) -> RpcResult { let session = self.consensus_manager.consensus().unguarded_session(); Ok(GetSinkBlueScoreResponse::new(session.async_get_ghostdag_data(session.async_get_sink().await).await?.blue_score)) } async fn get_virtual_chain_from_block_call( &self, + _connection: Option<&DynRpcConnection>, request: GetVirtualChainFromBlockRequest, ) -> RpcResult { let session = self.consensus_manager.consensus().session().await; @@ -537,11 +576,19 @@ NOTE: This error usually indicates an RPC conversion error between the node and Ok(GetVirtualChainFromBlockResponse::new(virtual_chain.removed, virtual_chain.added, accepted_transaction_ids)) } - async fn get_block_count_call(&self, _: GetBlockCountRequest) -> RpcResult { + async fn get_block_count_call( + &self, + _connection: Option<&DynRpcConnection>, + _: GetBlockCountRequest, + ) -> RpcResult { Ok(self.consensus_manager.consensus().unguarded_session().async_estimate_block_count().await) } - async fn get_utxos_by_addresses_call(&self, request: GetUtxosByAddressesRequest) -> RpcResult { + async fn get_utxos_by_addresses_call( + &self, + _connection: Option<&DynRpcConnection>, + request: GetUtxosByAddressesRequest, + ) -> RpcResult { if !self.config.utxoindex { return Err(RpcError::NoUtxoIndex); } @@ -551,7 +598,11 @@ NOTE: This error usually indicates an RPC conversion error between the node and Ok(GetUtxosByAddressesResponse::new(self.index_converter.get_utxos_by_addresses_entries(&entry_map))) } - async fn get_balance_by_address_call(&self, request: GetBalanceByAddressRequest) -> RpcResult { + async fn get_balance_by_address_call( + &self, + _connection: Option<&DynRpcConnection>, + request: GetBalanceByAddressRequest, + ) -> RpcResult { if !self.config.utxoindex { return Err(RpcError::NoUtxoIndex); } @@ -562,6 +613,7 @@ NOTE: This error usually indicates an RPC conversion error between the node and async fn get_balances_by_addresses_call( &self, + _connection: Option<&DynRpcConnection>, request: GetBalancesByAddressesRequest, ) -> RpcResult { if !self.config.utxoindex { @@ -580,7 +632,11 @@ NOTE: This error usually indicates an RPC conversion error between the node and Ok(GetBalancesByAddressesResponse::new(entries)) } - async fn get_coin_supply_call(&self, _: GetCoinSupplyRequest) -> RpcResult { + async fn get_coin_supply_call( + &self, + _connection: Option<&DynRpcConnection>, + _: GetCoinSupplyRequest, + ) -> RpcResult { if !self.config.utxoindex { return Err(RpcError::NoUtxoIndex); } @@ -591,6 +647,7 @@ NOTE: This error usually indicates an RPC conversion error between the node and async fn get_daa_score_timestamp_estimate_call( &self, + _connection: Option<&DynRpcConnection>, request: GetDaaScoreTimestampEstimateRequest, ) -> RpcResult { let session = self.consensus_manager.consensus().session().await; @@ -647,15 +704,23 @@ NOTE: This error usually indicates an RPC conversion error between the node and Ok(GetDaaScoreTimestampEstimateResponse::new(timestamps)) } - async fn ping_call(&self, _: PingRequest) -> RpcResult { + async fn ping_call(&self, _connection: Option<&DynRpcConnection>, _: PingRequest) -> RpcResult { Ok(PingResponse {}) } - async fn get_headers_call(&self, _request: GetHeadersRequest) -> RpcResult { + async fn get_headers_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetHeadersRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_block_dag_info_call(&self, _: GetBlockDagInfoRequest) -> RpcResult { + async fn get_block_dag_info_call( + &self, + _connection: Option<&DynRpcConnection>, + _: GetBlockDagInfoRequest, + ) -> RpcResult { let session = self.consensus_manager.consensus().unguarded_session(); let (consensus_stats, tips, pruning_point, sink) = join!(session.async_get_stats(), session.async_get_tips(), session.async_pruning_point(), session.async_get_sink()); @@ -675,6 +740,7 @@ NOTE: This error usually indicates an RPC conversion error between the node and async fn estimate_network_hashes_per_second_call( &self, + _connection: Option<&DynRpcConnection>, request: EstimateNetworkHashesPerSecondRequest, ) -> RpcResult { if !self.config.unsafe_rpc && request.window_size > MAX_SAFE_WINDOW_SIZE { @@ -704,7 +770,7 @@ NOTE: This error usually indicates an RPC conversion error between the node and )) } - async fn add_peer_call(&self, request: AddPeerRequest) -> RpcResult { + async fn add_peer_call(&self, _connection: Option<&DynRpcConnection>, request: AddPeerRequest) -> RpcResult { if !self.config.unsafe_rpc { warn!("AddPeer RPC command called while node in safe RPC mode -- ignoring."); return Err(RpcError::UnavailableInSafeMode); @@ -718,12 +784,16 @@ NOTE: This error usually indicates an RPC conversion error between the node and Ok(AddPeerResponse {}) } - async fn get_peer_addresses_call(&self, _: GetPeerAddressesRequest) -> RpcResult { + async fn get_peer_addresses_call( + &self, + _connection: Option<&DynRpcConnection>, + _: GetPeerAddressesRequest, + ) -> RpcResult { let address_manager = self.flow_context.address_manager.lock(); Ok(GetPeerAddressesResponse::new(address_manager.get_all_addresses(), address_manager.get_all_banned_addresses())) } - async fn ban_call(&self, request: BanRequest) -> RpcResult { + async fn ban_call(&self, _connection: Option<&DynRpcConnection>, request: BanRequest) -> RpcResult { if !self.config.unsafe_rpc { warn!("Ban RPC command called while node in safe RPC mode -- ignoring."); return Err(RpcError::UnavailableInSafeMode); @@ -740,7 +810,7 @@ NOTE: This error usually indicates an RPC conversion error between the node and Ok(BanResponse {}) } - async fn unban_call(&self, request: UnbanRequest) -> RpcResult { + async fn unban_call(&self, _connection: Option<&DynRpcConnection>, request: UnbanRequest) -> RpcResult { if !self.config.unsafe_rpc { warn!("Unban RPC command called while node in safe RPC mode -- ignoring."); return Err(RpcError::UnavailableInSafeMode); @@ -754,13 +824,17 @@ NOTE: This error usually indicates an RPC conversion error between the node and Ok(UnbanResponse {}) } - async fn get_connected_peer_info_call(&self, _: GetConnectedPeerInfoRequest) -> RpcResult { + async fn get_connected_peer_info_call( + &self, + _connection: Option<&DynRpcConnection>, + _: GetConnectedPeerInfoRequest, + ) -> RpcResult { let peers = self.flow_context.hub().active_peers(); let peer_info = self.protocol_converter.get_peers_info(&peers); Ok(GetConnectedPeerInfoResponse::new(peer_info)) } - async fn shutdown_call(&self, _: ShutdownRequest) -> RpcResult { + async fn shutdown_call(&self, _connection: Option<&DynRpcConnection>, _: ShutdownRequest) -> RpcResult { if !self.config.unsafe_rpc { warn!("Shutdown RPC command called while node in safe RPC mode -- ignoring."); return Err(RpcError::UnavailableInSafeMode); @@ -783,6 +857,7 @@ NOTE: This error usually indicates an RPC conversion error between the node and async fn resolve_finality_conflict_call( &self, + _connection: Option<&DynRpcConnection>, _request: ResolveFinalityConflictRequest, ) -> RpcResult { if !self.config.unsafe_rpc { @@ -792,7 +867,7 @@ NOTE: This error usually indicates an RPC conversion error between the node and Err(RpcError::NotImplemented) } - async fn get_metrics_call(&self, req: GetMetricsRequest) -> RpcResult { + async fn get_metrics_call(&self, _connection: Option<&DynRpcConnection>, req: GetMetricsRequest) -> RpcResult { let CountersSnapshot { resident_set_size, virtual_memory_size, @@ -873,7 +948,11 @@ NOTE: This error usually indicates an RPC conversion error between the node and Ok(response) } - async fn get_server_info_call(&self, _request: GetServerInfoRequest) -> RpcResult { + async fn get_server_info_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetServerInfoRequest, + ) -> RpcResult { let session = self.consensus_manager.consensus().unguarded_session(); let is_synced: bool = self.has_sufficient_peer_connectivity() && session.async_is_nearly_synced().await; let virtual_daa_score = session.get_virtual_daa_score(); @@ -888,7 +967,11 @@ NOTE: This error usually indicates an RPC conversion error between the node and }) } - async fn get_sync_status_call(&self, _request: GetSyncStatusRequest) -> RpcResult { + async fn get_sync_status_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetSyncStatusRequest, + ) -> RpcResult { let session = self.consensus_manager.consensus().unguarded_session(); let is_synced: bool = self.has_sufficient_peer_connectivity() && session.async_is_nearly_synced().await; Ok(GetSyncStatusResponse { is_synced }) diff --git a/testing/integration/src/rpc_tests.rs b/testing/integration/src/rpc_tests.rs index 3224cefee..0ab19ba18 100644 --- a/testing/integration/src/rpc_tests.rs +++ b/testing/integration/src/rpc_tests.rs @@ -79,21 +79,24 @@ async fn sanity_test() { .unwrap(); // Before submitting a first block, the sink is the genesis, - let response = rpc_client.get_sink_call(GetSinkRequest {}).await.unwrap(); + let response = rpc_client.get_sink_call(None, GetSinkRequest {}).await.unwrap(); assert_eq!(response.sink, SIMNET_GENESIS.hash); - let response = rpc_client.get_sink_blue_score_call(GetSinkBlueScoreRequest {}).await.unwrap(); + let response = rpc_client.get_sink_blue_score_call(None, GetSinkBlueScoreRequest {}).await.unwrap(); assert_eq!(response.blue_score, 0); // the block count is 0 - let response = rpc_client.get_block_count_call(GetBlockCountRequest {}).await.unwrap(); + let response = rpc_client.get_block_count_call(None, GetBlockCountRequest {}).await.unwrap(); assert_eq!(response.block_count, 0); // and the virtual chain is the genesis only let response = rpc_client - .get_virtual_chain_from_block_call(GetVirtualChainFromBlockRequest { - start_hash: SIMNET_GENESIS.hash, - include_accepted_transaction_ids: false, - }) + .get_virtual_chain_from_block_call( + None, + GetVirtualChainFromBlockRequest { + start_hash: SIMNET_GENESIS.hash, + include_accepted_transaction_ids: false, + }, + ) .await .unwrap(); assert!(response.added_chain_block_hashes.is_empty()); @@ -101,10 +104,13 @@ async fn sanity_test() { // Get a block template let GetBlockTemplateResponse { block, is_synced } = rpc_client - .get_block_template_call(GetBlockTemplateRequest { - pay_address: Address::new(Prefix::Simnet, Version::PubKey, &[0u8; 32]), - extra_data: Vec::new(), - }) + .get_block_template_call( + None, + GetBlockTemplateRequest { + pay_address: Address::new(Prefix::Simnet, Version::PubKey, &[0u8; 32]), + extra_data: Vec::new(), + }, + ) .await .unwrap(); assert!(!is_synced); @@ -131,19 +137,22 @@ async fn sanity_test() { } // After submitting a first block, the sink is the submitted block, - let response = rpc_client.get_sink_call(GetSinkRequest {}).await.unwrap(); + let response = rpc_client.get_sink_call(None, GetSinkRequest {}).await.unwrap(); assert_eq!(response.sink, block.header.hash); // the block count is 1 - let response = rpc_client.get_block_count_call(GetBlockCountRequest {}).await.unwrap(); + let response = rpc_client.get_block_count_call(None, GetBlockCountRequest {}).await.unwrap(); assert_eq!(response.block_count, 1); // and the virtual chain from genesis contains the added block let response = rpc_client - .get_virtual_chain_from_block_call(GetVirtualChainFromBlockRequest { - start_hash: SIMNET_GENESIS.hash, - include_accepted_transaction_ids: false, - }) + .get_virtual_chain_from_block_call( + None, + GetVirtualChainFromBlockRequest { + start_hash: SIMNET_GENESIS.hash, + include_accepted_transaction_ids: false, + }, + ) .await .unwrap(); assert!(response.added_chain_block_hashes.contains(&block.header.hash)); @@ -158,7 +167,7 @@ async fn sanity_test() { KaspadPayloadOps::GetCurrentNetwork => { let rpc_client = client.clone(); tst!(op, { - let response = rpc_client.get_current_network_call(GetCurrentNetworkRequest {}).await.unwrap(); + let response = rpc_client.get_current_network_call(None, GetCurrentNetworkRequest {}).await.unwrap(); assert_eq!(response.network, network_id.network_type); }) } @@ -166,11 +175,12 @@ async fn sanity_test() { KaspadPayloadOps::GetBlock => { let rpc_client = client.clone(); tst!(op, { - let result = rpc_client.get_block_call(GetBlockRequest { hash: 0.into(), include_transactions: false }).await; + let result = + rpc_client.get_block_call(None, GetBlockRequest { hash: 0.into(), include_transactions: false }).await; assert!(result.is_err()); let response = rpc_client - .get_block_call(GetBlockRequest { hash: SIMNET_GENESIS.hash, include_transactions: false }) + .get_block_call(None, GetBlockRequest { hash: SIMNET_GENESIS.hash, include_transactions: false }) .await .unwrap(); assert_eq!(response.block.header.hash, SIMNET_GENESIS.hash); @@ -181,7 +191,7 @@ async fn sanity_test() { let rpc_client = client.clone(); tst!(op, { let response = rpc_client - .get_blocks_call(GetBlocksRequest { include_blocks: true, include_transactions: false, low_hash: None }) + .get_blocks_call(None, GetBlocksRequest { include_blocks: true, include_transactions: false, low_hash: None }) .await .unwrap(); assert_eq!(response.blocks.len(), 1, "genesis block should be returned"); @@ -193,7 +203,7 @@ async fn sanity_test() { KaspadPayloadOps::GetInfo => { let rpc_client = client.clone(); tst!(op, { - let response = rpc_client.get_info_call(GetInfoRequest {}).await.unwrap(); + let response = rpc_client.get_info_call(None, GetInfoRequest {}).await.unwrap(); assert_eq!(response.server_version, kaspa_core::kaspad_env::version().to_string()); assert_eq!(response.mempool_size, 0); assert!(response.is_utxo_indexed); @@ -220,11 +230,14 @@ async fn sanity_test() { let rpc_client = client.clone(); tst!(op, { let response_result = rpc_client - .get_mempool_entry_call(GetMempoolEntryRequest { - transaction_id: 0.into(), - include_orphan_pool: true, - filter_transaction_pool: false, - }) + .get_mempool_entry_call( + None, + GetMempoolEntryRequest { + transaction_id: 0.into(), + include_orphan_pool: true, + filter_transaction_pool: false, + }, + ) .await; // Test Get Mempool Entry: // TODO: Fix by adding actual mempool entries this can get because otherwise it errors out @@ -236,10 +249,10 @@ async fn sanity_test() { let rpc_client = client.clone(); tst!(op, { let response = rpc_client - .get_mempool_entries_call(GetMempoolEntriesRequest { - include_orphan_pool: true, - filter_transaction_pool: false, - }) + .get_mempool_entries_call( + None, + GetMempoolEntriesRequest { include_orphan_pool: true, filter_transaction_pool: false }, + ) .await .unwrap(); assert!(response.mempool_entries.is_empty()); @@ -249,7 +262,7 @@ async fn sanity_test() { KaspadPayloadOps::GetConnectedPeerInfo => { let rpc_client = client.clone(); tst!(op, { - let response = rpc_client.get_connected_peer_info_call(GetConnectedPeerInfoRequest {}).await.unwrap(); + let response = rpc_client.get_connected_peer_info_call(None, GetConnectedPeerInfoRequest {}).await.unwrap(); assert!(response.peer_info.is_empty()); }) } @@ -258,12 +271,12 @@ async fn sanity_test() { let rpc_client = client.clone(); tst!(op, { let peer_address = ContextualNetAddress::from_str("1.2.3.4").unwrap(); - let _ = rpc_client.add_peer_call(AddPeerRequest { peer_address, is_permanent: true }).await.unwrap(); + let _ = rpc_client.add_peer_call(None, AddPeerRequest { peer_address, is_permanent: true }).await.unwrap(); // Add peer only adds the IP to a connection request. It will only be added to known_addresses if it // actually can be connected to. So in this test we can't expect it to be added unless we set up an // actual peer. - let response = rpc_client.get_peer_addresses_call(GetPeerAddressesRequest {}).await.unwrap(); + let response = rpc_client.get_peer_addresses_call(None, GetPeerAddressesRequest {}).await.unwrap(); assert!(response.known_addresses.is_empty()); }) } @@ -274,14 +287,14 @@ async fn sanity_test() { let peer_address = ContextualNetAddress::from_str("5.6.7.8").unwrap(); let ip = peer_address.normalize(1).ip; - let _ = rpc_client.add_peer_call(AddPeerRequest { peer_address, is_permanent: false }).await.unwrap(); - let _ = rpc_client.ban_call(BanRequest { ip }).await.unwrap(); + let _ = rpc_client.add_peer_call(None, AddPeerRequest { peer_address, is_permanent: false }).await.unwrap(); + let _ = rpc_client.ban_call(None, BanRequest { ip }).await.unwrap(); - let response = rpc_client.get_peer_addresses_call(GetPeerAddressesRequest {}).await.unwrap(); + let response = rpc_client.get_peer_addresses_call(None, GetPeerAddressesRequest {}).await.unwrap(); assert!(response.banned_addresses.contains(&ip)); - let _ = rpc_client.unban_call(UnbanRequest { ip }).await.unwrap(); - let response = rpc_client.get_peer_addresses_call(GetPeerAddressesRequest {}).await.unwrap(); + let _ = rpc_client.unban_call(None, UnbanRequest { ip }).await.unwrap(); + let response = rpc_client.get_peer_addresses_call(None, GetPeerAddressesRequest {}).await.unwrap(); assert!(!response.banned_addresses.contains(&ip)); }) } @@ -305,7 +318,7 @@ async fn sanity_test() { let rpc_client = client.clone(); tst!(op, { let result = - rpc_client.get_subnetwork_call(GetSubnetworkRequest { subnetwork_id: SubnetworkId::from_byte(0) }).await; + rpc_client.get_subnetwork_call(None, GetSubnetworkRequest { subnetwork_id: SubnetworkId::from_byte(0) }).await; // Err because it's currently unimplemented assert!(result.is_err()); @@ -323,7 +336,7 @@ async fn sanity_test() { KaspadPayloadOps::GetBlockDagInfo => { let rpc_client = client.clone(); tst!(op, { - let response = rpc_client.get_block_dag_info_call(GetBlockDagInfoRequest {}).await.unwrap(); + let response = rpc_client.get_block_dag_info_call(None, GetBlockDagInfoRequest {}).await.unwrap(); assert_eq!(response.network, network_id); }) } @@ -332,9 +345,10 @@ async fn sanity_test() { let rpc_client = client.clone(); tst!(op, { let response_result = rpc_client - .resolve_finality_conflict_call(ResolveFinalityConflictRequest { - finality_block_hash: Hash::from_bytes([0; 32]), - }) + .resolve_finality_conflict_call( + None, + ResolveFinalityConflictRequest { finality_block_hash: Hash::from_bytes([0; 32]) }, + ) .await; // Err because it's currently unimplemented @@ -346,7 +360,7 @@ async fn sanity_test() { let rpc_client = client.clone(); tst!(op, { let response_result = rpc_client - .get_headers_call(GetHeadersRequest { start_hash: SIMNET_GENESIS.hash, limit: 1, is_ascending: true }) + .get_headers_call(None, GetHeadersRequest { start_hash: SIMNET_GENESIS.hash, limit: 1, is_ascending: true }) .await; // Err because it's currently unimplemented @@ -358,7 +372,8 @@ async fn sanity_test() { let rpc_client = client.clone(); tst!(op, { let addresses = vec![Address::new(Prefix::Simnet, Version::PubKey, &[0u8; 32])]; - let response = rpc_client.get_utxos_by_addresses_call(GetUtxosByAddressesRequest { addresses }).await.unwrap(); + let response = + rpc_client.get_utxos_by_addresses_call(None, GetUtxosByAddressesRequest { addresses }).await.unwrap(); assert!(response.entries.is_empty()); }) } @@ -367,9 +382,10 @@ async fn sanity_test() { let rpc_client = client.clone(); tst!(op, { let response = rpc_client - .get_balance_by_address_call(GetBalanceByAddressRequest { - address: Address::new(Prefix::Simnet, Version::PubKey, &[0u8; 32]), - }) + .get_balance_by_address_call( + None, + GetBalanceByAddressRequest { address: Address::new(Prefix::Simnet, Version::PubKey, &[0u8; 32]) }, + ) .await .unwrap(); assert_eq!(response.balance, 0); @@ -381,7 +397,7 @@ async fn sanity_test() { tst!(op, { let addresses = vec![Address::new(Prefix::Simnet, Version::PubKey, &[1u8; 32])]; let response = rpc_client - .get_balances_by_addresses_call(GetBalancesByAddressesRequest::new(addresses.clone())) + .get_balances_by_addresses_call(None, GetBalancesByAddressesRequest::new(addresses.clone())) .await .unwrap(); assert_eq!(response.entries.len(), 1); @@ -389,7 +405,7 @@ async fn sanity_test() { assert_eq!(response.entries[0].balance, Some(0)); let response = - rpc_client.get_balances_by_addresses_call(GetBalancesByAddressesRequest::new(vec![])).await.unwrap(); + rpc_client.get_balances_by_addresses_call(None, GetBalancesByAddressesRequest::new(vec![])).await.unwrap(); assert!(response.entries.is_empty()); }) } @@ -397,7 +413,7 @@ async fn sanity_test() { KaspadPayloadOps::GetSinkBlueScore => { let rpc_client = client.clone(); tst!(op, { - let response = rpc_client.get_sink_blue_score_call(GetSinkBlueScoreRequest {}).await.unwrap(); + let response = rpc_client.get_sink_blue_score_call(None, GetSinkBlueScoreRequest {}).await.unwrap(); // A concurrent test may have added a single block so the blue score can be either 0 or 1 assert!(response.blue_score < 2); }) @@ -407,10 +423,10 @@ async fn sanity_test() { let rpc_client = client.clone(); tst!(op, { let response_result = rpc_client - .estimate_network_hashes_per_second_call(EstimateNetworkHashesPerSecondRequest { - window_size: 1000, - start_hash: None, - }) + .estimate_network_hashes_per_second_call( + None, + EstimateNetworkHashesPerSecondRequest { window_size: 1000, start_hash: None }, + ) .await; // The current DAA window is almost empty so an error is expected assert!(response_result.is_err()); @@ -422,11 +438,10 @@ async fn sanity_test() { tst!(op, { let addresses = vec![Address::new(Prefix::Simnet, Version::PubKey, &[0u8; 32])]; let response = rpc_client - .get_mempool_entries_by_addresses_call(GetMempoolEntriesByAddressesRequest::new( - addresses.clone(), - true, - false, - )) + .get_mempool_entries_by_addresses_call( + None, + GetMempoolEntriesByAddressesRequest::new(addresses.clone(), true, false), + ) .await .unwrap(); assert_eq!(response.entries.len(), 1); @@ -439,7 +454,7 @@ async fn sanity_test() { KaspadPayloadOps::GetCoinSupply => { let rpc_client = client.clone(); tst!(op, { - let response = rpc_client.get_coin_supply_call(GetCoinSupplyRequest {}).await.unwrap(); + let response = rpc_client.get_coin_supply_call(None, GetCoinSupplyRequest {}).await.unwrap(); assert_eq!(response.circulating_sompi, 0); assert_eq!(response.max_sompi, MAX_SOMPI); }) @@ -448,7 +463,7 @@ async fn sanity_test() { KaspadPayloadOps::Ping => { let rpc_client = client.clone(); tst!(op, { - let _ = rpc_client.ping_call(PingRequest {}).await.unwrap(); + let _ = rpc_client.ping_call(None, PingRequest {}).await.unwrap(); }) } @@ -456,48 +471,60 @@ async fn sanity_test() { let rpc_client = client.clone(); tst!(op, { let get_metrics_call_response = rpc_client - .get_metrics_call(GetMetricsRequest { - consensus_metrics: true, - connection_metrics: true, - bandwidth_metrics: true, - process_metrics: true, - }) + .get_metrics_call( + None, + GetMetricsRequest { + consensus_metrics: true, + connection_metrics: true, + bandwidth_metrics: true, + process_metrics: true, + }, + ) .await .unwrap(); assert!(get_metrics_call_response.process_metrics.is_some()); assert!(get_metrics_call_response.consensus_metrics.is_some()); let get_metrics_call_response = rpc_client - .get_metrics_call(GetMetricsRequest { - consensus_metrics: false, - connection_metrics: true, - bandwidth_metrics: true, - process_metrics: true, - }) + .get_metrics_call( + None, + GetMetricsRequest { + consensus_metrics: false, + connection_metrics: true, + bandwidth_metrics: true, + process_metrics: true, + }, + ) .await .unwrap(); assert!(get_metrics_call_response.process_metrics.is_some()); assert!(get_metrics_call_response.consensus_metrics.is_none()); let get_metrics_call_response = rpc_client - .get_metrics_call(GetMetricsRequest { - consensus_metrics: true, - connection_metrics: true, - bandwidth_metrics: false, - process_metrics: false, - }) + .get_metrics_call( + None, + GetMetricsRequest { + consensus_metrics: true, + connection_metrics: true, + bandwidth_metrics: false, + process_metrics: false, + }, + ) .await .unwrap(); assert!(get_metrics_call_response.process_metrics.is_none()); assert!(get_metrics_call_response.consensus_metrics.is_some()); let get_metrics_call_response = rpc_client - .get_metrics_call(GetMetricsRequest { - consensus_metrics: false, - connection_metrics: true, - bandwidth_metrics: false, - process_metrics: false, - }) + .get_metrics_call( + None, + GetMetricsRequest { + consensus_metrics: false, + connection_metrics: true, + bandwidth_metrics: false, + process_metrics: false, + }, + ) .await .unwrap(); assert!(get_metrics_call_response.process_metrics.is_none()); @@ -508,7 +535,7 @@ async fn sanity_test() { KaspadPayloadOps::GetServerInfo => { let rpc_client = client.clone(); tst!(op, { - let response = rpc_client.get_server_info_call(GetServerInfoRequest {}).await.unwrap(); + let response = rpc_client.get_server_info_call(None, GetServerInfoRequest {}).await.unwrap(); assert!(response.has_utxo_index); // we set utxoindex above assert_eq!(response.network_id, network_id); }) @@ -517,7 +544,7 @@ async fn sanity_test() { KaspadPayloadOps::GetSyncStatus => { let rpc_client = client.clone(); tst!(op, { - let _ = rpc_client.get_sync_status_call(GetSyncStatusRequest {}).await.unwrap(); + let _ = rpc_client.get_sync_status_call(None, GetSyncStatusRequest {}).await.unwrap(); }) } @@ -525,9 +552,10 @@ async fn sanity_test() { let rpc_client = client.clone(); tst!(op, { let results = rpc_client - .get_daa_score_timestamp_estimate_call(GetDaaScoreTimestampEstimateRequest { - daa_scores: vec![0, 500, 2000, u64::MAX], - }) + .get_daa_score_timestamp_estimate_call( + None, + GetDaaScoreTimestampEstimateRequest { daa_scores: vec![0, 500, 2000, u64::MAX] }, + ) .await .unwrap(); @@ -536,7 +564,7 @@ async fn sanity_test() { } let results = rpc_client - .get_daa_score_timestamp_estimate_call(GetDaaScoreTimestampEstimateRequest { daa_scores: vec![] }) + .get_daa_score_timestamp_estimate_call(None, GetDaaScoreTimestampEstimateRequest { daa_scores: vec![] }) .await .unwrap(); @@ -632,7 +660,7 @@ async fn sanity_test() { // Shutdown should only be tested after everything let rpc_client = client.clone(); - let _ = rpc_client.shutdown_call(ShutdownRequest {}).await.unwrap(); + let _ = rpc_client.shutdown_call(None, ShutdownRequest {}).await.unwrap(); // // Fold-up diff --git a/wallet/core/src/tests/rpc_core_mock.rs b/wallet/core/src/tests/rpc_core_mock.rs index 6c335d59a..3143fc546 100644 --- a/wallet/core/src/tests/rpc_core_mock.rs +++ b/wallet/core/src/tests/rpc_core_mock.rs @@ -9,7 +9,7 @@ use kaspa_notify::scope::Scope; use kaspa_notify::subscription::context::SubscriptionContext; use kaspa_notify::subscription::{MutationPolicies, UtxosChangedMutationPolicy}; use kaspa_rpc_core::api::ctl::RpcCtl; -use kaspa_rpc_core::{api::rpc::RpcApi, *}; +use kaspa_rpc_core::{api::connection::DynRpcConnection, api::rpc::RpcApi, *}; use kaspa_rpc_core::{notify::connection::ChannelConnection, RpcResult}; use std::sync::Arc; @@ -83,7 +83,7 @@ impl Default for RpcCoreMock { #[async_trait] impl RpcApi for RpcCoreMock { // This fn needs to succeed while the client connects - async fn get_info_call(&self, _request: GetInfoRequest) -> RpcResult { + async fn get_info_call(&self, _connection: Option<&DynRpcConnection>, _request: GetInfoRequest) -> RpcResult { Ok(GetInfoResponse { p2p_id: "wallet-mock".to_string(), mempool_size: 1234, @@ -95,133 +95,213 @@ impl RpcApi for RpcCoreMock { }) } - async fn ping_call(&self, _request: PingRequest) -> RpcResult { + async fn ping_call(&self, _connection: Option<&DynRpcConnection>, _request: PingRequest) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_metrics_call(&self, _request: GetMetricsRequest) -> RpcResult { + async fn get_metrics_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetMetricsRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_server_info_call(&self, _request: GetServerInfoRequest) -> RpcResult { + async fn get_server_info_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetServerInfoRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_sync_status_call(&self, _request: GetSyncStatusRequest) -> RpcResult { + async fn get_sync_status_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetSyncStatusRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_current_network_call(&self, _request: GetCurrentNetworkRequest) -> RpcResult { + async fn get_current_network_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetCurrentNetworkRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn submit_block_call(&self, _request: SubmitBlockRequest) -> RpcResult { + async fn submit_block_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: SubmitBlockRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_block_template_call(&self, _request: GetBlockTemplateRequest) -> RpcResult { + async fn get_block_template_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetBlockTemplateRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_peer_addresses_call(&self, _request: GetPeerAddressesRequest) -> RpcResult { + async fn get_peer_addresses_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetPeerAddressesRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_sink_call(&self, _request: GetSinkRequest) -> RpcResult { + async fn get_sink_call(&self, _connection: Option<&DynRpcConnection>, _request: GetSinkRequest) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_mempool_entry_call(&self, _request: GetMempoolEntryRequest) -> RpcResult { + async fn get_mempool_entry_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetMempoolEntryRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_mempool_entries_call(&self, _request: GetMempoolEntriesRequest) -> RpcResult { + async fn get_mempool_entries_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetMempoolEntriesRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_connected_peer_info_call(&self, _request: GetConnectedPeerInfoRequest) -> RpcResult { + async fn get_connected_peer_info_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetConnectedPeerInfoRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn add_peer_call(&self, _request: AddPeerRequest) -> RpcResult { + async fn add_peer_call(&self, _connection: Option<&DynRpcConnection>, _request: AddPeerRequest) -> RpcResult { Err(RpcError::NotImplemented) } - async fn submit_transaction_call(&self, _request: SubmitTransactionRequest) -> RpcResult { + async fn submit_transaction_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: SubmitTransactionRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_block_call(&self, _request: GetBlockRequest) -> RpcResult { + async fn get_block_call(&self, _connection: Option<&DynRpcConnection>, _request: GetBlockRequest) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_subnetwork_call(&self, _request: GetSubnetworkRequest) -> RpcResult { + async fn get_subnetwork_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetSubnetworkRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } async fn get_virtual_chain_from_block_call( &self, + _connection: Option<&DynRpcConnection>, _request: GetVirtualChainFromBlockRequest, ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_blocks_call(&self, _request: GetBlocksRequest) -> RpcResult { + async fn get_blocks_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetBlocksRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_block_count_call(&self, _request: GetBlockCountRequest) -> RpcResult { + async fn get_block_count_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetBlockCountRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_block_dag_info_call(&self, _request: GetBlockDagInfoRequest) -> RpcResult { + async fn get_block_dag_info_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetBlockDagInfoRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } async fn resolve_finality_conflict_call( &self, + _connection: Option<&DynRpcConnection>, _request: ResolveFinalityConflictRequest, ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn shutdown_call(&self, _request: ShutdownRequest) -> RpcResult { + async fn shutdown_call(&self, _connection: Option<&DynRpcConnection>, _request: ShutdownRequest) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_headers_call(&self, _request: GetHeadersRequest) -> RpcResult { + async fn get_headers_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetHeadersRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_balance_by_address_call(&self, _request: GetBalanceByAddressRequest) -> RpcResult { + async fn get_balance_by_address_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetBalanceByAddressRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } async fn get_balances_by_addresses_call( &self, + _connection: Option<&DynRpcConnection>, _request: GetBalancesByAddressesRequest, ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_utxos_by_addresses_call(&self, _request: GetUtxosByAddressesRequest) -> RpcResult { + async fn get_utxos_by_addresses_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetUtxosByAddressesRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_sink_blue_score_call(&self, _request: GetSinkBlueScoreRequest) -> RpcResult { + async fn get_sink_blue_score_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetSinkBlueScoreRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn ban_call(&self, _request: BanRequest) -> RpcResult { + async fn ban_call(&self, _connection: Option<&DynRpcConnection>, _request: BanRequest) -> RpcResult { Err(RpcError::NotImplemented) } - async fn unban_call(&self, _request: UnbanRequest) -> RpcResult { + async fn unban_call(&self, _connection: Option<&DynRpcConnection>, _request: UnbanRequest) -> RpcResult { Err(RpcError::NotImplemented) } async fn estimate_network_hashes_per_second_call( &self, + _connection: Option<&DynRpcConnection>, _request: EstimateNetworkHashesPerSecondRequest, ) -> RpcResult { Err(RpcError::NotImplemented) @@ -229,17 +309,23 @@ impl RpcApi for RpcCoreMock { async fn get_mempool_entries_by_addresses_call( &self, + _connection: Option<&DynRpcConnection>, _request: GetMempoolEntriesByAddressesRequest, ) -> RpcResult { Err(RpcError::NotImplemented) } - async fn get_coin_supply_call(&self, _request: GetCoinSupplyRequest) -> RpcResult { + async fn get_coin_supply_call( + &self, + _connection: Option<&DynRpcConnection>, + _request: GetCoinSupplyRequest, + ) -> RpcResult { Err(RpcError::NotImplemented) } async fn get_daa_score_timestamp_estimate_call( &self, + _connection: Option<&DynRpcConnection>, _request: GetDaaScoreTimestampEstimateRequest, ) -> RpcResult { Err(RpcError::NotImplemented)