diff --git a/lib/src/json_rpc.rs b/lib/src/json_rpc.rs index f4eeb51409..dddade84a1 100644 --- a/lib/src/json_rpc.rs +++ b/lib/src/json_rpc.rs @@ -61,7 +61,10 @@ // TODO: write docs about usage ^ +pub mod clients_multiplexer; pub mod methods; pub mod parse; pub mod payment_info; +// TODO: uncomment or remove module altogether pub mod reverse_proxy; +pub mod servers_multiplexer; pub mod service; diff --git a/lib/src/json_rpc/clients_multiplexer.rs b/lib/src/json_rpc/clients_multiplexer.rs new file mode 100644 index 0000000000..540bbcd553 --- /dev/null +++ b/lib/src/json_rpc/clients_multiplexer.rs @@ -0,0 +1,696 @@ +// Smoldot +// Copyright (C) 2024 Pierre Krieger +// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! Accepts multiple clients and merges their requests into a single stream of requests, then +//! distributes back the responses to the relevant clients. +//! +//! Because multiple different clients might accidentally use the same IDs for their requests, +//! the [`ClientsMultiplexer`] modifies the ID of every request. This is also necessary in order +//! to match a server's response with its corresponding client. +//! +//! When a client is removed, dummy JSON-RPC requests are added to the stream of requests that +//! unsubscribe from the subscriptions that this client was maintaining. +//! +//! The [`ClientsMultiplexer`] might silently discard or merge notifications inserted through +//! [`ClientsMultiplexer::push_server_to_client`] in order to guarantee a bound to the maximum +//! number to the memory usage in situations where a client doesn't pull responses quickly enough. +//! +//! When a client sends a `chain_subscribeAllHeads`, `chain_subscribeFinalizedHeads`, +//! `chain_subscribeNewHeads`, or `state_subscribeRuntimeVersion` request, the +//! [`ClientsMultiplexer`] tries to re-use an existing server subscription. Notifications +//! concerning these subscriptions are multiplexed. +//! +//! When a client sends a `chainHead_v1_follow` JSON-RPC request, the [`ClientsMultiplexer`] +//! will let it pass through, no matter how many other existing `chainHead_v1_follow` +//! subscriptions exist. However, because there exists a limit to the maximum number of +//! `chainHead_v1_follow` subscriptions, the server might return `null` to indicate that +//! this limit has been reached. When that happens, the [`ClientsMultiplexer`] will use the same +//! server-side `chainHead_v1_follow` subscription to feed to multiple clients. + +use core::cmp; + +use alloc::{ + borrow::{Cow, ToOwned as _}, + collections::{BTreeMap, BTreeSet, VecDeque}, + format, + string::String, + sync::Arc, +}; +use rand::seq::IteratorRandom as _; +use rand_chacha::{ + rand_core::{RngCore as _, SeedableRng as _}, + ChaCha20Rng, +}; + +use crate::{ + json_rpc::{methods, parse}, + util::SipHasherBuild, +}; + +mod responses_queue; + +/// Identifier of a client within the [`ClientsMultiplexer`]. +#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ClientId(usize); + +/// Limits enforced to clients. +#[derive(Debug, Copy, Clone, PartialEq, Eq)] +pub struct ClientLimits { + /// Maximum number of requests that haven't been answered yet that the client is allowed to + /// make. + pub max_unanswered_parallel_requests: usize, + + /// Maximum number of concurrent subscriptions for the legacy JSON-RPC API, except for + /// `author_submitAndWatchExtrinsic` which is handled by + /// [`ClientConfig::max_transactions_subscriptions`]. + pub max_legacy_api_subscriptions: usize, + + /// Maximum number of concurrent `chainHead_follow` subscriptions. + /// + /// The specification mentions that this value must be at least 2. If the value is inferior to + /// 2, it is raised. + pub max_chainhead_follow_subscriptions: usize, + + /// Maximum number of concurrent transactions-related subscriptions. + pub max_transactions_subscriptions: usize, +} + +/// See [the module-level documentation](..). +// TODO: Debug impl +pub struct ClientsMultiplexer { + /// List of all clients, including clients that have been removed by the API user. + clients: slab::Slab>, + + /// Subset of [`ClientsMultiplexer::clients`] that has at least one request in + /// [`Client::requests_queue`]. + clients_with_requests: hashbrown::HashSet, + + /// List of subscriptions indexed by id. + /// + /// The same subscription ID can be used for multiple clients at the same time. + subscriptions_by_server_id: BTreeMap<(Arc, ClientId), Subscription>, + + /// List of subscriptions indexed by client. + subscriptions_by_client: BTreeSet<(ClientId, Arc)>, + + /// Source of randomness used for various purposes. + randomness: ChaCha20Rng, +} + +struct Client { + /// Queue of client-to-server JSON-RPC requests. + requests_queue: VecDeque, + + /// For each request ID (encoded as JSON), the in-progress requests with that ID. + /// + /// The vast majority of the time there will only be one entry, but there's in principle + /// nothing illegal in having multiple requests with the same ID. + requests_in_progress: + hashbrown::HashMap, SipHasherBuild>, + + /// Queue of server-to-client responses and notifications. + // TODO: occasionally shrink to fit? + responses_queue: responses_queue::ResponsesQueue, + + /// Limits enforced on the client. + limits: ClientLimits, + + /// User data decided by the API user. If `None`, the client has been removed by the API user. + user_data: Option, +} + +struct ResponseEntry { + /// Stringified version of the response or notification. + as_string: String, + + /// If this entry is a notification to a subscription, contains the identifier of this + /// subscription. + // TODO: is String maybe too expensive to clone? + subscription_notification: Option>, +} + +struct InProgressRequest { + /// Subscription ID this request wants to unsubscribe from. + unsubscribe: Option, +} + +// TODO: remove? +struct Subscription { + ty: SubscriptionTy, +} + +enum SubscriptionTy { + AllHeads { + latest_update_queue_index: Option, + }, + FinalizedHeads { + latest_update_queue_index: Option, + }, + NewHeads { + latest_update_queue_index: Option, + }, + RuntimeVersion { + latest_update_queue_index: Option, + }, + Storage { + latest_update_queue_index: Option, + }, +} + +impl ClientsMultiplexer { + /// Initializes a new [`ClientsMultiplexer`]. + pub fn new() -> Self { + let mut randomness = ChaCha20Rng::from_seed(todo!()); + ClientsMultiplexer { + clients: slab::Slab::with_capacity(todo!()), + clients_with_requests: hashbrown::HashSet::with_capacity_and_hasher( + todo!(), + Default::default(), + ), + subscriptions_by_server_id: BTreeMap::new(), + subscriptions_by_client: BTreeSet::new(), + randomness, + } + } + + /// Adds a new client to the collection. + pub fn add_client(&mut self, mut limits: ClientLimits, user_data: T) -> ClientId { + Self::sanitize_limits(&mut limits); + + ClientId(self.clients.insert(Client { + requests_queue: VecDeque::with_capacity(cmp::min( + 32, + limits.max_unanswered_parallel_requests, + )), + responses_queue: responses_queue::ResponsesQueue::with_capacity(cmp::min( + 32, + limits.max_unanswered_parallel_requests, + )), + limits, + user_data: Some(user_data), + ..todo!() + })) + } + + /// Removes a client from the collection. + /// + /// Since the client might have requests in progress and active subscriptions, the removed + /// client might still occupy a non-zero amount of memory after being destroyed, until + /// enough requests and responses have been exchanged with the server. + /// + /// # Panics + /// + /// Panics if the [`ClientId`] is invalid. + /// + pub fn remove_client(&mut self, client_id: ClientId) -> T { + let Some(client) = self.clients.get_mut(client_id.0) else { + // Client was already destroyed. + panic!() + }; + + // Extract user data ahead of time, in order to avoid going further if the client was + // already destroyed. + let Some(user_data) = client.user_data.take() else { + // Client was already destroyed. + panic!() + }; + + // Clear the queue of unprocessed requests, as we don't need them anymore. + client.requests_queue.clear(); + + // Extract the list of subscriptions from `subscriptions_by_client`. + let subscriptions = { + let mut client_and_after = self + .subscriptions_by_client + .split_off(&(client_id, Arc::from(""))); + self.subscriptions_by_client.append( + &mut client_and_after.split_off(&(ClientId(client_id.0 + 1), Arc::from(""))), + ); + client_and_after + }; + + // For each active subscription this client has, add to the queue of unprocessed + // requests one dummy unsubscribe request. + // TODO: do this ^ + for (_, subscription_id) in subscriptions { + let Some(subscription) = self + .subscriptions_by_server_id + .remove(&(subscription_id.clone(), client_id)) + else { + unreachable!() + }; + + // TODO: only unsubscribe if sub isn't used by anything else + client.requests_queue.push_back(match () { + _ => unreachable!(), + }); + } + + user_data + } + + /// Modifies the limits set for the given client. + /// + /// # Panic + /// + /// Panics if the [`ClientId`] is invalid. + /// + pub fn set_client_limits(&mut self, client: ClientId, mut limits: ClientLimits) { + let Some(client) = self.clients.get_mut(client.0) else { + panic!() + }; + + Self::sanitize_limits(&mut limits); + client.limits = limits; + } + + /// Equivalent to calling [`ClientsMultiplexer::set_client_limits`] for every client that + /// exists. + pub fn set_clients_limits(&mut self, mut limits: ClientLimits) { + Self::sanitize_limits(&mut limits); + for (_, client) in &mut self.clients { + client.limits = limits; + } + } + + fn sanitize_limits(limits: &mut ClientLimits) { + limits.max_chainhead_follow_subscriptions = + cmp::max(limits.max_chainhead_follow_subscriptions, 2); + } + + /// Returns the number of alive clients plus the number of clients that have been removed + /// but that are still occuping memory. + /// + /// When enforcing a limit to the number of clients in the multiplexer, this is the number + /// that should be compared against a certain limit. + pub fn num_clients_footprints(&self) -> usize { + self.clients.len() + } + + /// Adds a new request to the back of the client-to-server queue. + /// + /// # Panic + /// + /// Panics if the [`ClientId`] is invalid. + /// + pub fn push_client_to_server( + &mut self, + client_id: ClientId, + request: &str, + ) -> Result { + // Check whether the client exists. + let Some(client) = self.clients.get_mut(client_id.0) else { + // Client doesn't exist. + panic!() + }; + + // Return an error if the client has already queued too many requests. + // Note the usage of `>=`, as it can happen that the limit has been modified to be + // lower than it was. + if client + .requests_queue + .len() + .checked_add(client.requests_in_progress.len()) + .map_or(true, |total| { + total >= client.limits.max_unanswered_parallel_requests + }) + { + return Err(PushClientToServerError::TooManySimultaneousRequests); + } + + // Try to parse the request. + let Ok(request) = parse::parse_request(&request) else { + client.responses_queue.push_back(ResponseEntry { + as_string: parse::build_parse_error_response(), + subscription_notification: None, + }); + return Ok(PushClientToServer::ImmediateAnswer); + }; + + // Get the request ID. + // The request might not have an ID if it is a notification. No notification is supported + // in the JSON-RPC API. As mentioned in the JSON-RPC spec, nothing should ever be sent in + // reply to notifications, not even errors. For this reason, we silently discard + // notifications. + let Some(request_id_json) = request.id_json else { + return Ok(PushClientToServer::Discarded); + }; + + // Try parse the request name and parameters. + let Ok(method) = methods::parse_jsonrpc_client_to_server_method_name_and_parameters( + request.method, + request.params_json, + ) else { + client.responses_queue.push_back(ResponseEntry { + as_string: parse::build_error_response( + request_id_json, + parse::ErrorResponse::MethodNotFound, + None, + ), + subscription_notification: None, + }); + return Ok(PushClientToServer::ImmediateAnswer); + }; + + // TODO: for subscriptions, re-use existing subscription if any + // TODO: check limit to number of subscriptions + + // Adjust the request ID to put the numerical value of the `ClientId` in front. + // This not only ensures that clients can't have overlapping request IDs, but will also + // be used in order to determine to which client a response to a request belongs to. + let new_request_id = serde_json::to_string(&format!("{}-{request_id_json}", client_id.0)) + .unwrap_or_else(|_| unreachable!()); + + // Now insert the request in queue. + debug_assert_ne!( + self.clients_with_requests.contains(&client_id), + client.requests_queue.is_empty() + ); + if client.requests_queue.is_empty() { + let _was_inserted = self.clients_with_requests.insert(client_id); + } + client + .requests_queue + .push_back(parse::build_request(&parse::Request { + id_json: Some(&new_request_id), + method: request.method, + params_json: request.params_json, + })); + + // Success. + Ok(PushClientToServer::Queued) + } + + /// Pops a request from the client-to-server queue of one of the clients. + pub fn pop_client_to_server(&mut self) -> Option { + // In order to guarantee fairness, we first choose which client to grab the request from, + // then only grab a request from its queue. + let Some(&client_id) = self + .clients_with_requests + .iter() + .choose(&mut self.randomness) + else { + debug_assert!(self.clients_with_requests.is_empty()); + return None; + }; + + let Some(client) = self.clients.get_mut(client_id.0) else { + unreachable!() + }; + + let Some(request) = client.requests_queue.pop_front() else { + unreachable!() + }; + + if client.requests_queue.is_empty() { + let _was_removed = self.clients_with_requests.remove(&client_id); + debug_assert!(_was_removed); + } + + // TODO: client.requests_in_progress.insert(); + + Some(request) + } + + /// Adds a JSON-RPC response or notification coming from the server. + /// + /// Responses that fail to parse or do not match any known request ID, and notifications that + /// do not match any known active subscription are ignored. + pub fn push_server_to_client(&mut self, response: &str) { + // TODO: take by String? ^ + match parse::parse_response(response) { + Ok(response) => { + // The response is a response to a request. + + // Request ID as sent by the server. + let server_request_id = match response { + parse::Response::Success { id_json, .. } => id_json, + parse::Response::Error { id_json, .. } => id_json, + parse::Response::ParseError { .. } => return, + }; + + // Extract the client ID and request ID from the client's perspective. + // Return and silently discard the response if there is a parsing error. + let (client_id, client_request_id) = { + let Ok(as_string) = serde_json::from_str::>(server_request_id) else { + return; + }; + let Some((id, rq_id)) = as_string.split_once('-') else { + return; + }; + let Ok(id) = id.parse::() else { return }; + if !self.clients.contains(id) { + return; + } + if serde_json::from_str::(&rq_id).is_err() { + return; + } + (ClientId(id), rq_id.to_owned()) + }; + + // TODO: must do things in case of subscription or unsubscription + + // Insert the response in queue, adjusting the request ID. + self.clients[client_id.0] + .responses_queue + .push_back(ResponseEntry { + subscription_notification: None, + as_string: match response { + parse::Response::Success { result_json, .. } => { + parse::build_success_response(&client_request_id, result_json) + } + parse::Response::Error { + error_code, + error_message, + error_data_json, + .. + } => parse::build_error_response( + &client_request_id, + todo!(), + error_data_json, + ), + parse::Response::ParseError { .. } => unreachable!(), + }, + }); + } + + Err(_) => { + // Try to parse the response as a subscription notification. + let Ok(notification) = methods::parse_notification(response) else { + return; + }; + + let subscription_id: Arc = Arc::from(&**notification.subscription()); + + for (&(_, client_id), subscription) in self.subscriptions_by_server_id.range_mut( + (subscription_id.clone(), ClientId(usize::MIN)) + ..=(subscription_id.clone(), ClientId(usize::MAX)), + ) { + let mut client = &mut self.clients[client_id.0]; + + match (&mut subscription.ty, ¬ification) { + // For some subscriptions, we overwrite any existing notification in the + // queue. + ( + SubscriptionTy::FinalizedHeads { + latest_update_queue_index: Some(latest_update_queue_index), + }, + methods::ServerToClient::chain_finalizedHead { .. }, + ) + | ( + SubscriptionTy::NewHeads { + latest_update_queue_index: Some(latest_update_queue_index), + }, + methods::ServerToClient::chain_newHead { .. }, + ) + | ( + SubscriptionTy::RuntimeVersion { + latest_update_queue_index: Some(latest_update_queue_index), + }, + methods::ServerToClient::state_runtimeVersion { .. }, + ) => { + // Update the existing entry in queue. + let entry = &mut client.responses_queue[*latest_update_queue_index]; + debug_assert_eq!( + entry.subscription_notification, + Some(subscription_id.clone()) + ); + entry.as_string = response.to_owned(); + return; + } + + // For some subscriptions, we push a notification if the queue is small + // enough. If the queue reaches a certain size, we overwrite the + // latest entry. + // The logic of the notifications makes it fundamentally wrong to + // overwrite notifications, however this is considered a defect in the + // logic of legacy JSON-RPC functions. + ( + SubscriptionTy::AllHeads { + latest_update_queue_index: Some(latest_update_queue_index), + }, + methods::ServerToClient::chain_allHead { .. }, + ) + | ( + SubscriptionTy::Storage { + latest_update_queue_index: Some(latest_update_queue_index), + }, + methods::ServerToClient::state_storage { .. }, + ) => { + // TODO: do properly; what's the criteria for adding and for overwriting? + // Update the existing entry in queue. + let entry = &mut client.responses_queue[*latest_update_queue_index]; + debug_assert_eq!( + entry.subscription_notification, + Some(subscription_id.clone()) + ); + entry.as_string = response.to_owned(); + return; + } + + // If there isn't any notification in the queue, insert one. + ( + SubscriptionTy::AllHeads { + latest_update_queue_index: ref mut latest_update_queue_index @ None, + }, + methods::ServerToClient::chain_allHead { .. }, + ) + | ( + SubscriptionTy::FinalizedHeads { + latest_update_queue_index: ref mut latest_update_queue_index @ None, + }, + methods::ServerToClient::chain_finalizedHead { .. }, + ) + | ( + SubscriptionTy::NewHeads { + latest_update_queue_index: ref mut latest_update_queue_index @ None, + }, + methods::ServerToClient::chain_newHead { .. }, + ) + | ( + SubscriptionTy::RuntimeVersion { + latest_update_queue_index: ref mut latest_update_queue_index @ None, + }, + methods::ServerToClient::state_runtimeVersion { .. }, + ) + | ( + SubscriptionTy::Storage { + latest_update_queue_index: ref mut latest_update_queue_index @ None, + }, + methods::ServerToClient::state_storage { .. }, + ) => { + *latest_update_queue_index = + Some(client.responses_queue.push_back(ResponseEntry { + as_string: response.to_owned(), + subscription_notification: Some(subscription_id.clone()), + })); + } + + _ => {} // TODO: no + } + } + } + } + } + + /// Pops an entry in the queue of server-to-client responses or notifications of the given + /// client. + /// + /// Returns `None` if the queue is empty. + /// + /// # Panic + /// + /// Panics if the [`ClientId`] is invalid. + /// + pub fn pop_server_to_client(&mut self, client_id: ClientId) -> Option { + let Some(client) = self + .clients + .get_mut(client_id.0) + .filter(|c| c.user_data.is_some()) + else { + // Invalid `ClientId`. + panic!() + }; + + let Some((entry_index, entry)) = client.responses_queue.pop_front() else { + return None; + }; + + // Update the local state regarding the position of subscriptions notifications within the + // queue. + if let Some(subscription_id) = entry.subscription_notification { + let Some(subscription_info) = self + .subscriptions_by_server_id + .get_mut(&(subscription_id, client_id)) + else { + unreachable!() + }; + + match &mut subscription_info.ty { + SubscriptionTy::AllHeads { + latest_update_queue_index, + } + | SubscriptionTy::NewHeads { + latest_update_queue_index, + } + | SubscriptionTy::FinalizedHeads { + latest_update_queue_index, + } + | SubscriptionTy::RuntimeVersion { + latest_update_queue_index, + } + | SubscriptionTy::Storage { + latest_update_queue_index, + } => { + if *latest_update_queue_index == Some(entry_index) { + *latest_update_queue_index = None; + } + } + } + } + + // Success. + Some(entry.as_string) + } +} + +/// Outcome of a call to [`ClientsMultiplexer::push_client_to_server`]. +#[derive(Debug)] +pub enum PushClientToServer { + /// The request has been silently discarded. + /// + /// This happens for example if the JSON-RPC client sends a notification. + Discarded, + + /// The request has been immediately answered and doesn't need any further processing. + /// [`ClientsMultiplexer::pop_server_to_client`] should be called in order to pull the + /// response and send it to the client. + ImmediateAnswer, + + /// The request can be processed by the server. + /// [`ClientsMultiplexer::pop_client_to_server`] will return `Some`. + Queued, +} + +/// Error potentially returned by a call to [`ClientsMultiplexer::push_client_to_server`]. +#[derive(Debug, derive_more::Display, Clone)] +pub enum PushClientToServerError { + /// The number of requests that this client has queued is already equal to + /// [`ClientLimits::max_unanswered_parallel_requests`]. + #[display(fmt = "Too many simultaneous requests")] + TooManySimultaneousRequests, +} diff --git a/lib/src/json_rpc/clients_multiplexer/responses_queue.rs b/lib/src/json_rpc/clients_multiplexer/responses_queue.rs new file mode 100644 index 0000000000..9319432279 --- /dev/null +++ b/lib/src/json_rpc/clients_multiplexer/responses_queue.rs @@ -0,0 +1,118 @@ +// Smoldot +// Copyright (C) 2024 Pierre Krieger +// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +use core::{fmt, iter, ops}; + +/// Index of an entry within the [`ResponsesQueue`]. +#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct EntryIndex(usize); + +/// One can think of this data structure as a linked list, except that indices aren't linear +/// (i.e. item 4 doesn't necessarily follow item 3) and do not change when entries are added or +/// removed. +pub struct ResponsesQueue { + container: slab::Slab>, + + first_entry: usize, + last_entry: usize, +} + +struct Entry { + value: T, + next_entry: usize, +} + +impl ResponsesQueue { + pub fn with_capacity(capacity: usize) -> Self { + ResponsesQueue { + container: slab::Slab::with_capacity(capacity), + first_entry: 0, + last_entry: 0, + } + } + + pub fn push_back(&mut self, value: T) -> EntryIndex { + let container_was_empty = self.container.is_empty(); + + let new_entry_index = self.container.insert(Entry { + value, + next_entry: 0, + }); + + if container_was_empty { + self.first_entry = new_entry_index; + } else { + self.container[self.last_entry].next_entry = new_entry_index; + } + + self.last_entry = new_entry_index; + + EntryIndex(new_entry_index) + } + + pub fn pop_front(&mut self) -> Option<(EntryIndex, T)> { + if self.container.is_empty() { + return None; + } + + let entry_index = self.first_entry; + let Some(entry) = self.container.try_remove(entry_index) else { + unreachable!() + }; + + self.first_entry = entry.next_entry; + Some((EntryIndex(entry_index), entry.value)) + } +} + +impl ops::Index for ResponsesQueue { + type Output = T; + + fn index(&self, index: EntryIndex) -> &Self::Output { + &self.container[index.0].value + } +} + +impl ops::IndexMut for ResponsesQueue { + fn index_mut(&mut self, index: EntryIndex) -> &mut Self::Output { + &mut self.container[index.0].value + } +} + +impl fmt::Debug for ResponsesQueue { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_list() + .entries( + iter::successors( + if self.container.is_empty() { + None + } else { + Some(self.first_entry) + }, + |&index| { + if index == self.last_entry { + None + } else { + Some(self.container[index].next_entry) + } + }, + ) + .map(|index| &self.container[index].value), + ) + .finish() + } +} diff --git a/lib/src/json_rpc/methods.rs b/lib/src/json_rpc/methods.rs index ffe327db90..7ac9e4a848 100644 --- a/lib/src/json_rpc/methods.rs +++ b/lib/src/json_rpc/methods.rs @@ -59,6 +59,24 @@ pub fn parse_jsonrpc_client_to_server( Ok((request_id, call)) } +/// Parses a JSON call (usually sent from a JSON-RPC client and received by a JSON-RPC server) +/// from the method name and parameters. +/// +/// The parameters are encoded in JSON. For example, `parameters_json` might be `["hello"]`. +pub fn parse_jsonrpc_client_to_server_method_name_and_parameters<'a>( + name: &'a str, + parameters_json: Option<&'a str>, +) -> Result, MethodError<'a>> { + MethodCall::from_defs(name, parameters_json) +} + +pub fn parse_jsonrpc_response<'a>( + request_method_name: &'_ str, + response_json: &'a str, +) -> Result, ()> { + Response::parse_from_request_method(request_method_name, response_json) +} + /// Error produced by [`parse_jsonrpc_client_to_server`]. #[derive(Debug, derive_more::Display)] pub enum ParseClientToServerError<'a> { @@ -366,6 +384,17 @@ macro_rules! define_methods { )* } } + + fn parse_from_request_method(request_method_name: &str, response_json: &str) -> Result { + match request_method_name { + $( + stringify!($name) => { + Ok($rp_name::$name(serde_json::from_str(response_json).unwrap())) + } + )* + _ => Err(()) + } + } } }; } @@ -403,9 +432,9 @@ define_methods! { chain_subscribeAllHeads() -> Cow<'a, str>, chain_subscribeFinalizedHeads() -> Cow<'a, str> [chain_subscribeFinalisedHeads], chain_subscribeNewHeads() -> Cow<'a, str> [subscribe_newHead, chain_subscribeNewHead], - chain_unsubscribeAllHeads(subscription: String) -> bool, - chain_unsubscribeFinalizedHeads(subscription: String) -> bool [chain_unsubscribeFinalisedHeads], - chain_unsubscribeNewHeads(subscription: String) -> bool [unsubscribe_newHead, chain_unsubscribeNewHead], + chain_unsubscribeAllHeads(subscription: Cow<'a, str>) -> bool, + chain_unsubscribeFinalizedHeads(subscription: Cow<'a, str>) -> bool [chain_unsubscribeFinalisedHeads], + chain_unsubscribeNewHeads(subscription: Cow<'a, str>) -> bool [unsubscribe_newHead, chain_unsubscribeNewHead], childstate_getKeys() -> (), // TODO: childstate_getStorage() -> (), // TODO: childstate_getStorageHash() -> (), // TODO: @@ -531,6 +560,40 @@ define_methods! { sudo_networkState_event(subscription: Cow<'a, str>, result: NetworkEvent) -> (), } +impl<'a> ServerToClient<'a> { + /// Returns the subscription identifier stored in the request. + pub fn subscription(&self) -> &Cow<'a, str> { + match self { + ServerToClient::author_extrinsicUpdate { subscription, .. } => subscription, + ServerToClient::chain_finalizedHead { subscription, .. } => subscription, + ServerToClient::chain_newHead { subscription, .. } => subscription, + ServerToClient::chain_allHead { subscription, .. } => subscription, + ServerToClient::state_runtimeVersion { subscription, .. } => subscription, + ServerToClient::state_storage { subscription, .. } => subscription, + ServerToClient::chainHead_v1_followEvent { subscription, .. } => subscription, + ServerToClient::transactionWatch_v1_watchEvent { subscription, .. } => subscription, + ServerToClient::sudo_networkState_event { subscription, .. } => subscription, + } + } + + /// Modifies the subscription identifier stored in the request. + pub fn set_subscription(&mut self, new_value: Cow<'a, str>) { + let sub = match self { + ServerToClient::author_extrinsicUpdate { subscription, .. } => subscription, + ServerToClient::chain_finalizedHead { subscription, .. } => subscription, + ServerToClient::chain_newHead { subscription, .. } => subscription, + ServerToClient::chain_allHead { subscription, .. } => subscription, + ServerToClient::state_runtimeVersion { subscription, .. } => subscription, + ServerToClient::state_storage { subscription, .. } => subscription, + ServerToClient::chainHead_v1_followEvent { subscription, .. } => subscription, + ServerToClient::transactionWatch_v1_watchEvent { subscription, .. } => subscription, + ServerToClient::sudo_networkState_event { subscription, .. } => subscription, + }; + + *sub = new_value; + } +} + #[derive(Clone, PartialEq, Eq, Hash)] pub struct HexString(pub Vec); @@ -1038,7 +1101,7 @@ pub struct SystemHealth { pub should_have_peers: bool, } -#[derive(Debug, Clone, serde::Serialize)] +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct SystemPeer { #[serde(rename = "peerId")] pub peer_id: String, // Example: "12D3KooWHEQXbvCzLYvc87obHV6HY4rruHz8BJ9Lw1Gg2csVfR6Z" @@ -1049,7 +1112,7 @@ pub struct SystemPeer { pub best_number: u64, } -#[derive(Debug, Clone, serde::Serialize)] +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub enum SystemPeerRole { #[serde(rename = "AUTHORITY")] Authority, @@ -1124,6 +1187,23 @@ impl serde::Serialize for RpcMethods { } } +impl<'de> serde::Deserialize<'de> for RpcMethods { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + #[derive(serde::Deserialize)] + struct SerdeRpcMethods { + methods: Vec, + } + + let serde_methods = ::deserialize(deserializer)?; + Ok(RpcMethods { + methods: serde_methods.methods, + }) + } +} + impl serde::Serialize for Block { fn serialize(&self, serializer: S) -> Result where @@ -1156,6 +1236,32 @@ impl serde::Serialize for Block { } } +impl<'de> serde::Deserialize<'de> for Block { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + #[derive(serde::Deserialize)] + struct SerdeBlock { + block: SerdeBlockInner, + } + + #[derive(serde::Deserialize)] + struct SerdeBlockInner { + extrinsics: Vec, + header: Header, + justifications: Option>>>, + } + + let serde_block = ::deserialize(deserializer)?; + Ok(Block { + extrinsics: serde_block.block.extrinsics, + header: serde_block.block.header, + justifications: todo!(), // TODO: + }) + } +} + impl serde::Serialize for RuntimeDispatchInfo { fn serialize(&self, serializer: S) -> Result where @@ -1183,6 +1289,16 @@ impl serde::Serialize for RuntimeDispatchInfo { } } +impl<'de> serde::Deserialize<'de> for RuntimeDispatchInfo { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + // TODO: + todo!() + } +} + #[derive(serde::Serialize, serde::Deserialize)] struct SerdeSystemHealth { #[serde(rename = "isSyncing")] diff --git a/lib/src/json_rpc/reverse_proxy.rs b/lib/src/json_rpc/reverse_proxy.rs new file mode 100644 index 0000000000..07c8c20095 --- /dev/null +++ b/lib/src/json_rpc/reverse_proxy.rs @@ -0,0 +1,1526 @@ +// Smoldot +// Copyright (C) 2023 Pierre Krieger +// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! +//! # Usage +//! +//! TODO +//! +//! Call [`ReverseProxy::add_client`] whenever a client connects, and +//! [`ReverseProxy::remove_client`] whenever a client disconnects. +//! +//! # Behavior +//! +//! The behavior is as follows: +//! +//! For the legacy JSON-RPC API: +//! +//! - `system_name` and `system_version` are answered directly by the proxy, and not directed +//! towards any of the servers. +//! - TODO: functions that always error +//! - For all other legacy JSON-RPC API functions: the first time one of these functions is called, +//! it is directed towards a randomly-chosen server. Then, further calls to one of the functions +//! are always redirect to the same server. For example, if the JSON-RPC client calls +//! `chain_subscribeAllHeads` then `chain_getBlock`, these two function calls are guaranteed to be +//! redirected to the same server. If this server is removed, a different server is randomly +//! chosen. Note that there's no mechanism for a server to "steal" clients from another server, as +//! doing so would create inconsistencies from the point of view of the JSON-RPC client. This is +//! considered as a defect in the legacy JSON-RPC API. +//! - If a server is dropped, a `Dropped` update is generated for any active +//! `author_submitAndWatchExtrinsic` subscription. For all other subscriptions, the subscription +//! is silently re-opened on a different server. +//! +//! For the new JSON-RPC API: +//! +//! - `sudo_unstable_version` is answered directly by the proxy, and not directed towards any of +//! the servers. +//! - `sudo_unstable_p2pDiscover` is answered directly by the proxy and is a no-op. TODO: is that correct? +//! - `chainSpec_v1_chainName`, `chainSpec_v1_genesisHash`, and `chainSpec_v1_properties` are +//! redirected towards a randomly-chosen server. +//! - Each call to `chainHead_unstable_follow` is redirected to a randomly-chosen server (possibly +//! multiple different servers for each call). All the other `chainHead` JSON-RPC functions are +//! then redirected to the server corresponding to the provided `followSubscriptionId`. If the +//! server is removed, a `stop` event is generated. +//! - Each call to `transaction_unstable_submitAndWatch` is redirected to a randomly-chosen +//! server. If the server is removed, a `dropped` event is generated. +//! +//! If no server is available, the reverse proxy will delay answering JSON-RPC requests until one +//! server is added (except for the JSON-RPC functions that are answered immediately, as they are +//! always answered immediately). +//! +//! JSON-RPC requests that can't be parsed, use an unknown function, have invalid parameters, etc. +//! are answered immediately by the reverse proxy. +//! +//! If a server misbehaves or returns an internal error, it gets blacklisted and no further +//! requests are sent to it. +//! +//! When a request sent by a client is transferred to a server, the identifier for that request +//! is modified to a randomly-generated identifier. This is necessary because multiple different +//! clients might use the same identifiers for their requests. +//! Similarly, when a server sends a JSON-RPC response containing a subscription ID, the +//! identifier of that subscription is modified to become a randomly-generated value. + +// TODO: what about rpc_methods + +use alloc::{ + borrow::Cow, + collections::{btree_map, BTreeMap, BTreeSet, VecDeque}, + format, + string::{String, ToString as _}, + vec::Vec, +}; +use core::{cmp, mem, ops}; +use rand::{seq::IteratorRandom as _, Rng as _}; +use rand_chacha::{ + rand_core::{RngCore as _, SeedableRng as _}, + ChaCha20Rng, +}; + +use crate::json_rpc::methods; + +use super::parse; + +/// Configuration for a new [`ReverseProxy`]. +pub struct Config { + /// Value to return when a call to the `system_name` JSON-RPC function is received. + pub system_name: Cow<'static, str>, + + /// Value to return when a call to the `system_version` JSON-RPC function is received. + pub system_version: Cow<'static, str>, + + /// Seed used for randomness. Used to avoid HashDoS attacks and to attribute clients and + /// requests to servers. + pub randomness_seed: [u8; 32], +} + +/// Configuration for a new [`ReverseProxy`]. +pub struct ClientConfig { + /// Maximum number of requests that haven't been answered yet that the client is allowed to + /// make. + pub max_unanswered_parallel_requests: usize, + + /// Maximum number of concurrent subscriptions for the legacy JSON-RPC API, except for + /// `author_submitAndWatchExtrinsic` which is handled by + /// [`ClientConfig::max_transactions_subscriptions`]. + pub max_legacy_api_subscriptions: usize, + + /// Maximum number of concurrent `chainHead_follow` subscriptions. + /// + /// The specification mentions that this value must be at least 2. If the value is inferior to + /// 2, it is raised. + pub max_chainhead_follow_subscriptions: usize, + + /// Maximum number of concurrent transactions-related subscriptions. + pub max_transactions_subscriptions: usize, + + /// Opaque data stored for this client. + /// Can later be accessed using the `ops::Index` trait implementation of [`ReverseProxy`]. + pub user_data: TClient, +} + +/// Reverse proxy state machine. See [..](the module-level documentation). +pub struct ReverseProxy { + /// List of all clients. Indices serve as [`ClientId`]. + clients: slab::Slab>, + + /// List of all servers. Indices serve as [`ServerId`]. + servers: slab::Slab>, + + /// Queues of requests waiting to be sent to a server. + /// Indexed by client and by server. + /// + /// The `VecDeque`s must never be empty. If a queue is emptied, the item must be removed from + /// the `BTreeMap` altogether. + // TODO: call shrink to fit from time to time? + client_requests_queued: BTreeMap<(ClientId, ServerTarget), VecDeque>, + + /// Same entries as [`ReverseProxy::client_requests_queued`], but indexed + /// differently. + /// [`ServerTarget::ServerAgnostic`] and [`ServerTarget::LegacyApiUnassigned`] both + /// correspond to a key of `None`, while [`ServerTarget::Specific`] corresponds to a key of + /// `Some`. + clients_with_request_queued: BTreeSet<(Option, ClientId)>, + + /// Alternative representation for [`Client::legacy_api_assigned_server`]. + legacy_api_server_assignments: BTreeSet<(ServerId, ClientId)>, + + /// List of all requests that have been extracted with + /// [`ReverseProxy::next_proxied_json_rpc_request`] and are being processed by the server. + /// + /// Entries are removed when a response is inserted with + /// [`ReverseProxy::insert_proxied_json_rpc_response`] or the server blacklisted through + /// [`ReverseProxy::blacklist_server`]. + /// + /// Keys are server IDs and the request ID from the point of view of the server. Values are + /// the client and request from the point of view of the client. + requests_in_progress: BTreeMap<(ServerId, String), (ClientId, QueuedRequest)>, + + /// List of all subscriptions that are currently active according to servers. + // TODO: clarify when entries are added/removed + active_subscriptions_by_server: BTreeMap<(ServerId, String), (ClientId, String)>, + + /// List of all subscriptions that are currently active according to clients. + /// + /// Entries are inserted when we queue for the client a JSON-RPC response containing a + /// subscription confirmation, and removed when we queue for the client a JSON-RPC response + /// containing an unsubscription confirmation. + /// + /// Contains `Some` only if the subscription is active on a server. + /// If it contains `None`, then there must be a subscription request either in queue or + /// currently being processed by a server. + active_subscriptions_by_client: + BTreeMap<(ClientId, String), (SubscriptionTyWithParams, Option<(ServerId, String)>)>, + + /// Source of randomness used for various purposes. + // TODO: is a crypto secure randomness overkill? + randomness: ChaCha20Rng, + + /// See [`Config::system_name`] + system_name: Cow<'static, str>, + + /// See [`Config::system_version`] + system_version: Cow<'static, str>, +} + +#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +enum ServerTarget { + ServerAgnostic, + LegacyApiUnassigned, + Specific(ServerId), +} + +// TODO: unnecessary? +impl ServerTarget { + const MIN: ServerTarget = ServerTarget::ServerAgnostic; + const MAX: ServerTarget = ServerTarget::Specific(ServerId(usize::MAX)); +} + +enum SubscriptionTyWithParams { + AuthorSubmitAndWatchExtrinsic, + ChainSubscribeAllHeads, + ChainSubscribeFinalizedHeads, + ChainSubscribeNewHeads, + StateSubscribeRuntimeVersion, + StateSubscribeStorage { keys: Vec }, + ChainHeadFollow, + TransactionSubmitAndWatch, +} + +enum SubscriptionTy { + AuthorSubmitAndWatchExtrinsic, + ChainSubscribeAllHeads, + ChainSubscribeFinalizedHeads, + ChainSubscribeNewHeads, + StateSubscribeRuntimeVersion, + StateSubscribeStorage, + ChainHeadFollow, + TransactionSubmitAndWatch, +} + +struct Client { + /// Returns the number of requests inserted with + /// [`ReverseProxy::insert_client_json_rpc_request`] whose response hasn't been pulled with + /// [`ReverseProxy::next_client_json_rpc_response`] yet. + num_unanswered_requests: usize, + + /// See [`ClientConfig::max_unanswered_parallel_requests`] + max_unanswered_parallel_requests: usize, + + /// Number of legacy JSON-RPC API subscriptions that are active, from the moment when the + /// request to subscribe is inserted in the queue to the moment when the response of the + /// unsubscription is extracted with [`ReverseProxy::next_client_json_rpc_response`]. + num_legacy_api_subscriptions: usize, + + /// See [`ClientConfig::max_legacy_api_subscriptions`]. + max_legacy_api_subscriptions: usize, + + /// Number of `chainHead_follow` subscriptions that are active, from the moment when the + /// request to subscribe is inserted in the queue to the moment when the response of the + /// unsubscription or the `stop` event is extracted with + /// [`ReverseProxy::next_client_json_rpc_response`]. + num_chainhead_follow_subscriptions: usize, + + /// See [`ClientConfig::max_chainhead_follow_subscriptions`]. + max_chainhead_follow_subscriptions: usize, + + /// Number of `author_submitAndWatchExtrinsic` and `transaction_unstable_submitAndWatch` + /// subscriptions that are active, from the moment when the request to subscribe is inserted + /// in the queue to the moment when the response of the unsubscription or the `dropped` event + /// is extracted with [`ReverseProxy::next_client_json_rpc_response`]. + num_transactions_subscriptions: usize, + + /// See [`ClientConfig::max_transactions_subscriptions`]. + max_transactions_subscriptions: usize, + + /// Queue of responses waiting to be sent to the client. + // TODO: call shrink to fit from time to time? + json_rpc_responses_queue: VecDeque, + + /// Server assigned to this client when it calls legacy JSON-RPC functions. Initially set + /// to `None`. A server is chosen the first time the client calls a legacy JSON-RPC function. + legacy_api_assigned_server: Option, + + /// Opaque data chosen by the API user. + /// + /// If `None`, then this client is considered non-existent for public-API-related purposes. + /// This value is set to `None` when the API user wants to remove a client, but that this + /// client still has active subscriptions that need to be cleaned up. + user_data: Option, +} + +struct QueuedRequest { + id_json: String, + + method: String, + + parameters_json: Option, + + ty: QueuedRequestTy, +} + +enum QueuedRequestTy { + Regular { + /// `true` if the JSON-RPC function belongs to the category of legacy JSON-RPC functions + /// that are all redirected to the same server. + is_legacy_api_server_specific: bool, + }, + Subscription { + ty: SubscriptionTyWithParams, + /// `Some` if the JSON-RPC client thinks that the subscription is already active, in which + /// case this field contains the client-side subscription ID. + assigned_subscription_id: Option, + }, + Unsubscribe(SubscriptionTy), +} + +struct QueuedResponse { + /// The JSON-RPC response itself. + response: String, + + /// If `true`, pulling this response decreases [`Client::num_unanswered_requests`]. + decreases_num_unanswered_requests: bool, +} + +struct Server { + /// `true` if the given server has misbehaved and must not process new requests. + is_blacklisted: bool, + + /// Opaque data chosen by the API user. + user_data: TServer, +} + +/// Identifier of a client within the [`ReverseProxy`]. +#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ClientId(usize); + +/// Identifier of a server within the [`ReverseProxy`]. +#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ServerId(usize); + +impl ReverseProxy { + /// Initializes a new [`ReverseProxy`]. + pub fn new(config: Config) -> Self { + ReverseProxy { + clients: slab::Slab::new(), // TODO: capacity + servers: slab::Slab::new(), // TODO: capacity + requests_in_progress: BTreeMap::new(), + randomness: ChaCha20Rng::from_seed(config.randomness_seed), + ..todo!() + } + } + + /// Adds a client to the list of clients managed by this state machine. + /// + /// API users should be aware of the fact that clients are distributed evenly between servers. + /// As more clients as added, the latency of the requests of each client is increased, unless + /// additional servers are added as well. + pub fn insert_client(&mut self, config: ClientConfig) -> ClientId { + ClientId(self.clients.insert(Client { + num_unanswered_requests: 0, + max_unanswered_parallel_requests: config.max_unanswered_parallel_requests, + num_legacy_api_subscriptions: 0, + max_legacy_api_subscriptions: config.max_legacy_api_subscriptions, + num_chainhead_follow_subscriptions: 0, + max_chainhead_follow_subscriptions: cmp::max( + config.max_chainhead_follow_subscriptions, + 2, + ), + num_transactions_subscriptions: 0, + max_transactions_subscriptions: config.max_transactions_subscriptions, + json_rpc_responses_queue: VecDeque::with_capacity(16), // TODO: capacity? + legacy_api_assigned_server: None, + user_data: Some(config.user_data), + })) + } + + /// Removes a client previously-inserted with [`ReverseProxy::insert_client`]. + /// + /// # Panic + /// + /// Panics if the given [`ClientId`] is invalid. + /// + // TODO: return list of servers that must wake up + pub fn remove_client(&mut self, client_id: ClientId) -> TClient { + let client = self + .clients + .get_mut(client_id.0) + .unwrap_or_else(|| panic!("client doesn't exist")); + let user_data = client + .user_data + .take() + .unwrap_or_else(|| panic!("client doesn't exist")); + + // Clear the queue of pending requests of this client. While leaving the queue as-is + // wouldn't be a problem in terms of logic, we clear it for optimization purposes, in + // order to avoid processing requests that we know are pointless. + // TODO: do it + + // For each subscription that was active on this client, push a request that unsubscribes. + for ((_, client_subscription_id), (server_id, server_subscription_id)) in self + .active_subscriptions_by_client + .range((client_id, String::new())..(ClientId(client_id.0 + 1), String::new())) + { + // TODO: + } + + // Try to remove the client entirely if possible. + self.try_remove_client(client_id); + + // Client successfully removed for API-related purposes. + user_data + } + + /// + /// + /// An error is returned if the JSON-RPC client has queued too many requests that haven't been + /// answered yet. Try again after a call to [`ReverseProxy::next_client_json_rpc_response`] + /// has returned `Some`. + /// + /// # Panic + /// + /// Panics if the given [`ClientId`] is invalid. + /// + pub fn insert_client_json_rpc_request( + &mut self, + client_id: ClientId, + request: &str, + ) -> Result { + // Check that the client ID is valid. + assert!(self.clients[client_id.0].user_data.is_some()); + + // Check the limit to the number of unanswered requests. + if self.clients[client_id.0].num_unanswered_requests + >= self.clients[client_id.0].max_unanswered_parallel_requests + { + return Err(InsertClientRequestError::TooManySimultaneousRequests); + } + self.clients[client_id.0].num_unanswered_requests += 1; + + // Determine the request information, or answer the request directly if possible. + let queued_request = match methods::parse_jsonrpc_client_to_server(request) { + Ok((request_id_json, method)) => { + let ty = match method { + // Answer the request directly if possible. + methods::MethodCall::system_name {} => { + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back(QueuedResponse { + response: methods::Response::system_name(Cow::Borrowed( + &*self.system_name, + )) + .to_json_response(request_id_json), + decreases_num_unanswered_requests: true, + }); + return Ok(InsertClientRequest::ImmediateAnswer); + } + methods::MethodCall::system_version {} => { + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back(QueuedResponse { + response: methods::Response::system_version(Cow::Borrowed( + &*self.system_version, + )) + .to_json_response(request_id_json), + decreases_num_unanswered_requests: true, + }); + return Ok(InsertClientRequest::ImmediateAnswer); + } + methods::MethodCall::sudo_unstable_version {} => { + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back(QueuedResponse { + response: methods::Response::sudo_unstable_version(Cow::Owned( + format!("{} {}", self.system_name, self.system_version), + )) + .to_json_response(request_id_json), + decreases_num_unanswered_requests: true, + }); + return Ok(InsertClientRequest::ImmediateAnswer); + } + methods::MethodCall::sudo_unstable_p2pDiscover { .. } => { + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back(QueuedResponse { + response: methods::Response::sudo_unstable_p2pDiscover(()) + .to_json_response(request_id_json), + decreases_num_unanswered_requests: true, + }); + return Ok(InsertClientRequest::ImmediateAnswer); + } + + // Subscription functions. + methods::MethodCall::state_subscribeRuntimeVersion {} + | methods::MethodCall::state_subscribeStorage { .. } + | methods::MethodCall::chain_subscribeAllHeads {} + | methods::MethodCall::chain_subscribeFinalizedHeads {} + | methods::MethodCall::chain_subscribeNewHeads {} => { + if self.clients[client_id.0].num_legacy_api_subscriptions + >= self.clients[client_id.0].max_legacy_api_subscriptions + { + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back(QueuedResponse { + response: parse::build_error_response( + request_id_json, + parse::ErrorResponse::ApplicationDefined( + -32800, + "Too many active subscriptions", + ), + None, + ), + decreases_num_unanswered_requests: true, + }); + return Ok(InsertClientRequest::ImmediateAnswer); + } + + self.clients[client_id.0].num_legacy_api_subscriptions += 1; + QueuedRequestTy::Subscription { + ty: match method { + methods::MethodCall::state_subscribeRuntimeVersion {} => { + SubscriptionTyWithParams::StateSubscribeRuntimeVersion + } + methods::MethodCall::state_subscribeStorage { list } => { + SubscriptionTyWithParams::StateSubscribeStorage { keys: list } + } + methods::MethodCall::chain_subscribeAllHeads {} => { + SubscriptionTyWithParams::ChainSubscribeAllHeads + } + methods::MethodCall::chain_subscribeFinalizedHeads {} => { + SubscriptionTyWithParams::ChainSubscribeFinalizedHeads + } + methods::MethodCall::chain_subscribeNewHeads {} => { + SubscriptionTyWithParams::ChainSubscribeNewHeads + } + _ => unreachable!(), + }, + assigned_subscription_id: None, + } + } + methods::MethodCall::chainHead_unstable_follow { .. } => { + if self.clients[client_id.0].num_chainhead_follow_subscriptions + >= self.clients[client_id.0].max_chainhead_follow_subscriptions + { + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back(QueuedResponse { + response: parse::build_error_response( + request_id_json, + parse::ErrorResponse::ApplicationDefined( + -32800, + "Too many active `chainHead_follow` subscriptions", + ), + None, + ), + decreases_num_unanswered_requests: true, + }); + return Ok(InsertClientRequest::ImmediateAnswer); + } + + self.clients[client_id.0].num_chainhead_follow_subscriptions += 1; + QueuedRequestTy::Subscription { + ty: SubscriptionTyWithParams::ChainHeadFollow, + assigned_subscription_id: None, + } + } + methods::MethodCall::author_submitAndWatchExtrinsic { .. } + | methods::MethodCall::transaction_unstable_submitAndWatch { .. } => { + if self.clients[client_id.0].num_transactions_subscriptions + >= self.clients[client_id.0].max_transactions_subscriptions + { + // TODO: send back error + todo!() + } + + self.clients[client_id.0].num_transactions_subscriptions += 1; + QueuedRequestTy::Subscription { + ty: match method { + methods::MethodCall::author_submitAndWatchExtrinsic { .. } => { + SubscriptionTyWithParams::ChainHeadFollow + } + methods::MethodCall::transaction_unstable_submitAndWatch { + .. + } => SubscriptionTyWithParams::TransactionSubmitAndWatch, + _ => unreachable!(), + }, + assigned_subscription_id: None, + } + } + + // Unsubscription functions. + methods::MethodCall::chain_unsubscribeAllHeads { subscription } + | methods::MethodCall::chain_unsubscribeFinalizedHeads { subscription } + | methods::MethodCall::chain_unsubscribeNewHeads { subscription } + | methods::MethodCall::state_unsubscribeRuntimeVersion { subscription } => { + todo!() + } + methods::MethodCall::state_unsubscribeStorage { subscription } => todo!(), + + // Legacy JSON-RPC API functions. + methods::MethodCall::account_nextIndex { .. } + | methods::MethodCall::author_hasKey { .. } + | methods::MethodCall::author_hasSessionKeys { .. } + | methods::MethodCall::author_insertKey { .. } + | methods::MethodCall::author_pendingExtrinsics { .. } + | methods::MethodCall::author_removeExtrinsic { .. } + | methods::MethodCall::author_rotateKeys { .. } + | methods::MethodCall::author_submitExtrinsic { .. } + | methods::MethodCall::author_unwatchExtrinsic { .. } + | methods::MethodCall::babe_epochAuthorship { .. } + | methods::MethodCall::chain_getBlock { .. } + | methods::MethodCall::chain_getBlockHash { .. } + | methods::MethodCall::chain_getFinalizedHead { .. } + | methods::MethodCall::chain_getHeader { .. } + | methods::MethodCall::childstate_getKeys { .. } + | methods::MethodCall::childstate_getStorage { .. } + | methods::MethodCall::childstate_getStorageHash { .. } + | methods::MethodCall::childstate_getStorageSize { .. } + | methods::MethodCall::grandpa_roundState { .. } + | methods::MethodCall::offchain_localStorageGet { .. } + | methods::MethodCall::offchain_localStorageSet { .. } + | methods::MethodCall::payment_queryInfo { .. } + | methods::MethodCall::rpc_methods { .. } + | methods::MethodCall::state_call { .. } + | methods::MethodCall::state_getKeys { .. } + | methods::MethodCall::state_getKeysPaged { .. } + | methods::MethodCall::state_getMetadata { .. } + | methods::MethodCall::state_getPairs { .. } + | methods::MethodCall::state_getReadProof { .. } + | methods::MethodCall::state_getRuntimeVersion { .. } + | methods::MethodCall::state_getStorage { .. } + | methods::MethodCall::state_getStorageHash { .. } + | methods::MethodCall::state_getStorageSize { .. } + | methods::MethodCall::state_queryStorage { .. } + | methods::MethodCall::state_queryStorageAt { .. } + | methods::MethodCall::system_accountNextIndex { .. } + | methods::MethodCall::system_addReservedPeer { .. } + | methods::MethodCall::system_chain { .. } + | methods::MethodCall::system_chainType { .. } + | methods::MethodCall::system_dryRun { .. } + | methods::MethodCall::system_health { .. } + | methods::MethodCall::system_localListenAddresses { .. } + | methods::MethodCall::system_localPeerId { .. } + | methods::MethodCall::system_networkState { .. } + | methods::MethodCall::system_nodeRoles { .. } + | methods::MethodCall::system_peers { .. } + | methods::MethodCall::system_properties { .. } + | methods::MethodCall::system_removeReservedPeer { .. } => { + QueuedRequestTy::Regular { + is_legacy_api_server_specific: true, + } + } + + // New JSON-RPC API. + methods::MethodCall::chainSpec_v1_chainName {} + | methods::MethodCall::chainSpec_v1_genesisHash {} + | methods::MethodCall::chainSpec_v1_properties {} + | methods::MethodCall::chainHead_unstable_finalizedDatabase { .. } => { + QueuedRequestTy::Regular { + is_legacy_api_server_specific: false, + } + } + + // ChainHead functions. + methods::MethodCall::chainHead_unstable_body { + follow_subscription, + hash, + } => todo!(), + methods::MethodCall::chainHead_unstable_call { + follow_subscription, + hash, + function, + call_parameters, + } => todo!(), + methods::MethodCall::chainHead_unstable_header { + follow_subscription, + hash, + } => todo!(), + methods::MethodCall::chainHead_unstable_stopOperation { + follow_subscription, + operation_id, + } => todo!(), + methods::MethodCall::chainHead_unstable_storage { + follow_subscription, + hash, + items, + child_trie, + } => todo!(), + methods::MethodCall::chainHead_unstable_continue { + follow_subscription, + operation_id, + } => todo!(), + methods::MethodCall::chainHead_unstable_unfollow { + follow_subscription, + } => todo!(), + methods::MethodCall::chainHead_unstable_unpin { + follow_subscription, + hash_or_hashes, + } => todo!(), + + methods::MethodCall::transaction_unstable_unwatch { subscription } => todo!(), + methods::MethodCall::network_unstable_subscribeEvents {} => todo!(), + methods::MethodCall::network_unstable_unsubscribeEvents { subscription } => { + todo!() + } + }; + + // TODO: to_owned? + QueuedRequest { + id_json: request_id_json.to_owned(), + method: method.name().to_owned(), + parameters_json: Some(method.params_to_json_object()), + ty, + } + } + + Err(methods::ParseClientToServerError::JsonRpcParse(_error)) => { + // Failed to parse the JSON-RPC request. + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back(parse::build_parse_error_response()); + return Ok(InsertClientRequest::ImmediateAnswer); + } + + Err(methods::ParseClientToServerError::Method { request_id, error }) => { + // JSON-RPC function not recognized. + + // Requests with an unknown method must not be blindly sent to a server, as it is + // not possible for the reverse proxy to guarantee that the logic of the request + // is respected. + // For example, if the request is a subscription request, the reverse proxy + // wouldn't be capable of understanding which client to redirect the notifications + // to. + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back(parse::build_error_response( + request_id, + parse::ErrorResponse::MethodNotFound, + None, + )); + return Ok(InsertClientRequest::ImmediateAnswer); + } + + Err(methods::ParseClientToServerError::UnknownNotification(function)) => { + // JSON-RPC function not recognized, and the call is a notification. + // According to the JSON-RPC specification, the server must not send any response + // to notifications, even in case of an error. + return Ok(InsertClientRequest::Discarded); + } + }; + + Ok(todo!()) + } + + /// Returns the next JSON-RPC response or notification to send to the given client. + /// + /// Returns `None` if none is available. + /// + /// The return type of [`ReverseProxy::insert_proxied_json_rpc_response`] indicates if a + /// JSON-RPC response or notification has become available for a client. + /// + /// # Panic + /// + /// Panics if the given [`ClientId`] is invalid. + /// + pub fn next_client_json_rpc_response(&mut self, client_id: ClientId) -> Option { + assert!(self.clients[client_id.0].user_data.is_some()); + + let response = self.clients[client_id.0] + .json_rpc_responses_queue + .pop_front()?; + + if response.decreases_num_unanswered_requests { + self.clients[client_id.0].num_unanswered_requests -= 1; + } + + Some(response.response) + } + + /// Adds a new server to the list of servers. + #[cold] + pub fn insert_server(&mut self, user_data: TServer) -> ServerId { + ServerId(self.servers.insert(Server { + is_blacklisted: false, + user_data, + })) + } + + /// Removes a server from the list of servers. + /// + /// All active subscriptions and requests are either stopped or redirected to a different + /// server. + /// + /// After this function returns, [`ReverseProxy::next_proxied_json_rpc_request`] should be + /// called with all idle servers, in order to pick up the requests that this server was + /// processing. + /// + /// # Panic + /// + /// Panics if the given [`ServerId`] is invalid. + /// + // TODO: return list of clients that have a response available + #[cold] + pub fn remove_server(&mut self, server_id: ServerId) -> TServer { + self.blacklist_server(server_id); + self.servers.remove(server_id.0).user_data + } + + // TODO: return list of clients that have a response available + #[cold] + fn blacklist_server(&mut self, server_id: ServerId) { + // Set `is_blacklisted` to `true`, and return immediately if it was already `true`. + if mem::replace(&mut self.servers[server_id.0].is_blacklisted, true) { + return; + } + + // Extract from `active_subscriptions` the subscriptions that were handled by that server. + let subscriptions_to_cancel_or_reopen = { + let mut server_and_after = self + .active_subscriptions_by_server + .split_off(&(server_id, String::new())); + let mut after = server_and_after.split_off(&(ServerId(server_id.0 + 1), String::new())); + self.active_subscriptions_by_server.append(&mut after); + server_and_after + }; + + // Find in the subscriptions that were handled by that server the subscriptions that can + // be cancelled by sending a notification to the client. + // If the client happened to have a request in queue that concerns that subscription, + // this guarantees that the notification about the cancellation is sent to the client + // before the responses to this request. + // For example, if the client has queued a `chainHead_unstable_header` request, it will + // receive the `stop` event of the `chainHead_unstable_follow` subscription before + // receiving the error response to the `chainHead_unstable_header` request. + // While this ordering is in no way a requirement, it is more polite to do so. + for ((_, server_subscription_id), (client_id, client_subscription_id, subscription_type)) in + &subscriptions_to_cancel_or_reopen + { + match subscription_type { + // Any active `chainHead_follow`, `transaction_submitAndWatch`, or + // `author_submitAndWatchExtrinsic` subscription is killed. + SubscriptionTyWithParams::AuthorSubmitAndWatchExtrinsic => { + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back( + methods::ServerToClient::author_extrinsicUpdate { + subscription: client_subscription_id.into(), + result: methods::TransactionStatus::Dropped, + } + .to_json_request_object_parameters(None), + ); + let _was_removed = self + .active_subscriptions_by_client + .remove(&(client_id, client_subscription_id.clone())); + debug_assert!(_was_removed.is_some()); + } + SubscriptionTyWithParams::TransactionSubmitAndWatch => { + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back( + methods::ServerToClient::transaction_unstable_watchEvent { + subscription: client_subscription_id.into(), + result: methods::TransactionWatchEvent::Dropped { + // Unfortunately, there is no way of knowing whether the server has + // broadcasted the transaction. Since `false` offers guarantees + // but `true` doesn't, we opt to always send back `true`. + // TODO: change the RPC spec to handle this more properly? + broadcasted: true, + error: "Proxied server gone".into(), + }, + } + .to_json_request_object_parameters(None), + ); + let _was_removed = self + .active_subscriptions_by_client + .remove(&(client_id, client_subscription_id.clone())); + debug_assert!(_was_removed.is_some()); + } + SubscriptionTyWithParams::ChainHeadFollow => { + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back( + methods::ServerToClient::chainHead_unstable_followEvent { + subscription: client_subscription_id.into(), + result: methods::FollowEvent::Stop {}, + } + .to_json_request_object_parameters(None), + ); + let _was_removed = self + .active_subscriptions_by_client + .remove(&(client_id, client_subscription_id.clone())); + debug_assert!(_was_removed.is_some()); + } + + // Other subscription types are handled below. + SubscriptionTyWithParams::ChainSubscribeAllHeads + | SubscriptionTyWithParams::ChainSubscribeFinalizedHeads + | SubscriptionTyWithParams::ChainSubscribeNewHeads + | SubscriptionTyWithParams::StateSubscribeRuntimeVersion + | SubscriptionTyWithParams::StateSubscribeStorage { .. } => {} + } + } + + // The server-specific requests that were queued for this server and the requests that + // were already sent to the server are processed the same way, as from the point of view + // of the JSON-RPC client there's no possible way to differentiate the two. + let requests_to_cancel = { + // Extract from `client_with_requests_queued` the list of clients with pending + // requests that can only target that server. + let client_with_requests_queued = { + let mut server_and_after = self + .clients_with_request_queued + .split_off(&(Some(server_id), ClientId(usize::MIN))); + let mut after = server_and_after + .split_off(&(Some(ServerId(server_id.0 + 1)), ClientId(usize::MIN))); + self.clients_with_request_queued.append(&mut after); + server_and_after + }; + + // Extract from `requests_in_progress` the requests that were being processed by + // that server. + let requests_in_progress = { + let mut server_and_after = self + .requests_in_progress + .split_off(&(server_id, String::new())); + let mut after = + server_and_after.split_off(&(ServerId(server_id.0 + 1), String::new())); + self.requests_in_progress.append(&mut after); + server_and_after + }; + + let requests_queued = + client_with_requests_queued + .into_iter() + .flat_map(|(_, client_id)| { + self.client_requests_queued + .remove(&(client_id, ServerTarget::Specific(server_id))) + .unwrap_or_else(|| unreachable!()) + .into_iter() + .map(|rq| (client_id, rq)) + }); + let requests_dispatched = requests_in_progress + .into_iter() + .map(|(_, (client_id, rq))| (client_id, rq)); + requests_dispatched.chain(requests_queued) + }; + + for (client_id, request_info) in requests_to_cancel { + // For the sake of simplicity, we don't special-case the situation where the client + // has been removed by the API user, as that would duplicate several code paths. + // TODO: should try remove client nonetheless, otherwise leak + + // Unsubscription requests are immediately processed. + if matches!(request_info.ty, QueuedRequestTy::Unsubscribe(unsub_ty)) { + // TODO: are there fake unsubscription requests? + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back(QueuedResponse { + response: match unsub_ty { + _ => todo!(), + }, + decreases_num_unanswered_requests: true, + }); + continue; + } + + // Any pending request targetting a `chainHead_follow` subscription is answered + // immediately, as a `stop` event has been generated above. + // TODO: + + // Any other request is added back to the head of the queue of its JSON-RPC client. + self.clients[client_id.0] + .server_agnostic_requests_queue + .push_front(request_info); + } + + // Process a second time the subscriptions to cancel, this time reopening legacy JSON-RPC + // API subscriptions by adding to the head of the JSON-RPC client requests queue a fake + // subscription request. + // This is done at the end, in order to avoid reopening subscriptions for which an + // unsubscribe request was in queue. + for ((_, server_subscription_id), (client_id, client_subscription_id, subscription_type)) in + subscriptions_to_cancel_or_reopen + { + match subscription_type { + SubscriptionTyWithParams::ChainSubscribeAllHeads + | SubscriptionTyWithParams::ChainSubscribeFinalizedHeads + | SubscriptionTyWithParams::ChainSubscribeNewHeads + | SubscriptionTyWithParams::StateSubscribeRuntimeVersion + | SubscriptionTyWithParams::StateSubscribeStorage { .. } => { + self.active_subscriptions_by_client + .get_mut(&(client_id, client_subscription_id.clone())) + .unwrap() + .1 = None; + } + + // Already handled above. + SubscriptionTyWithParams::AuthorSubmitAndWatchExtrinsic + | SubscriptionTyWithParams::TransactionSubmitAndWatch + | SubscriptionTyWithParams::ChainHeadFollow => {} + } + } + + // Unassign clients that were assigned to that server. + let legacy_api_assigned_clients = { + let mut server_and_after = self + .legacy_api_server_assignments + .split_off(&(server_id, ClientId(usize::MIN))); + let mut after = + server_and_after.split_off(&(ServerId(server_id.0 + 1), ClientId(usize::MIN))); + self.legacy_api_server_assignments.append(&mut after); + server_and_after + }; + for (_, legacy_api_assigned_client) in legacy_api_assigned_clients { + debug_assert_eq!( + self.clients[legacy_api_assigned_client.0].legacy_api_assigned_server, + Some(server_id) + ); + self.clients[legacy_api_assigned_client.0].legacy_api_assigned_server = None; + } + } + + /// Pick a JSON-RPC request waiting to be processed. + /// + /// Returns `None` if no JSON-RPC request is waiting to be processed. + /// + /// If `None` is returned, you should try calling this function again after + /// [`ReverseProxy::insert_client_json_rpc_request`] or [`ReverseProxy::remove_client`]. + /// + /// If `Some(_, Some(_))` is returned, then the pulled request belongs to the given client, + /// and [`ReverseProxy::insert_client_json_rpc_request`] can be called with that client + /// with the guarantee that there is space for a request. + /// + /// If `Some(_, None)` is returned, the pulled request is related to the internal maintenance + /// of the [`ReverseProxy`]. + /// + /// Note that the [`ReverseProxy`] state machine doesn't enforce any limit to the number of + /// JSON-RPC requests that a server processes simultaneously. A JSON-RPC server is expected to + /// back-pressure its socket once it gets too busy, in which case + /// [`ReverseProxy::next_proxied_json_rpc_request`] should no longer be called until the + /// server is ready to accept more data. + /// This ensures that for example a JSON-RPC server that is twice as powerful compared to + /// another one should get approximately twice the number of requests. + /// + /// # Panic + /// + /// Panics if the given [`ServerId`] is invalid. + /// + pub fn next_proxied_json_rpc_request( + &mut self, + server_id: ServerId, + ) -> Option<(String, Option)> { + let server = &mut self.servers[server_id.0]; + if server.is_blacklisted { + return None; + } + + // In order to guarantee fairness between the clients, choosing which request to send to + // the server is done in two steps: first, pick a client that has at least one request + // waiting to be sent, then pick the first request from that client's queue. This + // guarantees that a greedy client can't starve the other clients. + // There are two types of requests: requests that aren't attributed to any server, and + // requests that are attributed to a specific server. For this reason, the list of clients + // to pick a request from depends on the server. This complicates fairness. + // To solve this problem, we join two lists together: the list of clients with at least one + // server-agnostic request waiting, and the list of clients with at least one + // server-specific request waiting. The second list is weighted based on the total number + // of clients and servers. The client to pick a request from is picked randomly from the + // concatenation of these two lists. + + loop { + // Choose the client to pick a request from. + let (client_with_request, pick_from_server_specific) = { + let mut clients_with_server_specific_request = self + .clients_with_request_queued + .range( + (Some(server_id), ClientId(usize::MIN)) + ..=(Some(server_id), ClientId(usize::MAX)), + ) + .map(|(_, client_id)| *client_id); + let mut clients_with_server_agnostic_request = self + .clients_with_request_queued + .range((None, ClientId(usize::MIN))..=(None, ClientId(usize::MAX))) + .map(|(_, client_id)| *client_id); + let clients_with_server_agnostic_request_len = + clients_with_server_agnostic_request.clone().count(); + let server_specific_weight: usize = + 1 + (self.clients.len().saturating_sub(1) / self.servers.len()); + // While we could in theory use `rand::seq::IteratorRandom` with something + // like `(0..server_specific_weight).flat_map(...)`, it's hard to guarantee + // that doing so would be `O(1)`. Since we want this guarantee, we do it manually. + let total_weight = clients_with_server_agnostic_request_len.saturating_add( + server_specific_weight + .saturating_mul(clients_with_server_specific_request.clone().count()), + ); + let index = self.randomness.gen_range(0..total_weight); + if index < clients_with_server_agnostic_request_len { + let client = *clients_with_server_agnostic_request.nth(index).unwrap(); + (client, false) + } else { + let client = clients_with_server_specific_request + .nth( + (index - clients_with_server_agnostic_request_len) + / server_specific_weight, + ) + .unwrap(); + (client, true) + } + }; + + // Extract a request from that client. + let queued_request = if pick_from_server_specific { + // Pick a request from the server-agnostic queue of that client. + let Some(requests_queue) = self + .client_requests_queued + .get_mut(&(client_with_request, ServerTarget::Specific(server_id))) + else { + // A panic here indicates a bug in the code. + unreachable!() + }; + + let Some(request) = requests_queue.pop_front() else { + // A panic here indicates a bug in the code. + unreachable!() + }; + + // We need to update caches if this was the last request in queue. + if requests_queue.is_empty() { + self.client_requests_queued + .remove(&(client_with_request, ServerTarget::Specific(server_id))); + self.clients_with_request_queued + .remove(&(Some(server_id), client_with_request)); + } + + request + } else { + /// Pick a request from the server-specific queue of that client. + /// We currently always prefer the `LegacyApiUnassigned` requests queue over the + /// `ServerAgnostic` requests queue, for no specific reason except avoiding + /// making the implementation too complex. + let (queue, is_legacy_api_unassigned) = if let Some(queue) = self + .client_requests_queued + .get_mut(&(client_with_request, ServerTarget::LegacyApiUnassigned)) + { + (queue, true) + } else if let Some(queue) = self + .client_requests_queued + .get_mut(&(client_with_request, ServerTarget::ServerAgnostic)) + { + (queue, false) + } else { + // A panic here indicates a bug in the code. + unreachable!() + }; + + // Extract the request. + let Some(request) = queue.pop_front() else { + // As documented, queues must always be non-empty, otherwise they should have + // been removed altogether. + unreachable!() + }; + + // If the request is a legacy API server-specific request, assign the client to + // the server. + if is_legacy_api_unassigned { + debug_assert!(self.clients[client_with_request.0] + .legacy_api_assigned_server + .is_none()); + self.clients[client_with_request.0].legacy_api_assigned_server = + Some(server_id); + let _was_inserted = self + .legacy_api_server_assignments + .insert((server_id, client_with_request)); + debug_assert!(_was_inserted); + } + + // Update the local state, either by removing the queue if it is empty, or, now + // that the client is assigned a server, by merging it with the server-specific + // queue. + if queue.is_empty() { + self.client_requests_queued.remove(&( + client_with_request, + if is_legacy_api_unassigned { + ServerTarget::LegacyApiUnassigned + } else { + ServerTarget::ServerAgnostic + }, + )); + if !self.client_requests_queued.contains_key(&( + client_with_request, + if is_legacy_api_unassigned { + ServerTarget::ServerAgnostic + } else { + ServerTarget::LegacyApiUnassigned + }, + )) { + self.clients_with_request_queued + .remove(&(client_with_request, None)); + } + } else if is_legacy_api_unassigned { + // Queue is non-empty, and client was assigned to that server. + // We need to move around the queue of unassigned legacy API requests so that + // it targets the server. + let queue = mem::take(queue); + debug_assert!(!queue.is_empty()); + self.client_requests_queued + .entry((client_with_request, ServerTarget::Specific(server_id))) + .or_insert_with(|| VecDeque::new()) + .extend(queue); + self.clients_with_request_queued + .insert((Some(server_id), client_with_request)); + self.client_requests_queued + .remove(&(client_with_request, ServerTarget::LegacyApiUnassigned)); + if !self + .client_requests_queued + .contains_key(&(client_with_request, ServerTarget::ServerAgnostic)) + { + let _was_removed = self + .clients_with_request_queued + .remove(&(None, client_with_request)); + debug_assert!(_was_removed); + } + } + + request + }; + + // At this point, we have extracted a request from the queue. + + // The next step is to generate a new request ID and rewrite the request in order to + // change the request ID. + let new_request_id = hex::encode({ + let mut bytes = [0; 48]; + self.randomness.fill_bytes(&mut bytes); + bytes + }); + let request_with_adjusted_id = parse::build_request(&parse::Request { + id_json: Some(&new_request_id), + method: &queued_request.method, + params_json: queued_request.parameters_json.as_deref(), + }); + + // Update `self` to track that the server is processing this request. + let _previous_value = self.requests_in_progress.insert( + (server_id, new_request_id), + (client_with_request, queued_request), + ); + debug_assert!(_previous_value.is_none()); + + // Success. + break Some(( + request_with_adjusted_id, + if self.clients[client_with_request.0].user_data.is_some() { + Some(client_with_request) + } else { + None + }, + )); + } + } + + /// Inserts a response or notification sent by a server. + /// + /// Note that there exists no back-pressure system here. Responses and notifications sent by + /// servers are always accepted and buffered in order to be picked up later + /// by [`ReverseProxy::next_client_json_rpc_response`]. + /// + /// This cannot lead to an excessive memory usage, because the number of responses is bounded + /// by the maximum number of in-flight JSON-RPC requests enforced on clients, and the + /// number of notifications is bounded by the maximum number of active subscriptions enforced + /// on clients. The state machine might merge multiple notifications from the same + /// subscription into one in order to enforce this bound. + /// + /// # Panic + /// + /// Panics if the given [`ServerId`] is invalid. + /// + pub fn insert_proxied_json_rpc_response( + &mut self, + server_id: ServerId, + response: &str, + ) -> InsertProxiedJsonRpcResponse { + match parse::parse_response(response) { + Ok(parse::Response::ParseError { .. }) + | Ok(parse::Response::Error { + error_code: -32603, // Internal server error. + .. + }) => { + // JSON-RPC server has returned an "internal server error" or indicates that it + // has failed to parse our JSON-RPC request as a valid request. This is never + // supposed to happen and indicates that something is very wrong with the server. + + // The server is blacklisted. While the response might correspond to a request, + // we do the blacklisting without removing that request from the state, as the + // blacklisting will automatically remove all requests. + self.blacklist_server(server_id); + InsertProxiedJsonRpcResponse::Blacklisted("") // TODO: + } + + Ok(parse::Response::Success { + id_json, + result_json, + }) => { + // Find in our local state the request being answered. + // Because clients are removed only after all their in-progress requests have been + // answered, it is guaranteed that the client is still in the list. + // TODO: to_owned overhead + let Some((client_id, request_info)) = self + .requests_in_progress + .remove(&(server_id, id_json.to_owned())) + else { + // Server has answered a non-existing request. Blacklist it. + self.blacklist_server(server_id); + return InsertProxiedJsonRpcResponse::Blacklisted(""); + // TODO: ^ + }; + + match request_info.ty { + QueuedRequestTy::Subscription { + ty, + assigned_subscription_id, + } => {} + QueuedRequestTy::Unsubscribe(_) => {} + QueuedRequestTy::Regular { + is_legacy_api_server_specific, + } => {} + } + + // TODO: add subscription if this was a subscribe request + // TODO: remove subscription if this was an unsubscribe request + + // It is possible that this response concerns a client that has been + // destroyed by the API user, in which case we simply discard the + // response. + if self.clients[client_id.0].user_data.is_none() { + // Remove the client for real if it was previously removed by the API user and + // that this was its last request. + self.try_remove_client(client_id); + return InsertProxiedJsonRpcResponse::Discarded; + } + + // Rewrite the request ID found in the response in order to match what the + // client expects. + let response_with_translated_id = + parse::build_success_response(&request_info.id_json, result_json); + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back(response_with_translated_id); + + // Success. + InsertProxiedJsonRpcResponse::Ok(client_id) + } + + Ok(parse::Response::Error { + id_json, + error_code, + error_message, + error_data_json, + }) => { + // TODO: if this was a "fake subscription" request, re-queue it + + // JSON-RPC server has returned an error for this JSON-RPC call. + // TODO: translate ID + parse::build_error_response( + id_json, + parse::ErrorResponse::ApplicationDefined(error_code, error_message), + error_data_json, + ); + // TODO: ?! + } + + Err(response_parse_error) => { + // If the message couldn't be parsed as a response, attempt to parse it as a + // notification. + match methods::parse_notification(response) { + Ok(mut notification) => { + // This is a subscription notification. + // Because clients are removed only after they have finished + // unsubscribing, it is guaranteed that the client is still in the + // list. + // TODO: overhead of into_owned + let Some((client_id, client_notification_id)) = self + .active_subscriptions_by_server + .get(&(server_id, notification.subscription().clone().into_owned())) + else { + // The subscription ID isn't recognized. This indicates something very + // wrong with the server. We handle this by blacklisting the server. + self.blacklist_server(server_id); + return InsertProxiedJsonRpcResponse::Blacklisted(""); + // TODO: ^ + }; + + // It is possible that this notification concerns a client that has been + // destroyed by the API user, in which case we simply discard the + // notification. + if self.clients[client_id.0].user_data.is_none() { + return InsertProxiedJsonRpcResponse::Discarded; + } + + // Rewrite the subscription ID in the notification in order to match what + // the client expects. + notification.set_subscription(Cow::Borrowed(&client_notification_id)); + let rewritten_notification = + notification.to_json_request_object_parameters(None); + // TODO: must handle situation where client doesn't pull its data + self.clients[client_id.0] + .json_rpc_responses_queue + .push_back(QueuedResponse { + response: rewritten_notification, + decreases_num_unanswered_requests: false, + }); + + // Success + InsertProxiedJsonRpcResponse::Ok(*client_id) + } + + Err(notification_parse_error) => { + // Failed to parse the message from the JSON-RPC server. + self.blacklist_server(server_id); + InsertProxiedJsonRpcResponse::Blacklisted("") + // TODO: ^ + } + } + } + } + } + + /// Checks if the given client has been removed using [`ReverseProxy::remove_client`] and that + /// it has no request in progress and no active subscription, and if so removes it entirely + /// from the state of `self`. + fn try_remove_client(&mut self, client_id: ClientId) { + if self.clients[client_id.0].user_data.is_some() { + return; + } + + if self.clients[client_id.0].num_unanswered_requests != 0 { + return; + } + + debug_assert!(!self + .clients_with_server_agnostic_request_waiting + .contains(&client_id)); + debug_assert!(self + .client_requests_queued + .range((client_id, ServerTarget::MIN)..(ClientId(client_id.0 + 1), ServerTarget::MIN)) + .next() + .is_none()); + + if self + .active_subscriptions_by_client + .range((client_id, String::new())..(ClientId(client_id.0 + 1), String::new())) + .next() + .is_some() + { + return; + } + + let removed_client = self.clients.remove(client_id.0); + + if let Some(legacy_api_assigned_server) = removed_client.legacy_api_assigned_server { + let _was_removed = self + .legacy_api_server_assignments + .remove(&(server_id, client_id)); + debug_assert!(_was_removed); + } + } +} + +impl ops::Index for ReverseProxy { + type Output = TClient; + + fn index(&self, id: ClientId) -> &TClient { + self.clients[id.0].user_data.as_ref().unwrap() + } +} + +impl ops::IndexMut for ReverseProxy { + fn index_mut(&mut self, id: ClientId) -> &mut TClient { + self.clients[id.0].user_data.as_mut().unwrap() + } +} + +impl ops::Index for ReverseProxy { + type Output = TServer; + + fn index(&self, id: ServerId) -> &TServer { + &self.servers[id.0].user_data + } +} + +impl ops::IndexMut for ReverseProxy { + fn index_mut(&mut self, id: ServerId) -> &mut TServer { + &mut self.servers[id.0].user_data + } +} + +/// Outcome of a call to [`ReverseProxy::insert_client_json_rpc_request`]. +#[derive(Debug)] +pub enum InsertClientRequest { + /// The request has been silently discarded. + /// + /// This happens for example if the JSON-RPC client sends a notification. + Discarded, + + /// The request has been immediately answered or discarded and doesn't need any + /// further processing. + /// [`ReverseProxy::next_client_json_rpc_response`] should be called in order to pull the + /// response and send it to the client. + ImmediateAnswer, + + /// The request can be processed by any server. + /// [`ReverseProxy::next_proxied_json_rpc_request`] should be called with any [`ServerId`]. + AnyServerWakeUp, + + /// The request must be processed specifically by the indicated server. + /// [`ReverseProxy::next_proxied_json_rpc_request`] should be called with the + /// given [`ServerId`]. + ServerWakeUp(ServerId), +} + +/// Error potentially returned by a call to [`ReverseProxy::insert_client_json_rpc_request`]. +#[derive(Debug, derive_more::Display, Clone)] +pub enum InsertClientRequestError { + /// The number of requests that this client has queued is already equal to + /// [`ClientConfig::max_unanswered_parallel_requests`]. + #[display(fmt = "Too many simultaneous requests")] + TooManySimultaneousRequests, +} + +pub enum InsertProxiedJsonRpcResponse { + Ok(ClientId), + Discarded, + Blacklisted(&'static str), +} diff --git a/lib/src/json_rpc/servers_multiplexer.rs b/lib/src/json_rpc/servers_multiplexer.rs new file mode 100644 index 0000000000..42f8652a9d --- /dev/null +++ b/lib/src/json_rpc/servers_multiplexer.rs @@ -0,0 +1,1853 @@ +// Smoldot +// Copyright (C) 2024 Pierre Krieger +// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! Accept an incoming stream of JSON-RPC requests, and distributes them agmonst multiple +//! JSON-RPC servers. +//! +//! The [`ServersMultiplexer`] struct contains a list of servers, a queue of JSON-RPC requests, +//! and a queue of JSON-RPC responses. +//! +//! If a message sent by a JSON-RPC server is deemed to be incorrect, the server might be +//! blacklisted depending on the circumstance. No request will be sent to a blacklisted server, +//! and its messages will be ignored. +//! +//! Because multiple servers might accidentally assign the subscription ID, the subscription ID +//! of all subscriptions is rewritten by the [`ServersMultiplexer`]. In other words, the +//! subscription ID that the client observes are assigned by the [`ServersMultiplexer`]. +//! Requests IDs are left untouched. +//! +//! When a server is removed or blacklisted, for each legacy JSON-RPC API subscription that this +//! server was handling (apart from `author_submitAndWatchExtrinsic`), a dummy subscription request +//! is sent to a different server and the notifications sent by this other server are +//! transparently redirected to the client as if they were coming from the original server. +//! This might lead to some confusing situations, such as the latest finalized block going back +//! to an earlier block, but because the legacy JSON-RPC API doesn't provide any way to handle +//! this situation in a clean way, this is the last bad way to handle it. +//! +//! When a server is removed or blacklisted, each active `chainHead_v1_follow` subscription +//! generates a `stop` event, and each active `author_submitAndWatchExtrinsic` and +//! `transactionWatch_v1_submitAndWatch` subscription generates a `dropped` event. +//! +//! JSON-RPC requests for `chainHead_v1_follow` and `transaction_unstable_broadcast` are +//! sent to a randomly-chosen server. If this server returns `null`, indicating that it has reached +//! its limits, the request is sent to a different randomly-chosen server instead. After 3 failed +//! attempts, `null` is returned to the client. +// TODO: document transaction_broadcast when server is removed +// TODO: more doc + +// TODO: what about rpc_methods? should we not query servers for the methods they support or something? + +use alloc::{ + borrow::{Cow, ToOwned as _}, + collections::{btree_map, BTreeMap, BTreeSet, VecDeque}, + format, + string::String, + sync::Arc, +}; +use core::{fmt, mem, ops}; +use rand_chacha::{ + rand_core::{RngCore as _, SeedableRng as _}, + ChaCha20Rng, +}; + +use crate::{ + json_rpc::{methods, parse}, + util::SipHasherBuild, +}; + +/// Configuration for a new [`ServersMultiplexer`]. +pub struct Config { + /// Number of entries to pre-allocate for the list of servers. + pub servers_capacity: usize, + + /// Number of entries to pre-allocate for the list of requests queued or in progress. + pub requests_capacity: usize, + + /// Number of entries to pre-allocate for the list of active subscriptions. + pub subscriptions_capacity: usize, + + /// Value to return when a call to the `system_name` JSON-RPC function is received. + pub system_name: Cow<'static, str>, + + /// Value to return when a call to the `system_version` JSON-RPC function is received. + pub system_version: Cow<'static, str>, + + /// Seed used for randomness. Used to avoid HashDoS attacks and to attribute clients and + /// requests to servers. + pub randomness_seed: [u8; 32], +} + +/// Identifier of a server within the [`ServersMultiplexer`]. +#[derive(Debug, Copy, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct ServerId(usize); + +/// See [the module-level documentation](..). +// TODO: can we replace String with Arc in some of these fields? +pub struct ServersMultiplexer { + /// List of all servers. Indices serve as [`ServerId`]. + servers: slab::Slab>, + + /// Queues of requests waiting to be sent to a server. + /// + /// Entries are the request ID (in JSON), and the request information. + /// + /// Indexed by `None` if the request doesn't have to target any specific server, or by `Some` + /// if the request must target a specific server. + /// + /// The `VecDeque`s must never be empty. If a queue is emptied, the item must be removed from + /// the `BTreeMap` altogether. + queued_requests: BTreeMap, VecDeque<(String, Request)>>, + + /// List of all requests that have been extracted with + /// [`ServersMultiplexer::next_proxied_json_rpc_request`] and are being processed by a server. + /// + /// Entries are removed when a response is inserted with + /// [`ServersMultiplexer::insert_proxied_json_rpc_response`] or the server blacklisted through + /// [`ServersMultiplexer::blacklist_server`]. + /// + /// Keys are server IDs and the request ID (in JSON). + requests_in_progress: BTreeMap<(ServerId, String), Request>, + + /// Queue of responses waiting to be sent to the client. + responses_queue: VecDeque, + + /// List of all subscriptions that are currently active. + /// + /// Keys are subscription IDs from the client's perspective, and values are the server that is + /// handling this subscription and the subscription ID from the server's perspective. + /// + /// Entries are inserted when a server successfully accepts a subscription request, and removed + /// when a server sends back a confirmation of unsubscription, or a `stop` or `dropped` event + /// or similar. In other words, entries are removed only once we don't expect any new + /// notification coming from the server. + /// + /// This collection does **not** include subscriptions that the client thinks are active but + /// that aren't active on any server, which can happen after a server is removed. + active_subscriptions: + hashbrown::HashMap, (ServerId, ActiveSubscription), SipHasherBuild>, + + /// Same entries as [`ServersMultiplexer::active_subscriptions`], but indexed by server. + /// + /// Keys are the server and subscription ID from the server's perspective, and values are + /// the subscription ID from the client's perspective. + active_subscriptions_by_server: BTreeMap<(ServerId, Arc), Arc>, + + /// List of subscriptions that the client thinks are alive but that aren't active on + /// any server. + /// + /// Keys are the subscription ID from the point of view of the client. + zombie_subscriptions: hashbrown::HashMap, ZombieSubscription, SipHasherBuild>, + + /// Subset of the items of [`ServersMultiplexer::zombie_subscriptions`] for which a + /// re-subscription is waiting to be sent to a server. + /// + /// Keys are the subscription ID from the point of view of the client. + zombie_subscriptions_pending: BTreeSet>, + + /// Subset of the items of [`ServersMultiplexer::zombie_subscriptions`] for which a + /// re-subscription has been sent to a server. + /// + /// Keys are the request ID of the re-subscription request, and values are the subscription + /// ID from the point of view of the client. + zombie_subscriptions_by_resubscribe_id: BTreeMap<(ServerId, String), Arc>, + + /// See [`Config::system_name`] + system_name: Cow<'static, str>, + + /// See [`Config::system_version`] + system_version: Cow<'static, str>, + + /// Source of randomness used for various purposes. + randomness: ChaCha20Rng, +} + +struct Request { + /// Name of the method of the request. + method_name: String, + + /// JSON-encoded parameters of the request. `None` if no parameters were provided. + /// + /// If the request targets a specific subscription (i.e. contains a subscription ID in its + /// parameters), the parameters have been adjusted to target the subscription ID from the + /// server's point of view. + method_params_json: Option, + + /// Number of times this request has previously been sent to a server, and the server has + /// rejected it. + previous_failed_attempts: u32, +} + +struct Server { + /// `true` if the given server has misbehaved and must not process new requests. + is_blacklisted: bool, + + /// Opaque data chosen by the API user. + user_data: T, +} + +struct ActiveSubscription { + /// Subscription ID according to the server. + server_subscription_id: Arc, + + /// Name of the method that has performed the subscription. + method_name: String, + + /// JSON-encoded parameters to the method that performed the subscription. `None` if no + /// parameters were provided. + method_params_json: Option, +} + +struct ZombieSubscription { + /// Name of the method that has performed the subscription. + method_name: String, + + /// JSON-encoded parameters to the method that performed the subscription. `None` if no + /// parameters were provided. + method_params_json: Option, + + resubscribe_request_id: Option, + + /// If `Some`, the client has sent an unsubscribe request for this subscription but that + /// hasn't been answered yet. Immediately after the subscription has been re-subscribed, + /// a unsubscribe request must be sent to the server. + unsubscribe_request_id: Option, +} + +impl ServersMultiplexer { + /// Creates a new multiplexer with an empty list of servers. + pub fn new(config: Config) -> Self { + let mut randomness = ChaCha20Rng::from_seed(config.randomness_seed); + + ServersMultiplexer { + servers: slab::Slab::with_capacity(config.servers_capacity), + queued_requests: BTreeMap::new(), + responses_queue: VecDeque::with_capacity(config.requests_capacity), + requests_in_progress: BTreeMap::new(), + active_subscriptions: hashbrown::HashMap::with_capacity_and_hasher( + config.subscriptions_capacity, + SipHasherBuild::new({ + let mut seed = [0; 16]; + randomness.fill_bytes(&mut seed); + seed + }), + ), + active_subscriptions_by_server: BTreeMap::new(), + zombie_subscriptions: hashbrown::HashMap::with_capacity_and_hasher( + config.subscriptions_capacity, + SipHasherBuild::new({ + let mut seed = [0; 16]; + randomness.fill_bytes(&mut seed); + seed + }), + ), + zombie_subscriptions_pending: BTreeSet::new(), + zombie_subscriptions_by_resubscribe_id: BTreeMap::new(), + system_name: config.system_name, + system_version: config.system_version, + randomness, + } + } + + /// Shrinks the list of servers to the given capacity. Has no effect if the capacity is + /// smaller than the requested one. + pub fn shrink_servers_to(&mut self, _min_capacity: usize) { + self.servers.shrink_to_fit() // TODO: implement Slab::shrink_to? + } + + /// Shrinks the list of active requests to the given capacity. Has no effect if the capacity + /// is smaller than the requested one. + pub fn shrink_requests_to(&mut self, min_capacity: usize) { + self.responses_queue.shrink_to(min_capacity); + } + + /// Shrinks the list of active subscriptions to the given capacity. Has no effect if the + /// capacity is smaller than the requested one. + pub fn shrink_subscriptions_to(&mut self, min_capacity: usize) { + self.active_subscriptions.shrink_to(min_capacity); + self.zombie_subscriptions.shrink_to(min_capacity); + } + + /// Adds a new server to the collection. + pub fn add_server(&mut self, user_data: T) -> ServerId { + ServerId(self.servers.insert(Server { + is_blacklisted: false, + user_data, + })) + } + + /// Removes a server from the list of servers. + /// + /// All active subscriptions and requests are either stopped or redirected to a different + /// server. + /// + /// After this function returns, [`ServersMultiplexer::next_proxied_json_rpc_request`] should be + /// called with all idle servers, in order to pick up the requests that this server was + /// processing. + /// + /// # Panic + /// + /// Panics if the given [`ServerId`] is invalid. + /// + #[cold] + pub fn remove_server(&mut self, server_id: ServerId) -> T { + self.blacklist_server(server_id); + self.servers.remove(server_id.0).user_data + } + + #[cold] + fn blacklist_server(&mut self, server_id: ServerId) { + // Set `is_blacklisted` to `true`, and return immediately if it was already `true`. + if mem::replace(&mut self.servers[server_id.0].is_blacklisted, true) { + return; + } + + // Extract from `zombie_subscriptions_by_resubscribe_id` the re-subscription requests + // that were handled by that server. + let zombie_resubscriptions = { + let mut server_and_after = self + .zombie_subscriptions_by_resubscribe_id + .split_off(&(server_id, String::new())); + let mut after = server_and_after.split_off(&(ServerId(server_id.0 + 1), String::new())); + self.zombie_subscriptions_by_resubscribe_id + .append(&mut after); + server_and_after + }; + + // Re-queue the zombie resubscriptions. + for (_, zombie_subscription_id) in zombie_resubscriptions { + let _was_inserted = self + .zombie_subscriptions_pending + .insert(zombie_subscription_id); + debug_assert!(_was_inserted); + } + + // Extract from `active_subscriptions_by_server` the subscriptions that were handled + // by that server. + let mut subscriptions_to_cancel_or_reopen = { + let mut server_and_after = self + .active_subscriptions_by_server + .split_off(&(server_id, Arc::from(""))); + let mut after = server_and_after.split_off(&(ServerId(server_id.0 + 1), Arc::from(""))); + self.active_subscriptions_by_server.append(&mut after); + server_and_after + }; + + // Find in the subscriptions that were handled by that server the subscriptions that can + // be cancelled by sending a notification to the client. + // If the client happened to have a request in queue that concerns that subscription, + // this guarantees that the notification about the cancellation is sent to the client + // before the responses to this request. + // For example, if the client has queued a `chainHead_v1_header` request, it will + // receive the `stop` event of the `chainHead_v1_follow` subscription before + // receiving the error response to the `chainHead_v1_header` request. + // While this ordering is in no way a requirement, it is more polite to do so. + for (_, client_subscription_id) in &subscriptions_to_cancel_or_reopen { + let hashbrown::hash_map::EntryRef::Occupied(active_subscriptions_entry) = + self.active_subscriptions.entry_ref(client_subscription_id) + else { + unreachable!() + }; + + // TODO: not great to compare method names by string + match &*active_subscriptions_entry.get().1.method_name { + // Any active `chainHead_follow`, `transaction_submitAndWatch`, or + // `author_submitAndWatchExtrinsic` subscription is killed. + "author_submitAndWatchExtrinsic" => { + self.responses_queue.push_back( + methods::ServerToClient::author_extrinsicUpdate { + subscription: Cow::Borrowed(&*client_subscription_id), + result: methods::TransactionStatus::Dropped, + } + .to_json_request_object_parameters(None), + ); + active_subscriptions_entry.remove(); + } + "transactionWatch_v1_submitAndWatch" => { + self.responses_queue.push_back( + methods::ServerToClient::transactionWatch_v1_watchEvent { + subscription: Cow::Borrowed(&*client_subscription_id), + result: methods::TransactionWatchEvent::Dropped { + // Unfortunately, there is no way of knowing whether the server + // has broadcasted the transaction. Since `false` offers + // guarantees but `true` doesn't, we opt to always send + // back `true`. + // TODO: https://github.com/paritytech/json-rpc-interface-spec/issues/132 + broadcasted: true, + error: "Proxied server gone".into(), + }, + } + .to_json_request_object_parameters(None), + ); + active_subscriptions_entry.remove(); + } + "chainHead_v1_follow" => { + self.responses_queue.push_back( + methods::ServerToClient::chainHead_v1_followEvent { + subscription: Cow::Borrowed(&*client_subscription_id), + result: methods::FollowEvent::Stop {}, + } + .to_json_request_object_parameters(None), + ); + active_subscriptions_entry.remove(); + } + + // Other subscription types are handled below. + _ => {} + } + } + + // The server-specific requests that were queued for this server and the requests that + // were already sent to the server are processed the same way, as from the point of view + // of the JSON-RPC client there's no possible way to differentiate the two. + let requests_to_cancel = { + // Extract the list of requests that can only target that server. + let requests_queued = { + let mut server_and_after = self.queued_requests.split_off(&Some(server_id)); + let mut after = server_and_after.split_off(&Some(ServerId(server_id.0 + 1))); + self.queued_requests.append(&mut after); + server_and_after + }; + + // Extract from `requests_in_progress` the requests that were being processed by + // that server. + let requests_in_progress = { + let mut server_and_after = self + .requests_in_progress + .split_off(&(server_id, String::new())); + let mut after = + server_and_after.split_off(&(ServerId(server_id.0 + 1), String::new())); + self.requests_in_progress.append(&mut after); + server_and_after + }; + + let requests_queued = requests_queued + .into_iter() + .flat_map(|(_, requests)| requests.into_iter().map(|(id, rq)| (id, rq, false))); + let requests_dispatched = requests_in_progress + .into_iter() + .map(|((_, rq_id), rq)| (rq_id, rq, true)); + requests_dispatched.chain(requests_queued) + }; + + for (request_id_json, request_info, already_dispatched) in requests_to_cancel { + // Parse again the method. This is guaranteed to succeed, as otherwise the request + // wouldn't have been inserted into the local state in the first place. + let Ok(method) = methods::parse_jsonrpc_client_to_server_method_name_and_parameters( + &request_info.method_name, + request_info.method_params_json.as_deref(), + ) else { + unreachable!() + }; + + match &method { + // Unsubscription requests are immediately processed, and any request targetting a + // `chainHead_follow` subscription is answered immediately, as a `stop` event has + // been generated above. + methods::MethodCall::chain_unsubscribeAllHeads { subscription } + | methods::MethodCall::chain_unsubscribeFinalizedHeads { subscription } + | methods::MethodCall::chain_unsubscribeNewHeads { subscription } + | methods::MethodCall::state_unsubscribeRuntimeVersion { subscription } + | methods::MethodCall::state_unsubscribeStorage { subscription } + | methods::MethodCall::transactionWatch_v1_unwatch { subscription } + | methods::MethodCall::transaction_v1_stop { + operation_id: subscription, + } + | methods::MethodCall::chainHead_v1_body { + follow_subscription: subscription, + .. + } + | methods::MethodCall::chainHead_v1_call { + follow_subscription: subscription, + .. + } + | methods::MethodCall::chainHead_v1_header { + follow_subscription: subscription, + .. + } + | methods::MethodCall::chainHead_v1_stopOperation { + follow_subscription: subscription, + .. + } + | methods::MethodCall::chainHead_v1_storage { + follow_subscription: subscription, + .. + } + | methods::MethodCall::chainHead_v1_continue { + follow_subscription: subscription, + .. + } + | methods::MethodCall::chainHead_v1_unfollow { + follow_subscription: subscription, + } + | methods::MethodCall::chainHead_v1_unpin { + follow_subscription: subscription, + .. + } => { + let subscription_exists = subscriptions_to_cancel_or_reopen + .remove(&(server_id, Arc::from(&**subscription))) + .is_some(); + self.responses_queue.push_back(match method { + methods::MethodCall::chain_unsubscribeAllHeads { .. } => { + methods::Response::chain_unsubscribeAllHeads(subscription_exists) + .to_json_response(&request_id_json) + } + methods::MethodCall::chain_unsubscribeFinalizedHeads { .. } => { + methods::Response::chain_unsubscribeFinalizedHeads(subscription_exists) + .to_json_response(&request_id_json) + } + methods::MethodCall::chain_unsubscribeNewHeads { .. } => { + methods::Response::chain_unsubscribeNewHeads(subscription_exists) + .to_json_response(&request_id_json) + } + methods::MethodCall::state_unsubscribeRuntimeVersion { .. } => { + methods::Response::state_unsubscribeRuntimeVersion(subscription_exists) + .to_json_response(&request_id_json) + } + methods::MethodCall::state_unsubscribeStorage { .. } => { + methods::Response::state_unsubscribeStorage(subscription_exists) + .to_json_response(&request_id_json) + } + methods::MethodCall::transactionWatch_v1_unwatch { .. } => { + parse::build_error_response( + &request_id_json, + parse::ErrorResponse::InvalidParams, + None, + ) + } + methods::MethodCall::transaction_v1_stop { .. } => { + parse::build_error_response( + &request_id_json, + parse::ErrorResponse::InvalidParams, + None, + ) + } + methods::MethodCall::chainHead_v1_body { .. } => { + methods::Response::chainHead_v1_body( + methods::ChainHeadBodyCallReturn::LimitReached {}, + ) + .to_json_response(&request_id_json) + } + methods::MethodCall::chainHead_v1_call { .. } => { + methods::Response::chainHead_v1_call( + methods::ChainHeadBodyCallReturn::LimitReached {}, + ) + .to_json_response(&request_id_json) + } + methods::MethodCall::chainHead_v1_header { .. } => { + methods::Response::chainHead_v1_header(None) + .to_json_response(&request_id_json) + } + methods::MethodCall::chainHead_v1_stopOperation { .. } => { + methods::Response::chainHead_v1_stopOperation(()) + .to_json_response(&request_id_json) + } + methods::MethodCall::chainHead_v1_storage { .. } => { + methods::Response::chainHead_v1_storage( + methods::ChainHeadStorageReturn::LimitReached {}, + ) + .to_json_response(&request_id_json) + } + methods::MethodCall::chainHead_v1_continue { .. } => { + methods::Response::chainHead_v1_continue(()) + .to_json_response(&request_id_json) + } + methods::MethodCall::chainHead_v1_unfollow { .. } => { + methods::Response::chainHead_v1_unfollow(()) + .to_json_response(&request_id_json) + } + methods::MethodCall::chainHead_v1_unpin { .. } => { + methods::Response::chainHead_v1_unpin(()) + .to_json_response(&request_id_json) + } + _ => unreachable!(), + }); + } + + // Any other request is added back to the head of the queue. + _ => { + debug_assert!(already_dispatched); + self.queued_requests + .entry(None) + .or_insert_with(VecDeque::new) + .push_front((request_id_json, request_info)); + } + } + } + + // Process a second time the subscriptions to cancel. The list of subscriptions has been + // adjusted by the unsubscribe requests processed above, which is why this is done + // separately at the end. + // This time, we reopen legacy JSON-RPC API subscriptions by inserting them into a + // "zombie subscriptions" list. + for (_, client_subscription_id) in subscriptions_to_cancel_or_reopen { + let Some((_, active_subscription)) = + self.active_subscriptions.remove(&client_subscription_id) + else { + // We might have already removed the subscription earlier. + continue; + }; + + self.zombie_subscriptions.insert( + client_subscription_id.clone(), + ZombieSubscription { + method_name: active_subscription.method_name, + method_params_json: active_subscription.method_params_json, + resubscribe_request_id: None, + unsubscribe_request_id: None, + }, + ); + + self.zombie_subscriptions_pending + .insert(client_subscription_id); + } + } + + /// Adds a request to the queue of requests waiting to be picked up by a server. + pub fn insert_json_rpc_request(&mut self, request: String) -> InsertRequest { + // Parse the request, making sure that it is valid JSON-RPC. + let request = match parse::parse_request(&request) { + Ok(rq) => rq, + Err(_) => { + // Failed to parse the JSON-RPC request. + self.responses_queue + .push_back(parse::build_parse_error_response()); + return InsertRequest::ImmediateAnswer; + } + }; + + // Extract the request ID. + let Some(request_id_json) = request.id_json else { + // The call is a notification. No notification is supported. + // According to the JSON-RPC specification, the server must not send any response + // to notifications, even in case of an error. + return InsertRequest::Discarded; + }; + + // Now parse the method name and parameters. + let Ok(method) = methods::parse_jsonrpc_client_to_server_method_name_and_parameters( + request.method, + request.params_json, + ) else { + // JSON-RPC function not recognized. + + // Requests with an unknown method must not be blindly sent to a server, as it is + // not possible for the reverse proxy to guarantee that the logic of the request + // is respected. + // For example, if the request is a subscription request, the reverse proxy + // wouldn't be capable of understanding which client to redirect the notifications + // to. + self.responses_queue.push_back(parse::build_error_response( + request_id_json, + parse::ErrorResponse::MethodNotFound, + None, + )); + return InsertRequest::ImmediateAnswer; + }; + + // Try to answer the request directly. If not possible, determine which server the request + // must target, and potentially adjust its parameters. + let (rewritten_parameters, assigned_server) = match &method { + // Answer the request directly if possible. + methods::MethodCall::system_name {} => { + self.responses_queue.push_back( + methods::Response::system_name(Cow::Borrowed(&*self.system_name)) + .to_json_response(request_id_json), + ); + return InsertRequest::ImmediateAnswer; + } + methods::MethodCall::system_version {} => { + self.responses_queue.push_back( + methods::Response::system_version(Cow::Borrowed(&*self.system_version)) + .to_json_response(request_id_json), + ); + return InsertRequest::ImmediateAnswer; + } + methods::MethodCall::sudo_unstable_version {} => { + self.responses_queue.push_back( + methods::Response::sudo_unstable_version(Cow::Owned(format!( + "{} {}", + self.system_name, self.system_version + ))) + .to_json_response(request_id_json), + ); + return InsertRequest::ImmediateAnswer; + } + methods::MethodCall::sudo_unstable_p2pDiscover { .. } => { + self.responses_queue.push_back( + methods::Response::sudo_unstable_p2pDiscover(()) + .to_json_response(request_id_json), + ); + return InsertRequest::ImmediateAnswer; + } + + // Some requests are forbidden. + methods::MethodCall::sudo_network_unstable_watch {} + | methods::MethodCall::sudo_network_unstable_unwatch { .. } + | methods::MethodCall::author_removeExtrinsic { .. } + | methods::MethodCall::author_rotateKeys { .. } + | methods::MethodCall::system_addReservedPeer { .. } + | methods::MethodCall::system_localListenAddresses { .. } + | methods::MethodCall::system_localPeerId { .. } + | methods::MethodCall::system_networkState { .. } + | methods::MethodCall::system_nodeRoles { .. } + | methods::MethodCall::system_peers { .. } + | methods::MethodCall::system_removeReservedPeer { .. } => { + self.responses_queue.push_back(parse::build_error_response( + request_id_json, + parse::ErrorResponse::MethodNotFound, + None, + )); + return InsertRequest::ImmediateAnswer; + } + + // Unsubscription functions or functions that target a specific subscription. + methods::MethodCall::chain_unsubscribeAllHeads { subscription } + | methods::MethodCall::chain_unsubscribeFinalizedHeads { subscription } + | methods::MethodCall::chain_unsubscribeNewHeads { subscription } + | methods::MethodCall::state_unsubscribeRuntimeVersion { subscription } + | methods::MethodCall::state_unsubscribeStorage { subscription } + | methods::MethodCall::author_unwatchExtrinsic { subscription } + | methods::MethodCall::transactionWatch_v1_unwatch { subscription } + | methods::MethodCall::transaction_v1_stop { + operation_id: subscription, + } + | methods::MethodCall::chainHead_v1_body { + follow_subscription: subscription, + .. + } + | methods::MethodCall::chainHead_v1_call { + follow_subscription: subscription, + .. + } + | methods::MethodCall::chainHead_v1_header { + follow_subscription: subscription, + .. + } + | methods::MethodCall::chainHead_v1_stopOperation { + follow_subscription: subscription, + .. + } + | methods::MethodCall::chainHead_v1_storage { + follow_subscription: subscription, + .. + } + | methods::MethodCall::chainHead_v1_continue { + follow_subscription: subscription, + .. + } + | methods::MethodCall::chainHead_v1_unfollow { + follow_subscription: subscription, + } + | methods::MethodCall::chainHead_v1_unpin { + follow_subscription: subscription, + .. + } => { + // TODO: must check whether the subscription type matches the expected one + if let Some(&(server_id, ref subscription_info)) = + self.active_subscriptions.get(&**subscription) + { + // The subscription exists and is active. + // Pass the unsubscription request to the server. + // We have to adjust the parameters of the request so that the subscription + // ID becomes the one from the point of view of the server. + let parameters_rewrite = match method { + methods::MethodCall::chain_unsubscribeAllHeads { .. } => { + methods::MethodCall::chain_unsubscribeAllHeads { + subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + } + .params_to_json_object() + } + methods::MethodCall::chain_unsubscribeFinalizedHeads { .. } => { + methods::MethodCall::chain_unsubscribeFinalizedHeads { + subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + } + .params_to_json_object() + } + methods::MethodCall::chain_unsubscribeNewHeads { .. } => { + methods::MethodCall::chain_unsubscribeNewHeads { + subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + } + .params_to_json_object() + } + methods::MethodCall::state_unsubscribeRuntimeVersion { .. } => { + methods::MethodCall::state_unsubscribeRuntimeVersion { + subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + } + .params_to_json_object() + } + methods::MethodCall::state_unsubscribeStorage { .. } => { + methods::MethodCall::state_unsubscribeStorage { + subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + } + .params_to_json_object() + } + methods::MethodCall::author_unwatchExtrinsic { .. } => { + methods::MethodCall::author_unwatchExtrinsic { + subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + } + .params_to_json_object() + } + methods::MethodCall::transactionWatch_v1_unwatch { .. } => { + methods::MethodCall::transactionWatch_v1_unwatch { + subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + } + .params_to_json_object() + } + methods::MethodCall::transaction_v1_stop { .. } => { + methods::MethodCall::transaction_v1_stop { + operation_id: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + } + .params_to_json_object() + } + methods::MethodCall::chainHead_v1_body { hash, .. } => { + methods::MethodCall::chainHead_v1_body { + follow_subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + hash, + } + .params_to_json_object() + } + methods::MethodCall::chainHead_v1_call { + hash, + function, + call_parameters, + .. + } => methods::MethodCall::chainHead_v1_call { + follow_subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + hash, + function, + call_parameters, + } + .params_to_json_object(), + methods::MethodCall::chainHead_v1_header { hash, .. } => { + methods::MethodCall::chainHead_v1_header { + follow_subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + hash, + } + .params_to_json_object() + } + methods::MethodCall::chainHead_v1_stopOperation { + operation_id, .. + } => methods::MethodCall::chainHead_v1_stopOperation { + follow_subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + operation_id, + } + .params_to_json_object(), + methods::MethodCall::chainHead_v1_storage { + hash, + items, + child_trie, + .. + } => methods::MethodCall::chainHead_v1_storage { + follow_subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + hash, + items, + child_trie, + } + .params_to_json_object(), + methods::MethodCall::chainHead_v1_continue { operation_id, .. } => { + methods::MethodCall::chainHead_v1_continue { + follow_subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + operation_id, + } + .params_to_json_object() + } + methods::MethodCall::chainHead_v1_unfollow { .. } => { + methods::MethodCall::chainHead_v1_unfollow { + follow_subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + } + .params_to_json_object() + } + methods::MethodCall::chainHead_v1_unpin { hash_or_hashes, .. } => { + methods::MethodCall::chainHead_v1_unpin { + follow_subscription: Cow::Borrowed( + &subscription_info.server_subscription_id, + ), + hash_or_hashes, + } + .params_to_json_object() + } + _ => unreachable!(), + }; + + (Some(parameters_rewrite), Some(server_id)) + } else { + // The subscription isn't active on any server. It might exist in a special + // state. + let subscription_exists = + if let hashbrown::hash_map::EntryRef::Occupied(mut zombie_subscription) = + self.zombie_subscriptions.entry_ref(&**subscription) + { + // The subscription is a so-called "zombie subscription". It exists + // from the point of view of the client, but isn't active on any + // server. + if zombie_subscription.get().resubscribe_request_id.is_some() { + // We have sent a re-subscription request to one of the servers. + if zombie_subscription + .get_mut() + .unsubscribe_request_id + .is_none() + { + // Update the state of the zombie subscription so that we + // immediately unsubscribe as soon as the re-subscription + // happens. + zombie_subscription.get_mut().unsubscribe_request_id = + Some(request_id_json.to_owned()); + return InsertRequest::LocalStateUpdate; + } else { + // The client has already unsubscribed from this subscription, + // but we haven't sent an answer yet, and is now trying to + // unsubscribe a second time. Consider that the subscription + // doesn't exist for this second time. + // TODO: the second unsubscribe response will come before the first unsubscribe, is that a problem? + false + } + } else { + // We haven't sent any re-subscription request to any of the + // servers yet. The local state can be entirely cleaned up. + zombie_subscription.remove(); + let _was_in = + self.zombie_subscriptions_pending.remove(&**subscription); + debug_assert!(_was_in); + true + } + } else { + // Subscription isn't a "zombie subscription". It doesn't exist, or + // doesn't exist anymore. + false + }; + + // Immediately send back a response to the client. + self.responses_queue.push_back(match method { + methods::MethodCall::chain_unsubscribeAllHeads { .. } => { + methods::Response::chain_unsubscribeAllHeads(subscription_exists) + .to_json_response(request_id_json) + } + methods::MethodCall::chain_unsubscribeFinalizedHeads { .. } => { + methods::Response::chain_unsubscribeFinalizedHeads(subscription_exists) + .to_json_response(request_id_json) + } + methods::MethodCall::chain_unsubscribeNewHeads { .. } => { + methods::Response::chain_unsubscribeNewHeads(subscription_exists) + .to_json_response(request_id_json) + } + methods::MethodCall::state_unsubscribeRuntimeVersion { .. } => { + methods::Response::state_unsubscribeRuntimeVersion(subscription_exists) + .to_json_response(request_id_json) + } + methods::MethodCall::state_unsubscribeStorage { .. } => { + methods::Response::state_unsubscribeStorage(subscription_exists) + .to_json_response(request_id_json) + } + methods::MethodCall::author_unwatchExtrinsic { .. } => { + parse::build_error_response( + request_id_json, + parse::ErrorResponse::InvalidParams, + None, + ) + } + methods::MethodCall::transactionWatch_v1_unwatch { .. } => { + parse::build_error_response( + request_id_json, + parse::ErrorResponse::InvalidParams, + None, + ) + } + methods::MethodCall::transaction_v1_stop { .. } => { + parse::build_error_response( + request_id_json, + parse::ErrorResponse::InvalidParams, + None, + ) + } + methods::MethodCall::chainHead_v1_body { .. } => { + methods::Response::chainHead_v1_body( + methods::ChainHeadBodyCallReturn::LimitReached {}, + ) + .to_json_response(request_id_json) + } + methods::MethodCall::chainHead_v1_call { .. } => { + methods::Response::chainHead_v1_call( + methods::ChainHeadBodyCallReturn::LimitReached {}, + ) + .to_json_response(request_id_json) + } + methods::MethodCall::chainHead_v1_header { .. } => { + methods::Response::chainHead_v1_header(None) + .to_json_response(request_id_json) + } + methods::MethodCall::chainHead_v1_stopOperation { .. } => { + methods::Response::chainHead_v1_stopOperation(()) + .to_json_response(request_id_json) + } + methods::MethodCall::chainHead_v1_storage { .. } => { + methods::Response::chainHead_v1_storage( + methods::ChainHeadStorageReturn::LimitReached {}, + ) + .to_json_response(request_id_json) + } + methods::MethodCall::chainHead_v1_continue { .. } => { + methods::Response::chainHead_v1_continue(()) + .to_json_response(request_id_json) + } + methods::MethodCall::chainHead_v1_unfollow { .. } => { + methods::Response::chainHead_v1_unfollow(()) + .to_json_response(request_id_json) + } + methods::MethodCall::chainHead_v1_unpin { .. } => { + methods::Response::chainHead_v1_unpin(()) + .to_json_response(request_id_json) + } + _ => unreachable!(), + }); + return InsertRequest::ImmediateAnswer; + } + } + + // Other JSON-RPC API functions. + // Functions are listed individually so that if a function is added, this + // code needs to be tweaked and added to the right category. + methods::MethodCall::account_nextIndex { .. } + | methods::MethodCall::author_hasKey { .. } + | methods::MethodCall::author_hasSessionKeys { .. } + | methods::MethodCall::author_insertKey { .. } + | methods::MethodCall::author_pendingExtrinsics { .. } + | methods::MethodCall::author_submitExtrinsic { .. } + | methods::MethodCall::babe_epochAuthorship { .. } + | methods::MethodCall::chain_getBlock { .. } + | methods::MethodCall::chain_getBlockHash { .. } + | methods::MethodCall::chain_getFinalizedHead { .. } + | methods::MethodCall::chain_getHeader { .. } + | methods::MethodCall::childstate_getKeys { .. } + | methods::MethodCall::childstate_getStorage { .. } + | methods::MethodCall::childstate_getStorageHash { .. } + | methods::MethodCall::childstate_getStorageSize { .. } + | methods::MethodCall::grandpa_roundState { .. } + | methods::MethodCall::offchain_localStorageGet { .. } + | methods::MethodCall::offchain_localStorageSet { .. } + | methods::MethodCall::payment_queryInfo { .. } + | methods::MethodCall::rpc_methods { .. } + | methods::MethodCall::state_call { .. } + | methods::MethodCall::state_getKeys { .. } + | methods::MethodCall::state_getKeysPaged { .. } + | methods::MethodCall::state_getMetadata { .. } + | methods::MethodCall::state_getPairs { .. } + | methods::MethodCall::state_getReadProof { .. } + | methods::MethodCall::state_getRuntimeVersion { .. } + | methods::MethodCall::state_getStorage { .. } + | methods::MethodCall::state_getStorageHash { .. } + | methods::MethodCall::state_getStorageSize { .. } + | methods::MethodCall::state_queryStorage { .. } + | methods::MethodCall::state_queryStorageAt { .. } + | methods::MethodCall::system_accountNextIndex { .. } + | methods::MethodCall::system_chain { .. } + | methods::MethodCall::system_chainType { .. } + | methods::MethodCall::system_dryRun { .. } + | methods::MethodCall::system_health { .. } + | methods::MethodCall::system_properties { .. } + | methods::MethodCall::state_subscribeRuntimeVersion {} + | methods::MethodCall::state_subscribeStorage { .. } + | methods::MethodCall::chain_subscribeAllHeads {} + | methods::MethodCall::chain_subscribeFinalizedHeads {} + | methods::MethodCall::chain_subscribeNewHeads {} + | methods::MethodCall::author_submitAndWatchExtrinsic { .. } + | methods::MethodCall::chainHead_v1_follow { .. } + | methods::MethodCall::transactionWatch_v1_submitAndWatch { .. } + | methods::MethodCall::transaction_v1_broadcast { .. } + | methods::MethodCall::chainSpec_v1_chainName {} + | methods::MethodCall::chainSpec_v1_genesisHash {} + | methods::MethodCall::chainSpec_v1_properties {} + | methods::MethodCall::chainHead_unstable_finalizedDatabase { .. } => { + (request.params_json.map(|p| p.to_owned()), None) + } + }; + + // Everything went well. Insert the request in the queue. + self.queued_requests + .entry(assigned_server) + .or_insert(VecDeque::new()) + .push_back(( + request_id_json.to_owned(), + Request { + method_name: request.method.to_owned(), + method_params_json: rewritten_parameters, + previous_failed_attempts: 0, + }, + )); + if let Some(assigned_server) = assigned_server { + InsertRequest::ServerWakeUp(assigned_server) + } else { + InsertRequest::AnyServerWakeUp + } + } + + /// Returns the next JSON-RPC response or notification to send to the client. + /// + /// Returns `None` if none is available. + /// + /// The return type of [`ServersMultiplexer::insert_proxied_json_rpc_response`] indicates + /// if a JSON-RPC response or notification has become available. + pub fn next_json_rpc_response(&mut self) -> Option { + self.responses_queue.pop_front() + } + + /// Pick a JSON-RPC request waiting to be processed. + /// + /// Returns `None` if no JSON-RPC request is waiting to be processed, and you should try + /// calling this function again after [`ServersMultiplexer::insert_json_rpc_request`]. + /// + /// `None` is always returned if the server is blacklisted. + /// + /// Note that the [`ServersMultiplexer`] state machine doesn't enforce any limit to the number of + /// JSON-RPC requests that a server processes simultaneously. A JSON-RPC server is expected to + /// back-pressure its socket once it gets too busy, in which case + /// [`ServersMultiplexer::next_proxied_json_rpc_request`] should no longer be called until the + /// server is ready to accept more data. + /// This ensures that for example a JSON-RPC server that is twice as powerful compared to + /// another one should get approximately twice the number of requests. + /// + /// # Panic + /// + /// Panics if the given [`ServerId`] is invalid. + /// + pub fn next_proxied_json_rpc_request(&mut self, server_id: ServerId) -> Option { + let server = &mut self.servers[server_id.0]; + if server.is_blacklisted { + return None; + } + + // It might be that the client had an active subscription against a server and that this + // server got blacklisted or removed in the past. + // If that happens, send a dummy subscription request in order to re-open that + // subscription. + if let Some(zombie_subscription_id) = self.zombie_subscriptions_pending.pop_first() { + let Some(zombie_subscription) = self.zombie_subscriptions.get(&zombie_subscription_id) + else { + unreachable!() + }; + + let subscribe_request_id_json = serde_json::to_string(&{ + let mut subscription_id = [0u8; 32]; + self.randomness.fill_bytes(&mut subscription_id); + bs58::encode(subscription_id).into_string() + }) + .unwrap_or_else(|_| unreachable!()); + + let resub_request_json = parse::build_request(&parse::Request { + id_json: Some(&subscribe_request_id_json), + method: &zombie_subscription.method_name, + params_json: zombie_subscription.method_params_json.as_deref(), + }); + + let _prev_value = self.zombie_subscriptions_by_resubscribe_id.insert( + (server_id, subscribe_request_id_json), + zombie_subscription_id, + ); + debug_assert!(_prev_value.is_none()); + + return Some(resub_request_json); + } + + // In any other case, grab a request from the queue. + // There are two types of requests: requests that aren't attributed to any server, and + // requests that are attributed to a specific server. + // In order to guarantee some fairness, we pick from either list randomly. + let (queued_request_id, queued_request) = { + // Choose which queue to pick from. + let pick_from_specific_queue = { + let num_non_specific = self.queued_requests.get(&None).map_or(0, |l| l.len()); + let num_specific = (self + .queued_requests + .get(&Some(server_id)) + .map_or(0, |l| l.len()) + + (self.servers.len() - 1)) // `servers.len()` is necessarily non-zero, as a `ServerId` is provided as input. + / self.servers.len(); + if num_non_specific == 0 && num_specific == 0 { + // No request available. + // The code below would panic if we continued. + return None; + } + + let rand = rand::distributions::Distribution::sample( + &rand::distributions::Uniform::new(0, num_non_specific + num_specific), + &mut self.randomness, + ); + + rand >= num_non_specific + }; + + // Extract the request from the queue. + let btree_map::Entry::Occupied(mut entry) = + self.queued_requests.entry(if pick_from_specific_queue { + Some(server_id) + } else { + None + }) + else { + unreachable!() + }; + let request = entry + .get_mut() + .pop_front() + .unwrap_or_else(|| unreachable!()); + if entry.get().is_empty() { + entry.remove(); + } + request + }; + + // Turn the request parameters into a proper request. + let serialized_request = parse::build_request(&parse::Request { + id_json: Some(&queued_request_id), + method: &queued_request.method_name, + params_json: queued_request.method_params_json.as_deref(), + }); + + // Update local state to track that the server is processing this request. + let _previous_value = self + .requests_in_progress + .insert((server_id, queued_request_id), queued_request); + debug_assert!(_previous_value.is_none()); + + // Success. + Some(serialized_request) + } + + /// Inserts a response or notification sent by a server. + /// + /// Note that there exists no back-pressure system here. Responses and notifications sent by + /// servers are always accepted and buffered in order to be picked up later + /// by [`ServersMultiplexer::next_json_rpc_response`]. + /// + /// # Panic + /// + /// Panics if the given [`ServerId`] is invalid. + /// + pub fn insert_proxied_json_rpc_response( + &mut self, + server_id: ServerId, + mut response: String, + ) -> InsertProxiedJsonRpcResponse { + match parse::parse_response(&response) { + Err(_) => { + // If the message couldn't be parsed as a response, attempt to parse it as a + // notification. + let Ok(mut notification) = methods::parse_notification(&response) else { + // Failed to parse the message from the JSON-RPC server. + self.blacklist_server(server_id); + return InsertProxiedJsonRpcResponse::Blacklisted( + BlacklistReason::ParseFailure, + ); + }; + + // This is a subscription notification. + // TODO: overhead of into_owned + let btree_map::Entry::Occupied(subscription_entry) = self + .active_subscriptions_by_server + .entry((server_id, Arc::from(&**notification.subscription()))) + else { + // The subscription ID isn't recognized. This indicates something very + // wrong with the server. We handle this by blacklisting the server. + self.blacklist_server(server_id); + return InsertProxiedJsonRpcResponse::Blacklisted( + BlacklistReason::InvalidSubscriptionId, + ); + }; + + // Rewrite the subscription ID in the notification in order to match what + // the client expects. + notification.set_subscription(Cow::Borrowed(&subscription_entry.get())); + let rewritten_notification = notification.to_json_request_object_parameters(None); + + // Remove the subscription if the notification indicates that the + // subscription is finished. + if matches!( + notification, + methods::ServerToClient::author_extrinsicUpdate { + result: methods::TransactionStatus::Dropped, + .. + } | methods::ServerToClient::transactionWatch_v1_watchEvent { + result: methods::TransactionWatchEvent::Error { .. } + | methods::TransactionWatchEvent::Finalized { .. } + | methods::TransactionWatchEvent::Invalid { .. } + | methods::TransactionWatchEvent::Dropped { .. }, + .. + } | methods::ServerToClient::chainHead_v1_followEvent { + result: methods::FollowEvent::Stop {}, + .. + } + ) { + let _was_in = self.active_subscriptions.remove(subscription_entry.get()); + debug_assert!(_was_in.is_some()); + subscription_entry.remove(); + } + + // Success. + self.responses_queue.push_back(rewritten_notification); + InsertProxiedJsonRpcResponse::Queued + } + + Ok(parse_result) => { + // Grab the ID of the request that is being answered. + let request_id_json = match parse_result { + parse::Response::Success { id_json, .. } => id_json, + parse::Response::Error { id_json, .. } => id_json, + parse::Response::ParseError { .. } => { + // JSON-RPC server indicates that it has failed to parse a JSON-RPC + // request as a valid request. This is never supposed to happen and + // indicates that something is very wrong with the server. + // The server is blacklisted. + self.blacklist_server(server_id); + return InsertProxiedJsonRpcResponse::Blacklisted( + BlacklistReason::ParseErrorResponse, + ); + } + }; + + // TODO: detect internal server errors and blacklist the server + + // Find the answered request in the local state. + // We don't immediately remove the request, as it might have to stay there. + let mut requests_in_progress_entry = self + .requests_in_progress + .entry((server_id, request_id_json.to_owned())); // TODO: to_owned overhead + let mut zombie_subscriptions_entry = self + .zombie_subscriptions_by_resubscribe_id + .entry((server_id, request_id_json.to_owned())); // TODO: to_owned overhead + + // Extract information about the request. + let ( + method_name, + method_params_json, + unsubscribe_request_id, + previous_failed_attempts, + ) = match ( + &mut requests_in_progress_entry, + &mut zombie_subscriptions_entry, + ) { + (btree_map::Entry::Occupied(rq), _zombie) => { + debug_assert!(matches!(_zombie, btree_map::Entry::Vacant(_))); + let rq = rq.get_mut(); + ( + &rq.method_name, + &rq.method_params_json, + None, + Some(&mut rq.previous_failed_attempts), + ) + } + (btree_map::Entry::Vacant(_), btree_map::Entry::Occupied(zombie)) => { + let Some(subscription) = self.zombie_subscriptions.get_mut(zombie.get()) + else { + unreachable!() + }; + ( + &subscription.method_name, + &subscription.method_params_json, + Some(&mut subscription.unsubscribe_request_id), + None, + ) + } + (btree_map::Entry::Vacant(_), btree_map::Entry::Vacant(_)) => { + // Server has answered a non-existing request. Blacklist it. + self.blacklist_server(server_id); + return InsertProxiedJsonRpcResponse::Blacklisted( + BlacklistReason::InvalidRequestId, + ); + } + }; + + // Parse the request's method name and parameters again. + // This is guaranteed to always succeed, as otherwise the request wouldn't have + // been inserted into the local state. + let Ok(request_method) = + methods::parse_jsonrpc_client_to_server_method_name_and_parameters( + &method_name, + method_params_json.as_deref(), + ) + else { + unreachable!() + }; + + match ( + &request_method, + previous_failed_attempts.map_or(false, |n| *n >= 2), + ) { + // Some functions return `null` if the server has reached its limit, in which + // case we silently discard the response and try a different server instead. + // TODO: not finished + (methods::MethodCall::chainHead_v1_follow { .. }, false) => {} + + _ => {} + } + + // TODO: what if error and zombie request + match (&request_method, parse_result) { + // If the function is a subscription, we update our local state. + ( + methods::MethodCall::chainHead_v1_follow { .. } + | methods::MethodCall::chain_subscribeAllHeads {} + | methods::MethodCall::chain_subscribeFinalizedHeads {} + | methods::MethodCall::chain_subscribeNewHeads {} + | methods::MethodCall::state_subscribeRuntimeVersion {} + | methods::MethodCall::state_subscribeStorage { .. } + | methods::MethodCall::author_submitAndWatchExtrinsic { .. } + | methods::MethodCall::transactionWatch_v1_submitAndWatch { .. } + | methods::MethodCall::transaction_v1_broadcast { .. }, + parse::Response::Success { result_json, .. }, + ) => { + let subscription_id = match methods::parse_jsonrpc_response( + request_method.name(), + result_json, + ) { + Ok( + methods::Response::chainHead_v1_follow(subscription_id) + | methods::Response::chain_subscribeAllHeads(subscription_id) + | methods::Response::chain_subscribeFinalizedHeads(subscription_id) + | methods::Response::chain_subscribeNewHeads(subscription_id) + | methods::Response::state_subscribeRuntimeVersion(subscription_id) + | methods::Response::state_subscribeStorage(subscription_id) + | methods::Response::author_submitAndWatchExtrinsic(subscription_id) + | methods::Response::transactionWatch_v1_submitAndWatch( + subscription_id, + ) + | methods::Response::transaction_v1_broadcast(subscription_id), + ) => subscription_id, + Ok(_) => unreachable!(), + Err(_) => { + // TODO: ?! + todo!() + } + }; + + // Turn the subscription into an `Arc`. + let subscription_id: Arc = Arc::from(subscription_id); + + // Because multiple servers might assign the same subscription ID, we + // assign a new separate subscription ID to send back to the client. + let rellocated_subscription_id = { + let mut subscription_id = [0u8; 32]; + self.randomness.fill_bytes(&mut subscription_id); + Arc::::from(bs58::encode(subscription_id).into_string()) + }; + + // Update the list of active subscriptions stored in `self`. + match self + .active_subscriptions_by_server + .entry((server_id, subscription_id.clone())) + { + btree_map::Entry::Vacant(entry) => { + entry.insert(rellocated_subscription_id.clone()); + } + btree_map::Entry::Occupied(_) => { + // The server has assigned a subscription ID that it has + // already assigned in the past. This indicates something + // very wrong with the server. + self.blacklist_server(server_id); + return InsertProxiedJsonRpcResponse::Blacklisted( + BlacklistReason::DuplicateSubscriptionId, + ); + } + } + let _prev_value = self.active_subscriptions.insert( + rellocated_subscription_id.clone(), + ( + server_id, + ActiveSubscription { + method_name: method_name.clone(), + method_params_json: method_params_json.clone(), + server_subscription_id: subscription_id.clone(), + }, + ), + ); + debug_assert!(_prev_value.is_none()); + + // If this subscription was a zombie subscription (i.e. a subscription + // that the client thinks is active but wasn't actually active on any + // server), it is possible that the client sent a request to unsubscribe + // while the server was still responding to the subscription request. + // When that happens, the client's unsubscription request is silently + // ignored, and we now re-queue it. + // TODO: notify through return type that there's a request for the server to pick up? + if let Some(unsubscribe_request_id) = + unsubscribe_request_id.and_then(|rq_id| rq_id.take()) + { + let unsub_request = match request_method { + methods::MethodCall::chainHead_v1_follow { .. } => { + methods::MethodCall::chainHead_v1_unfollow { + follow_subscription: Cow::Borrowed(&*subscription_id), + } + } + methods::MethodCall::chain_subscribeAllHeads {} => { + methods::MethodCall::chain_unsubscribeAllHeads { + subscription: Cow::Borrowed(&*subscription_id), + } + } + methods::MethodCall::chain_subscribeFinalizedHeads {} => { + methods::MethodCall::chain_unsubscribeFinalizedHeads { + subscription: Cow::Borrowed(&*subscription_id), + } + } + methods::MethodCall::chain_subscribeNewHeads {} => { + methods::MethodCall::chain_unsubscribeNewHeads { + subscription: Cow::Borrowed(&*subscription_id), + } + } + methods::MethodCall::state_subscribeRuntimeVersion {} => { + methods::MethodCall::state_unsubscribeRuntimeVersion { + subscription: Cow::Borrowed(&*subscription_id), + } + } + methods::MethodCall::state_subscribeStorage { .. } => { + methods::MethodCall::state_unsubscribeStorage { + subscription: Cow::Borrowed(&*subscription_id), + } + } + methods::MethodCall::author_submitAndWatchExtrinsic { .. } => { + methods::MethodCall::author_unwatchExtrinsic { + subscription: Cow::Borrowed(&*subscription_id), + } + } + methods::MethodCall::transactionWatch_v1_submitAndWatch { + .. + } => methods::MethodCall::transactionWatch_v1_unwatch { + subscription: Cow::Borrowed(&*subscription_id), + }, + methods::MethodCall::transaction_v1_broadcast { .. } => { + methods::MethodCall::transaction_v1_stop { + operation_id: Cow::Borrowed(&*subscription_id), + } + } + _ => unreachable!(), + }; + + self.queued_requests + .entry(Some(server_id)) + .or_insert(VecDeque::new()) + .push_back(( + unsubscribe_request_id, + Request { + method_name: unsub_request.name().to_owned(), + method_params_json: Some( + unsub_request.params_to_json_object(), + ), + previous_failed_attempts: 0, + }, + )); + } + + // The response to the client needs to be adjusted for the fact that + // we modify the subscription ID. + response = match request_method { + methods::MethodCall::chainHead_v1_follow { .. } => { + methods::Response::chainHead_v1_follow(Cow::Borrowed( + &rellocated_subscription_id, + )) + } + methods::MethodCall::chain_subscribeAllHeads {} => { + methods::Response::chain_subscribeAllHeads(Cow::Borrowed( + &rellocated_subscription_id, + )) + } + methods::MethodCall::chain_subscribeFinalizedHeads {} => { + methods::Response::chain_subscribeFinalizedHeads(Cow::Borrowed( + &rellocated_subscription_id, + )) + } + methods::MethodCall::chain_subscribeNewHeads {} => { + methods::Response::chain_subscribeNewHeads(Cow::Borrowed( + &rellocated_subscription_id, + )) + } + methods::MethodCall::state_subscribeRuntimeVersion {} => { + methods::Response::state_subscribeRuntimeVersion(Cow::Borrowed( + &rellocated_subscription_id, + )) + } + methods::MethodCall::state_subscribeStorage { .. } => { + methods::Response::state_subscribeStorage(Cow::Borrowed( + &rellocated_subscription_id, + )) + } + methods::MethodCall::author_submitAndWatchExtrinsic { .. } => { + methods::Response::author_submitAndWatchExtrinsic(Cow::Borrowed( + &rellocated_subscription_id, + )) + } + methods::MethodCall::transactionWatch_v1_submitAndWatch { .. } => { + methods::Response::transactionWatch_v1_submitAndWatch( + Cow::Borrowed(&rellocated_subscription_id), + ) + } + methods::MethodCall::transaction_v1_broadcast { .. } => { + methods::Response::transaction_v1_broadcast(Cow::Borrowed( + &rellocated_subscription_id, + )) + } + _ => unreachable!(), + } + .to_json_response(request_id_json); + } + + // If the function is an unsubscription, we update our local state. + // It is possible that the server has sent back an error or indicating that + // the subscription was invalid. Because there is not much that can be done + // to handle that situation properly, we don't actually put much effort into + // detecting that case. In the best case scenario, the server has for some + // reason lost this subscription in the past and everything is fine. In the + // worst case scenario, the server will continue sending notifications which + // the multiplexer will ignore when they arrive. + ( + methods::MethodCall::chain_unsubscribeAllHeads { subscription } + | methods::MethodCall::chain_unsubscribeFinalizedHeads { subscription } + | methods::MethodCall::chain_unsubscribeNewHeads { subscription } + | methods::MethodCall::state_unsubscribeRuntimeVersion { subscription } + | methods::MethodCall::state_unsubscribeStorage { subscription } + | methods::MethodCall::author_unwatchExtrinsic { subscription } + | methods::MethodCall::transactionWatch_v1_unwatch { subscription } + | methods::MethodCall::transaction_v1_stop { + operation_id: subscription, + } + | methods::MethodCall::chainHead_v1_unfollow { + follow_subscription: subscription, + }, + _, + ) => { + // The request that has been extracted has had its parameters adjusted + // so that the subscription ID is the server-side ID. + if let Some(client_subscription_id) = self + .active_subscriptions_by_server + .remove(&(server_id, Arc::from(&**subscription))) + // TODO: overhead ^ + { + let _was_in = self.active_subscriptions.remove(&client_subscription_id); + debug_assert!(_was_in.is_some()); + } + + // The response to the client is adjusted to always be a successful + // unsubscription. + response = match request_method { + methods::MethodCall::chain_unsubscribeAllHeads { .. } => { + methods::Response::chain_unsubscribeAllHeads(true) + } + methods::MethodCall::chain_unsubscribeFinalizedHeads { .. } => { + methods::Response::chain_unsubscribeFinalizedHeads(true) + } + methods::MethodCall::chain_unsubscribeNewHeads { .. } => { + methods::Response::chain_unsubscribeNewHeads(true) + } + methods::MethodCall::state_unsubscribeRuntimeVersion { .. } => { + methods::Response::state_unsubscribeRuntimeVersion(true) + } + methods::MethodCall::state_unsubscribeStorage { .. } => { + methods::Response::state_unsubscribeStorage(true) + } + methods::MethodCall::author_unwatchExtrinsic { .. } => { + methods::Response::author_unwatchExtrinsic(true) + } + methods::MethodCall::transactionWatch_v1_unwatch { .. } => { + methods::Response::transactionWatch_v1_unwatch(()) + } + methods::MethodCall::transaction_v1_stop { .. } => { + methods::Response::transaction_v1_stop(()) + } + methods::MethodCall::chainHead_v1_unfollow { .. } => { + methods::Response::chainHead_v1_unfollow(()) + } + _ => unreachable!(), + } + .to_json_response(request_id_json); + } + + _ => {} + } + + // Success. + if let btree_map::Entry::Occupied(entry) = requests_in_progress_entry { + entry.remove(); + } + if let btree_map::Entry::Occupied(entry) = zombie_subscriptions_entry { + entry.remove(); + } + self.responses_queue.push_back(response); + InsertProxiedJsonRpcResponse::Queued + } + } + } +} + +impl fmt::Debug for ServersMultiplexer +where + T: fmt::Debug, +{ + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + struct Servers<'a, T>(&'a ServersMultiplexer); + impl<'a, T: fmt::Debug> fmt::Debug for Servers<'a, T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_list() + .entries( + self.0 + .servers + .iter() + .map(|(index, server)| (ServerId(index), &server.user_data)), + ) + .finish() + } + } + + struct Requests<'a, T>(&'a ServersMultiplexer); + impl<'a, T: fmt::Debug> fmt::Debug for Requests<'a, T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_list() + .entries( + self.0 + .requests_in_progress + .iter() + .map(|((_, rq_id), _)| rq_id) + .chain( + self.0 + .queued_requests + .values() + .flat_map(|list| list.iter()) + .map(|(rq_id, _)| rq_id), + ), + ) + .finish() + } + } + + struct Subscriptions<'a, T>(&'a ServersMultiplexer); + impl<'a, T: fmt::Debug> fmt::Debug for Subscriptions<'a, T> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_list() + .entries( + self.0 + .active_subscriptions + .iter() + .map(|(sub_id, _)| sub_id) + .chain(self.0.zombie_subscriptions.keys()), + ) + .finish() + } + } + + f.debug_struct("ServersMultiplexer") + .field("servers", &Servers(self)) + .field("requests", &Requests(self)) + .field("subscriptions", &Subscriptions(self)) + .finish() + } +} + +impl ops::Index for ServersMultiplexer { + type Output = T; + + fn index(&self, id: ServerId) -> &T { + &self.servers[id.0].user_data + } +} + +impl ops::IndexMut for ServersMultiplexer { + fn index_mut(&mut self, id: ServerId) -> &mut T { + &mut self.servers[id.0].user_data + } +} + +/// Outcome of a call to [`ServersMultiplexer::insert_json_rpc_request`]. +#[derive(Debug)] +pub enum InsertRequest { + /// The request has been silently discarded. + /// + /// This happens for example if the JSON-RPC client sends a notification. + Discarded, + + /// The request has been updated to take the request into account. No request needs to be + /// sent to any server. + LocalStateUpdate, + + /// The request has been immediately answered or discarded and doesn't need any + /// further processing. + /// [`ServersMultiplexer::next_json_rpc_response`] should be called in order to pull the response. + ImmediateAnswer, + + /// The request can be processed by any server. + /// [`ServersMultiplexer::next_proxied_json_rpc_request`] should be called with + /// any [`ServerId`]. + AnyServerWakeUp, + + /// The request must be processed specifically by the indicated server. + /// [`ServersMultiplexer::next_proxied_json_rpc_request`] should be called with the + /// given [`ServerId`]. + ServerWakeUp(ServerId), +} + +/// Outcome of a call to [`ServersMultiplexer::insert_proxied_json_rpc_response`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum InsertProxiedJsonRpcResponse { + /// The response or notification has been queued and can now be retrieved + /// using [`ServersMultiplexer::next_json_rpc_response`]. + Queued, + /// The server is misbehaving in some way. + Blacklisted(BlacklistReason), +} + +/// See [`InsertProxiedJsonRpcResponse::Blacklisted`]. +#[derive(Debug, derive_more::Display, Clone, PartialEq, Eq)] +#[non_exhaustive] + +pub enum BlacklistReason { + /// Failed to parse the server's response or notification. + ParseFailure, + /// The response sent by the server doesn't correspond to any known request. + InvalidRequestId, + /// The notification sent by the server doesn't correspond to any known subscription. + InvalidSubscriptionId, + /// Server has sent a "parse error" response indicating that it has failed to parse our of + /// our requests. + ParseErrorResponse, + /// Server has allocated a subscription ID that was already allocated against a different + /// subscription. + DuplicateSubscriptionId, +} diff --git a/light-base/src/json_rpc_service/background.rs b/light-base/src/json_rpc_service/background.rs index a20d1fd8df..26b525a428 100644 --- a/light-base/src/json_rpc_service/background.rs +++ b/light-base/src/json_rpc_service/background.rs @@ -1200,7 +1200,7 @@ pub(super) async fn run( } methods::MethodCall::chain_unsubscribeAllHeads { subscription } => { - let exists = me.all_heads_subscriptions.remove(&subscription); + let exists = me.all_heads_subscriptions.remove(&*subscription); let _ = me .responses_tx .send( @@ -1211,7 +1211,7 @@ pub(super) async fn run( } methods::MethodCall::chain_unsubscribeFinalizedHeads { subscription } => { - let exists = me.finalized_heads_subscriptions.remove(&subscription); + let exists = me.finalized_heads_subscriptions.remove(&*subscription); let _ = me .responses_tx .send( @@ -1222,7 +1222,7 @@ pub(super) async fn run( } methods::MethodCall::chain_unsubscribeNewHeads { subscription } => { - let exists = me.new_heads_subscriptions.remove(&subscription); + let exists = me.new_heads_subscriptions.remove(&*subscription); let _ = me .responses_tx .send(