diff --git a/CLAUDE.md b/CLAUDE.md index e662152..2125c9b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -210,6 +210,26 @@ block. The first cut of the roles index treated "contains an account" as a role and labelled half the chain's active addresses `Events`. The test is `value` being a string that is itself in the account set. +**A JSON-RPC error is the node answering; a transport failure is not.** +`RpcClient::call` fails over to the next endpoint on the second and never on the +first. Moving on a real error would hide a caller's own mistake behind a second +node making the same complaint — and would walk the whole endpoint list to do +it. `DataError::Rpc` means "answered, with an error" and stops; `Malformed`, +`Http` and timeouts mean "did not answer" and move on. + +**A pruned block is a success, not an unhealthy endpoint.** `state_getStorage` +for state a node has dropped returns `{"result": null}`, which reaches the +caller as `Null`. Treating that as a failure would fail over through every +endpoint asking a question none of them can answer, and turn a legitimate +absence into an outage. + +**Endpoints are stuck to, never round-robined.** A storage read at an old block +hash needs a node that still holds that block's state, and nodes prune on their +own schedules — alternating between them returns a mixture of answers and +absences that reads as sparse data rather than as a configuration problem. The +cursor is shared across clones so a failover one task discovers is not +rediscovered by every other task on that chain. + **`state_getKeysPaged` rejects a count over 1,000, it does not truncate.** Asking for 1,600 is an RPC *error*, and a caller that `unwrap_or_default()`s it has turned "could not ask" into "the answer is nothing" — which for a map like diff --git a/asset/config/config.toml.tmpl b/asset/config/config.toml.tmpl index 5a84362..0d872e5 100644 --- a/asset/config/config.toml.tmpl +++ b/asset/config/config.toml.tmpl @@ -115,8 +115,8 @@ genesis = "0xa5aa9e5c84d4a3722c152295e7973c9af522f2fb1ef7db5afaa3d5f4dc8d3b4f" # like the local node: same decoding, same leaderboard, same everything — which # is the point. It is third-party infrastructure, so it may disappear without # notice; the chain then reports `unreachable` and keeps its last standings. -rpc_url = "https://a1-heisenberg.quantus.cat" -ws_url = "wss://a1-heisenberg.quantus.cat" +rpc_urls = ["https://a1-heisenberg.quantus.cat", "https://a2-heisenberg.quantus.cat"] +ws_urls = ["wss://a1-heisenberg.quantus.cat", "wss://a2-heisenberg.quantus.cat"] target_block_time_seconds = 6.0 warm_start_blocks = 100800 max_gap_fill_blocks = 5000 @@ -134,8 +134,12 @@ genesis = "0x4901bf5c57fd3f9e726af399c763de6670dbdb115a91c0237e173f16eef65e72" # decentralised testnet infrastructure is what it is for. Measured alongside the # other two chains it costs about 5% of one core and does not move API latency — # the walk is bound by RPC round trips, not by anything here. -rpc_url = "https://a1-planck.quantus.cat" -ws_url = "wss://a1-planck.quantus.cat" +# Both public nodes, tried in order and stuck to rather than balanced across. +# One host being down should not decide what a whole chain looks like, and a +# storage read needs a node that still holds that block's state — see +# `RpcClient` for why alternating would be worse than a single endpoint. +rpc_urls = ["https://a1-planck.quantus.cat", "https://a2-planck.quantus.cat"] +ws_urls = ["wss://a1-planck.quantus.cat", "wss://a2-planck.quantus.cat"] # A testnet, so the nav files it under `TESTNETS` and says its balances are not # real. Never inferred from the name — see `mainnet` above. target_block_time_seconds = 6.0 diff --git a/crates/blackbeard-api/src/config.rs b/crates/blackbeard-api/src/config.rs index af39ec3..a6e4b64 100644 --- a/crates/blackbeard-api/src/config.rs +++ b/crates/blackbeard-api/src/config.rs @@ -146,10 +146,27 @@ pub struct ChainConfig { #[serde(default)] pub mainnet: bool, /// HTTP JSON-RPC endpoint, e.g. `http://127.0.0.1:9944`. - pub rpc_url: String, + /// + /// Superseded by [`Self::rpc_urls`] and kept because a config naming one + /// endpoint is still a valid config — there is no reason to make every + /// single-node chain rewrite its entry. + #[serde(default)] + pub rpc_url: Option, + /// Several HTTP endpoints for the same chain, tried in order. + /// + /// One host being down should not decide what a whole chain looks like. + /// These are tried in order and **stuck to** rather than balanced across — + /// see `RpcClient`, where the reasoning about pruning lives. + #[serde(default)] + pub rpc_urls: Vec, /// WebSocket JSON-RPC endpoint for the head subscription. Usually the same /// host and port as `rpc_url` — `9944` serves both. - pub ws_url: String, + #[serde(default)] + pub ws_url: Option, + /// Several WebSocket endpoints, used one at a time and rotated on + /// reconnect. + #[serde(default)] + pub ws_urls: Vec, /// The chain's target seconds per block. Used as the hashrate denominator /// whenever no measured interval is trustworthy. #[serde(default = "default_target_block_time")] @@ -170,6 +187,36 @@ pub struct ChainConfig { pub max_gap_fill_blocks: u64, } +impl ChainConfig { + /// Every HTTP endpoint for this chain, in the order to try them. + /// + /// `rpc_url` and `rpc_urls` both contribute, so a config can name one, or + /// several, or add a second to an existing entry without restructuring it. + /// Deduplicated because naming the same host twice would make a failover + /// list that fails over to itself. + pub fn rpc_endpoints(&self) -> Vec { + dedup(self.rpc_url.iter().chain(self.rpc_urls.iter())) + } + + /// Every WebSocket endpoint, in the order to try them. + /// + /// Falls back to none rather than to the HTTP ones: `ws://` and `http://` + /// are different schemes, and guessing one from the other would produce a + /// url that fails at connect for a reason nothing would explain. + pub fn ws_endpoints(&self) -> Vec { + dedup(self.ws_url.iter().chain(self.ws_urls.iter())) + } +} + +/// Preserve order, drop repeats. +fn dedup<'a>(urls: impl Iterator) -> Vec { + let mut seen = std::collections::HashSet::new(); + urls.filter(|u| !u.trim().is_empty()) + .filter(|u| seen.insert((*u).clone())) + .cloned() + .collect() +} + fn default_pg_port() -> u16 { 5432 } @@ -243,6 +290,14 @@ impl Config { chain.id ); } + if chain.rpc_endpoints().is_empty() { + anyhow::bail!( + "chain `{}` names no rpc endpoint; set `rpc_url` or `rpc_urls`. A chain entry \ + exists to give a chain an endpoint, so one without any would be listed and \ + never read", + chain.id + ); + } if chain.target_block_time_seconds <= 0.0 { anyhow::bail!( "chain `{}` has a target block time of {}; hashrate is difficulty divided by it", @@ -272,6 +327,72 @@ impl Config { mod tests { use super::*; + fn chain(rpc_url: Option<&str>, rpc_urls: &[&str]) -> ChainConfig { + ChainConfig { + id: "x".into(), + display_name: "X".into(), + genesis: None, + mainnet: false, + rpc_url: rpc_url.map(str::to_owned), + rpc_urls: rpc_urls.iter().map(|s| (*s).to_owned()).collect(), + ws_url: None, + ws_urls: Vec::new(), + target_block_time_seconds: 6.0, + warm_start_blocks: 10, + max_gap_fill_blocks: 10, + } + } + + /// A config naming one endpoint is still a valid config. Every existing + /// deployment writes `rpc_url`, and requiring them all to change would be + /// this feature breaking the thing it exists to make more reliable. + #[test] + fn a_single_endpoint_still_works() { + assert_eq!( + chain(Some("http://a"), &[]).rpc_endpoints(), + vec!["http://a"] + ); + } + + /// Both forms contribute, so a second endpoint can be added to an existing + /// entry without restructuring it — and the singular comes first, because + /// it is the one that was already working. + #[test] + fn both_forms_combine_in_order() { + assert_eq!( + chain(Some("http://a"), &["http://b", "http://c"]).rpc_endpoints(), + vec!["http://a", "http://b", "http://c"] + ); + } + + /// A list that repeats a host is a failover list that fails over to itself. + #[test] + fn a_repeated_endpoint_is_listed_once() { + assert_eq!( + chain(Some("http://a"), &["http://a", "http://b"]).rpc_endpoints(), + vec!["http://a", "http://b"] + ); + } + + #[test] + fn a_chain_with_no_endpoint_at_all_is_refused() { + // Not merely empty: a `[[chains]]` entry exists to give a chain an + // endpoint, so one without any would be listed and never read — the + // failure this validation exists to catch. + assert!(chain(None, &[]).rpc_endpoints().is_empty()); + } + + /// `ws://` and `http://` are different schemes. Deriving one from the other + /// would produce a url that fails at connect for a reason nothing explains. + #[test] + fn websocket_endpoints_are_not_guessed_from_http_ones() { + assert!( + chain(Some("http://a"), &["http://b"]) + .ws_endpoints() + .is_empty() + ); + } + fn write(dir: &std::path::Path, name: &str, body: &str) -> PathBuf { let p = dir.join(name); std::fs::write(&p, body).unwrap(); diff --git a/crates/blackbeard-api/src/ingest.rs b/crates/blackbeard-api/src/ingest.rs index f6457bd..6db3752 100644 --- a/crates/blackbeard-api/src/ingest.rs +++ b/crates/blackbeard-api/src/ingest.rs @@ -148,7 +148,7 @@ pub fn spawn(chain: Arc, store: Store, config: &crate::config::Con let ticker_blocks = config.server.ticker_blocks; let refresh = Duration::from_secs(config.server.leaderboard_refresh_seconds); - tokio::spawn(rpc::subscribe_new_heads(chain.spec.ws_url.clone(), tx)); + tokio::spawn(rpc::subscribe_new_heads(chain.spec.ws_urls.clone(), tx)); tokio::spawn(ingest(Arc::clone(&chain), store.clone(), rx, ticker_blocks)); tokio::spawn(poll( Arc::clone(&chain), diff --git a/crates/blackbeard-api/src/registry.rs b/crates/blackbeard-api/src/registry.rs index ec1814b..40a75ee 100644 --- a/crates/blackbeard-api/src/registry.rs +++ b/crates/blackbeard-api/src/registry.rs @@ -42,9 +42,11 @@ pub struct ChainSpec { /// Staging Mainnet" is exactly why this is not derived from the name. pub mainnet: bool, /// HTTP JSON-RPC endpoint. - pub rpc_url: String, + /// Every HTTP endpoint for this chain, in the order to try them. + pub rpc_urls: Vec, /// WebSocket JSON-RPC endpoint for the head subscription. - pub ws_url: String, + /// Every WebSocket endpoint, rotated on reconnect. + pub ws_urls: Vec, /// Target seconds per block, the hashrate denominator of last resort. pub target_block_time_seconds: f64, /// Blocks replayed from the database at startup. @@ -65,8 +67,8 @@ impl ChainSpec { display_name: chain.display_name.clone(), genesis: chain.genesis.clone(), mainnet: chain.mainnet, - rpc_url: chain.rpc_url.clone(), - ws_url: chain.ws_url.clone(), + rpc_urls: chain.rpc_endpoints(), + ws_urls: chain.ws_endpoints(), target_block_time_seconds: chain.target_block_time_seconds, warm_start_blocks: chain.warm_start_blocks, max_gap_fill_blocks: chain.max_gap_fill_blocks, @@ -123,12 +125,12 @@ impl Registry { for chain in &config.chains { let spec = ChainSpec::from_config(chain); - let rpc = match RpcClient::new(spec.rpc_url.clone(), crate::RPC_TIMEOUT) { + let rpc = match RpcClient::with_endpoints(spec.rpc_urls.clone(), crate::RPC_TIMEOUT) { Ok(rpc) => rpc, Err(e) => { // A malformed endpoint is a config error, but it must not // stop the other chains from being served. - tracing::error!(chain = %spec.slug, url = %spec.rpc_url, error = %e, + tracing::error!(chain = %spec.slug, urls = ?spec.rpc_urls, error = %e, "unusable rpc endpoint; this chain will not be tracked"); continue; } diff --git a/crates/blackbeard-api/src/state.rs b/crates/blackbeard-api/src/state.rs index f080778..13be1b4 100644 --- a/crates/blackbeard-api/src/state.rs +++ b/crates/blackbeard-api/src/state.rs @@ -478,8 +478,8 @@ mod tests { display_name: "Planck".into(), genesis: None, mainnet: false, - rpc_url: "http://127.0.0.1:9944".into(), - ws_url: "ws://127.0.0.1:9944".into(), + rpc_urls: vec!["http://127.0.0.1:9944".into()], + ws_urls: vec!["ws://127.0.0.1:9944".into()], target_block_time_seconds: 6.0, warm_start_blocks: 100, max_gap_fill_blocks: 100, diff --git a/crates/blackbeard-data/src/lib.rs b/crates/blackbeard-data/src/lib.rs index 26b5af8..4960a8d 100644 --- a/crates/blackbeard-data/src/lib.rs +++ b/crates/blackbeard-data/src/lib.rs @@ -40,6 +40,20 @@ pub enum DataError { #[error("decoding a response: {0}")] Decode(#[from] serde_json::Error), + /// A node answered with something that was not a JSON-RPC response at all. + /// + /// Distinct from [`Self::Rpc`], which is the node answering *correctly* + /// with an error. This one is a host misbehaving, and so is worth trying + /// another endpoint for — the difference is what keeps a caller's own + /// mistake from walking the whole endpoint list. + #[error("malformed response to {method}: {message}")] + Malformed { + /// Which method. + method: String, + /// What was wrong with it. + message: String, + }, + /// A database query failed. #[error("database: {0}")] Database(#[from] sqlx::Error), diff --git a/crates/blackbeard-data/src/rpc.rs b/crates/blackbeard-data/src/rpc.rs index b136f3a..cf01212 100644 --- a/crates/blackbeard-data/src/rpc.rs +++ b/crates/blackbeard-data/src/rpc.rs @@ -17,6 +17,8 @@ //! localhost — which would mean either TLS or an ssh tunnel between the //! observer and a node sitting on the same host. +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Duration; use futures_util::{SinkExt, StreamExt}; @@ -119,16 +121,42 @@ pub struct ChainProperties { /// looks exactly like a right one. pub const MAX_KEYS_PER_PAGE: u32 = 1000; -/// An HTTP JSON-RPC client for one node. +/// An HTTP JSON-RPC client for one chain, across however many nodes serve it. +/// +/// ## Failover, not round-robin +/// +/// Requests stick to one endpoint until it fails, and only then move. Spreading +/// consecutive calls across nodes would be worse than a single endpoint, not +/// better: a storage read at an old block hash needs a node that still holds +/// that block's state, and nodes prune on their own schedules — so alternating +/// would return a mixture of answers and absences that reads as sparse data +/// rather than as a configuration problem. +/// +/// The cursor is shared across clones. Every task on a chain holds a clone of +/// the same client, and a failover one of them discovers is one the rest should +/// not have to rediscover. #[derive(Debug, Clone)] pub struct RpcClient { http: reqwest::Client, - url: String, + endpoints: Arc>, + /// Index into `endpoints` to try first. + current: Arc, } impl RpcClient { /// Build a client for `url` (`http://host:9944`). pub fn new(url: impl Into, timeout: Duration) -> Result { + Self::with_endpoints(vec![url.into()], timeout) + } + + /// A client over several endpoints for the same chain. + pub fn with_endpoints(endpoints: Vec, timeout: Duration) -> Result { + if endpoints.is_empty() { + return Err(DataError::Rpc { + method: "new".into(), + message: "a chain needs at least one endpoint".into(), + }); + } Ok(Self { http: reqwest::Client::builder() .timeout(timeout) @@ -137,17 +165,77 @@ impl RpcClient { // handshake. .pool_idle_timeout(Duration::from_secs(90)) .build()?, - url: url.into(), + endpoints: Arc::new(endpoints), + current: Arc::new(AtomicUsize::new(0)), }) } + /// The endpoint currently in use, for logs and for saying which of several + /// is answering — "the chain is unreachable" and "one of three endpoints is + /// unreachable" are different operational facts. + pub fn endpoint(&self) -> &str { + &self.endpoints[self.current.load(Ordering::Relaxed) % self.endpoints.len()] + } + + /// How many endpoints this chain has. + pub fn endpoint_count(&self) -> usize { + self.endpoints.len() + } + /// Issue one JSON-RPC call. pub async fn call(&self, method: &str, params: Value) -> Result { let body = json!({"jsonrpc": "2.0", "id": 1, "method": method, "params": params}); + let start = self.current.load(Ordering::Relaxed); + let mut last: Option = None; + + for attempt in 0..self.endpoints.len() { + let index = (start + attempt) % self.endpoints.len(); + let url = &self.endpoints[index]; + match self.call_one(url, &body, method).await { + Ok(value) => { + // Stick here. Only worth a write when it actually moved, + // which is once per outage rather than once per call. + if index != start { + self.current.store(index, Ordering::Relaxed); + tracing::warn!( + from = %self.endpoints[start], to = %url, + "rpc endpoint failed over" + ); + } + return Ok(value); + } + // A JSON-RPC *error* is the node answering. `count exceeds + // maximum value` is not a reason to believe the host is down, + // and moving on it would hide a caller's mistake behind a + // second node making the same complaint. + Err(e @ DataError::Rpc { .. }) => return Err(e), + Err(e) => { + tracing::debug!(url = %url, method, error = %e, "rpc endpoint unreachable"); + last = Some(e); + } + } + } + Err(last.unwrap_or_else(|| DataError::Rpc { + method: method.to_owned(), + message: "no endpoint answered".into(), + })) + } + + /// One request against one endpoint. + /// + /// A transport failure comes back as its own error variant so `call` can + /// tell "this host is not answering" from "this host answered, with an + /// error" — the first is worth another endpoint and the second never is. + /// + /// A node that has pruned a block returns `{"result": null}`, which is a + /// success and reaches the caller as `Null`. It must never look like an + /// unhealthy endpoint: pruning is a legitimate answer, and failing over on + /// it would walk the whole list asking a question none of them can answer. + async fn call_one(&self, url: &str, body: &Value, method: &str) -> Result { let resp: Value = self .http - .post(&self.url) - .json(&body) + .post(url) + .json(body) .send() .await? .error_for_status()? @@ -159,10 +247,12 @@ impl RpcClient { message: err.to_string(), }); } - resp.get("result").cloned().ok_or_else(|| DataError::Rpc { - method: method.to_owned(), - message: "response carried neither result nor error".into(), - }) + resp.get("result") + .cloned() + .ok_or_else(|| DataError::Malformed { + method: method.to_owned(), + message: "response carried neither result nor error".into(), + }) } /// `system_health`. @@ -430,15 +520,25 @@ const RECONNECT_DELAY: Duration = Duration::from_secs(5); /// so the caller must fill gaps against `chain_getBlockHash`. It is a liveness /// signal, not a ledger. pub async fn subscribe_new_heads( - ws_url: String, + ws_urls: Vec, sink: mpsc::Sender
, ) -> Result<(), DataError> { + if ws_urls.is_empty() { + return Ok(()); + } let mut backoff_logged = false; + // Failover happens **at reconnect**, which is where this loop already is. + // The alternative — holding subscriptions to every endpoint at once and + // deduplicating heads — buys nothing: heads are a liveness signal and + // `ingest` fills gaps against `chain_getBlockHash` regardless, so a few + // seconds on the next endpoint costs a reconnect rather than data. + let mut index = 0usize; loop { if sink.is_closed() { return Ok(()); } - match follow_once(&ws_url, &sink).await { + let ws_url = &ws_urls[index % ws_urls.len()]; + match follow_once(ws_url, &sink).await { Ok(()) => { tracing::info!(url = %ws_url, "head subscription closed cleanly, reconnecting"); backoff_logged = false; @@ -450,9 +550,15 @@ pub async fn subscribe_new_heads( if backoff_logged { tracing::debug!(url = %ws_url, error = %e, "head subscription still down"); } else { - tracing::warn!(url = %ws_url, error = %e, "head subscription lost"); + tracing::warn!( + url = %ws_url, endpoints = ws_urls.len(), error = %e, + "head subscription lost" + ); backoff_logged = true; } + // Only a failure advances. A clean close is the node saying + // goodbye, not a reason to abandon an endpoint that works. + index = index.wrapping_add(1); } } tokio::time::sleep(RECONNECT_DELAY).await; diff --git a/readme.md b/readme.md index af2c647..e638c51 100644 --- a/readme.md +++ b/readme.md @@ -528,6 +528,37 @@ deliberate exception is a height resolving to its hash, which *replaces* rather than pushes — it is the same block under a better name, not a second place the reader has been. +## No single node can stop a chain + +A `[[chains]]` entry takes `rpc_urls` and `ws_urls`, tried in order. `rpc_url` +and `ws_url` still work as a one-element list, because a config naming one +endpoint is still a valid config. + +They are **stuck to, not balanced across**, and that is the whole design. A +storage read at an old block hash needs a node that still holds that block's +state; nodes prune on their own schedules, so alternating between them would +return a mixture of answers and absences that reads as sparse data rather than +as a configuration problem. Requests stay on one endpoint until it fails, and +the cursor is shared across every task on the chain so a failover one of them +discovers is not rediscovered by the rest. + +Failing over on the wrong thing is the trap. A **JSON-RPC error is the node +answering** — `count exceeds maximum value` is a caller's mistake, and moving on +it would hide that behind a second node making the same complaint. A **pruned +block is a success**: `state_getStorage` returns `{"result": null}`, and treating +that as a failure would walk every endpoint asking a question none of them can +answer. Only transport failures and malformed responses move. + +The WebSocket rotates at reconnect, where the loop already was. Holding +subscriptions to every endpoint and deduplicating heads buys nothing: heads are +a liveness signal and `ingest` fills gaps against `chain_getBlockHash` anyway, so +a few seconds on the next endpoint costs a reconnect rather than data. + +Verified against a chain configured with a dead endpoint first: Planck stayed +`full` at its live height, and the failover logged once — from, to, and how many +endpoints the chain has, because "the chain is unreachable" and "one of three +endpoints is unreachable" are different operational facts. + ## Backfill is cheap, and bound by somebody else's node Three chains walking their history at once — mainnet, Heisenberg and Planck's