diff --git a/CLAUDE.md b/CLAUDE.md index 09cac30..8bb4d60 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -251,6 +251,22 @@ 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. +**Which is why depth is a property of the endpoint, not of the response.** Null +is also the right answer for most keys read here — an account with no balance, +an item never set, the treasury's `System::Account` entry. So `classify_depth` +probes each endpoint once at startup, reading `System::Number` at block one: an +endpoint that answers holds everything after it. Reads that name a block hash go +through `call_deep`, which tries the sticky endpoint first — costing nothing in +the common case — and asks an archive for a second opinion only when a +*non-archive* endpoint returned null, or refused with an RPC error, which is +what `state_call` does when it cannot execute against dropped state. A null from +an archive is the answer and stops there. An endpoint that could not be probed +stays `Unknown` and routes as pruned: depth is demonstrated, never assumed, or +an unreachable host silently becomes the site's archive of record. All seven +configured endpoints probe as archives today, so none of this changes any +current behaviour — it is what keeps history from quietly going blank the day +one of them stops being one. + **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 diff --git a/crates/blackbeard-api/src/ingest.rs b/crates/blackbeard-api/src/ingest.rs index 43c4a6e..7a5bc45 100644 --- a/crates/blackbeard-api/src/ingest.rs +++ b/crates/blackbeard-api/src/ingest.rs @@ -148,6 +148,15 @@ 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); + // Which endpoints hold old state, before anything asks them for any. Its + // own task because the probe is one round trip per endpoint and nothing + // else waits on it: until it answers, every endpoint routes as shallow, + // which is the same behaviour this had before the probe existed. + { + let chain = Arc::clone(&chain); + tokio::spawn(async move { chain.rpc.classify_depth().await }); + } + 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( diff --git a/crates/blackbeard-core/src/runtime.rs b/crates/blackbeard-core/src/runtime.rs index d1a2240..dc8d2f7 100644 --- a/crates/blackbeard-core/src/runtime.rs +++ b/crates/blackbeard-core/src/runtime.rs @@ -280,7 +280,11 @@ fn hash_prefix_len(hasher: &frame_metadata::v14::StorageHasher) -> Option /// `twox128`, as Substrate builds storage prefixes: two little-endian xxhash64 /// digests, seeded 0 and 1, concatenated. -fn twox_128(input: &[u8]) -> [u8; 16] { +/// +/// Public so a caller outside this crate can *derive* a well-known key rather +/// than paste one: a mistyped storage key is absent everywhere, and absent is +/// indistinguishable from an empty entry. +pub fn twox_128(input: &[u8]) -> [u8; 16] { use twox_hash::XxHash64; let mut out = [0u8; 16]; out[..8].copy_from_slice(&XxHash64::oneshot(0, input).to_le_bytes()); diff --git a/crates/blackbeard-data/src/rpc.rs b/crates/blackbeard-data/src/rpc.rs index cf01212..932fb45 100644 --- a/crates/blackbeard-data/src/rpc.rs +++ b/crates/blackbeard-data/src/rpc.rs @@ -18,7 +18,7 @@ //! observer and a node sitting on the same host. use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering}; use std::time::Duration; use futures_util::{SinkExt, StreamExt}; @@ -141,8 +141,65 @@ pub struct RpcClient { endpoints: Arc>, /// Index into `endpoints` to try first. current: Arc, + /// What each endpoint was found to hold, parallel to `endpoints`. + depth: Arc>, } +/// Whether an endpoint still holds the state of old blocks. +/// +/// Failover buys availability, not depth, and the two are not the same problem. +/// `call` moves on a transport failure and never on a JSON-RPC error, which is +/// right — but a node that has pruned the state being asked for does neither. +/// It answers `{"result": null}`, and that is a *success*: the caller is +/// satisfied, and whichever endpoint happens to be sticky decides how far back +/// the whole site can see. +/// +/// The distinction is not in the response, because null is also the correct +/// answer for most of the keys read here — an account with no balance, an item +/// never set, the treasury's `System::Account` entry, which is absent because +/// the treasury was never funded. Retrying every one of those against every +/// endpoint would triple the load to re-derive an answer already in hand. The +/// distinction is whether the node *could* have known, and that is a property +/// of the endpoint rather than of the request. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Depth { + /// Not probed, or the probe could not reach it. Treated as shallow for + /// routing — never assume depth that has not been demonstrated. + Unknown, + /// Answered for block one's state, so it holds everything after it too. + Archive, + /// Did not. It can still serve the tip, and mostly does. + Pruned, +} + +impl Depth { + fn code(self) -> u8 { + match self { + Depth::Unknown => 0, + Depth::Archive => 1, + Depth::Pruned => 2, + } + } + + fn from_code(code: u8) -> Self { + match code { + 1 => Depth::Archive, + 2 => Depth::Pruned, + _ => Depth::Unknown, + } + } +} + +/// `twox128("System") ++ twox128("Number")` — the block number in state. +/// +/// The probe key, and it has to be one that certainly exists: a key genuinely +/// absent at block one would make every endpoint look pruned. `System::Number` +/// is written by every block of every FRAME chain and is four bytes, so the +/// probe costs one small round trip per endpoint. `:code` would also do and is +/// megabytes; the runtime version would do and takes seconds. +const SYSTEM_NUMBER_KEY: &str = + "0x26aa394eea5630e07c48ae0c9558cef702a5c1b19ab7a04f536c519aca4983ac"; + impl RpcClient { /// Build a client for `url` (`http://host:9944`). pub fn new(url: impl Into, timeout: Duration) -> Result { @@ -165,6 +222,7 @@ impl RpcClient { // handshake. .pool_idle_timeout(Duration::from_secs(90)) .build()?, + depth: Arc::new(endpoints.iter().map(|_| AtomicU8::new(0)).collect()), endpoints: Arc::new(endpoints), current: Arc::new(AtomicUsize::new(0)), }) @@ -182,6 +240,113 @@ impl RpcClient { self.endpoints.len() } + /// What this endpoint was found to hold. + pub fn depth_of(&self, index: usize) -> Depth { + Depth::from_code(self.depth[index].load(Ordering::Relaxed)) + } + + /// Every endpoint with what it holds, for logging and for the chain page. + pub fn depths(&self) -> Vec<(&str, Depth)> { + self.endpoints + .iter() + .enumerate() + .map(|(i, url)| (url.as_str(), self.depth_of(i))) + .collect() + } + + /// Ask each endpoint whether it still holds block one's state. + /// + /// Once, at startup. An endpoint that can answer for block one holds + /// everything after it, so one probe settles the whole range and there is + /// nothing to re-check: a node does not become an archive later, and one + /// that stops being one gets a restart of this service anyway. + /// + /// The probe reads `System::Number` at block one directly against each + /// endpoint, deliberately bypassing `call` — the point is to learn about + /// *this* endpoint, and failover would answer from a different one and + /// attribute it to the wrong host. + /// + /// An endpoint that cannot be reached stays `Unknown`, which routes exactly + /// as `Pruned` does. Depth is a claim to be demonstrated, never assumed. + pub async fn classify_depth(&self) { + let Ok(Some(block_one)) = self.block_hash(1).await else { + tracing::debug!("endpoint depth unclassified: no block one yet"); + return; + }; + let body = json!({ + "jsonrpc": "2.0", "id": 1, "method": "state_getStorage", + "params": [SYSTEM_NUMBER_KEY, block_one], + }); + + for (index, url) in self.endpoints.iter().enumerate() { + let found = match self.call_one(url, &body, "state_getStorage").await { + Ok(Value::Null) => Depth::Pruned, + Ok(_) => Depth::Archive, + Err(e) => { + tracing::debug!(url = %url, error = %e, "endpoint depth unknown"); + continue; + } + }; + self.depth[index].store(found.code(), Ordering::Relaxed); + tracing::info!(url = %url, depth = ?found, "endpoint depth"); + } + } + + /// Issue a call whose answer depends on state the node may have dropped. + /// + /// The ordinary `call` first, so nothing here costs an extra round trip in + /// the common case — the sticky endpoint usually has the state, and on + /// mainnet it is our own loopback node. What this adds is the one case + /// `call` gets wrong: a `null` from an endpoint that was never shown to + /// hold old state is not an answer, it is an absence of evidence, and there + /// may be a peer that knows better. + /// + /// So on a null from a non-archive endpoint, ask the archives. A null from + /// an archive is the answer and stops there. Never used for tip reads, + /// where every endpoint has the state and a null is simply the truth. + /// A node that cannot *execute* against pruned state — `state_call`, which + /// is how difficulty is read — refuses instead, with a JSON-RPC error. That + /// is still the node answering, so `call` is right not to fail over on it; + /// but on this path it means the same thing a null does, and gets the same + /// second opinion. + async fn call_deep(&self, method: &str, params: Value) -> Result { + let first = self.call(method, params.clone()).await; + match &first { + Ok(v) if !v.is_null() => return first, + Err(DataError::Rpc { .. }) => {} + Err(_) => return first, + Ok(_) => {} + } + let answered = self.current.load(Ordering::Relaxed) % self.endpoints.len(); + if self.depth_of(answered) == Depth::Archive { + return first; + } + + let body = json!({"jsonrpc": "2.0", "id": 1, "method": method, "params": params}); + for (index, url) in self.endpoints.iter().enumerate() { + if index == answered || self.depth_of(index) != Depth::Archive { + continue; + } + match self.call_one(url, &body, method).await { + Ok(Value::Null) => continue, + Ok(value) => { + // Worth saying out loud: it means the endpoint being used + // cannot see as far back as the site is asking, which is a + // configuration fact rather than a chain fact. + tracing::info!( + method, shallow = %self.endpoints[answered], archive = %url, + "an archive endpoint answered what the current one could not" + ); + return Ok(value); + } + Err(e) => { + tracing::debug!(url = %url, method, error = %e, "archive endpoint failed") + } + } + } + first + } + /// 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}); @@ -332,7 +497,14 @@ impl RpcClient { (None, Some(h)) => json!([prefix, count, null, h]), (None, None) => json!([prefix, count]), }; - let v = self.call("state_getKeysPaged", params).await?; + // Depth-aware: `at` is what makes genesis readable, and a node that + // has dropped that state answers with an empty page rather than an + // error — which reads as "the map is empty" and is how the endowed set + // would quietly disappear. + let v = match at { + Some(_) => self.call_deep("state_getKeysPaged", params).await?, + None => self.call("state_getKeysPaged", params).await?, + }; Ok(v.as_array() .map(|a| { a.iter() @@ -355,7 +527,10 @@ impl RpcClient { Some(h) => json!([h]), None => json!([]), }; - let v = self.call("state_getMetadata", params).await?; + let v = match hash { + Some(_) => self.call_deep("state_getMetadata", params).await?, + None => self.call("state_getMetadata", params).await?, + }; let hex = v.as_str().ok_or_else(|| DataError::Rpc { method: "state_getMetadata".into(), message: "metadata was not a hex string".into(), @@ -404,7 +579,10 @@ impl RpcClient { Some(h) => json!([key, h]), None => json!([key]), }; - let v = self.call("state_getStorage", params).await?; + let v = match hash { + Some(_) => self.call_deep("state_getStorage", params).await?, + None => self.call("state_getStorage", params).await?, + }; let Some(hex) = v.as_str() else { return Ok(None); }; @@ -492,7 +670,14 @@ impl RpcClient { Some(hash) => json!([api, "0x", hash]), None => json!([api, "0x"]), }; - let v = self.call("state_call", params).await?; + // Depth-aware when it names a block: `difficulty_for_child_of` asks at + // a parent hash, and the honest answer from a node without that state + // is "not known" rather than today's difficulty — but only after asking + // a node that would know. + let v = match at { + Some(_) => self.call_deep("state_call", params).await?, + None => self.call("state_call", params).await?, + }; let hex = v.as_str().ok_or_else(|| DataError::Rpc { method: api.to_owned(), message: "runtime call did not return a hex string".into(), @@ -616,6 +801,43 @@ async fn follow_once(ws_url: &str, sink: &mpsc::Sender
) -> Result<(), Da mod tests { use super::*; + /// The probe key is derived, not typed. A wrong key is absent everywhere, + /// which would classify every endpoint as pruned and route every historical + /// read to nobody. + #[test] + fn the_probe_key_is_system_number() { + let expected = format!( + "0x{}{}", + hex::encode(blackbeard_core::runtime::twox_128(b"System")), + hex::encode(blackbeard_core::runtime::twox_128(b"Number")), + ); + assert_eq!(SYSTEM_NUMBER_KEY, expected); + } + + /// Depth is demonstrated, never assumed: an endpoint nobody could reach + /// must route exactly as a pruned one does, or an unreachable host would + /// silently become the site's archive of record. + #[test] + fn an_unprobed_endpoint_is_not_treated_as_an_archive() { + let client = RpcClient::with_endpoints( + vec!["http://a.invalid".into(), "http://b.invalid".into()], + Duration::from_secs(1), + ) + .expect("builds"); + assert_eq!(client.depth_of(0), Depth::Unknown); + assert_ne!(client.depth_of(0), Depth::Archive); + assert_eq!(client.depths().len(), 2); + } + + #[test] + fn depth_codes_round_trip() { + for d in [Depth::Unknown, Depth::Archive, Depth::Pruned] { + assert_eq!(Depth::from_code(d.code()), d); + } + // Anything unrecognised is unknown, which routes as shallow. + assert_eq!(Depth::from_code(99), Depth::Unknown); + } + #[test] fn header_height_decodes_hex() { let h: Header = serde_json::from_value(json!({