rob thijssen
aaf37baaca
The site answered "who is mining" thoroughly and "what is this economy doing"
not at all. Everything needed was already reachable and none of it was
assembled anywhere.
Measured against mainnet rather than assumed:
total issuance 5,682,913.17 QTC
endowed at genesis 5,670,000.00 QTC
mined since 12,913.17 QTC <- emission to date, printed nowhere else
accounts 1,925 counted, not inferred from indexed events
locks/holds/freezes/reserves 0 all four, counted
vesting schedules 48
referenda 0
reversible transfers 0
calls used 7 of 58
events fired 18 of 107
**No "circulating supply" field, deliberately.** With every immobilising map
empty it would equal total issuance exactly, and printing it as a separate
headline would assert a distinction this chain does not currently make. The page
states what was counted and lets that be read. Vesting is a count rather than a
sum for the same reason: this runtime's vesting does not touch the balances
locks, so what it holds and when it releases would be a guess.
**Signed extrinsics are separated from the total**, because three quarters of
this chain's extrinsics are inherents — 513 signed of 2,052 on the day measured
— and a single "transactions per day" line would be mostly clockwork, a number
that reads as adoption and is not.
Two kinds of certainty share the page and are kept apart. Supply and the map
sizes are chain state, read now. The daily activity is our index, and the
Activity panel prints the range it covered and says "the whole chain" or "a
partial index" — a chart whose axis claims a month over a week of data is the
failure this repository has already shipped twice.
One thing found while building it: a counter that has never moved is *absent*
from storage, not zero. `TechReferenda::ReferendumCount` and
`ReversibleTransfers::NextTransactionId` both read as nothing, and reporting
them unknown would have hidden the most interesting fact about them — that
governance and reversible transfers are shipped and have never been used.
`plain_value` falls back to the entry's declared default, which is only sound
because `StorageTarget` says whether it is `Default` or `Optional`.
Refs #17
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Jp6a8EDar9ueEhAxzep4V5
918 lines
38 KiB
Rust
918 lines
38 KiB
Rust
//! Talking to a Quantus node's JSON-RPC.
|
|
//!
|
|
//! Two transports on the **same** port, because `9944` serves HTTP and
|
|
//! WebSocket both:
|
|
//!
|
|
//! - [`RpcClient`] over HTTP for request/response — block hashes, bodies,
|
|
//! runtime calls. Stateless, so a node restart costs one failed request
|
|
//! rather than a reconnect dance.
|
|
//! - [`subscribe_new_heads`] over WebSocket for the head stream. The node
|
|
//! pushes, so the observer learns about a block the moment the node imports
|
|
//! it instead of up to a poll interval later — which is what lets the
|
|
//! browser's socket be genuinely live rather than merely frequent.
|
|
//!
|
|
//! No `subxt`. It would bring runtime metadata, a codec and a type registry to
|
|
//! read two fixed-shape fields the observer already decodes by hand
|
|
//! (`blackbeard-core::digest`), and it refuses plain `ws://` to anything but
|
|
//! 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::{AtomicU8, AtomicUsize, Ordering};
|
|
use std::time::Duration;
|
|
|
|
use futures_util::{SinkExt, StreamExt};
|
|
use serde::Deserialize;
|
|
use serde_json::{Value, json};
|
|
use tokio::sync::mpsc;
|
|
|
|
use crate::DataError;
|
|
|
|
/// A Substrate block header, reduced to the fields the observer reads.
|
|
///
|
|
/// `#[serde(rename_all = "camelCase")]` matches the RPC's JSON. Unknown fields
|
|
/// are ignored rather than rejected — Quantus headers carry a `zkTreeRoot` the
|
|
/// upstream Substrate shape does not, and future fields must not break ingest.
|
|
#[derive(Debug, Clone, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct Header {
|
|
/// Block number as `0x`-prefixed hex.
|
|
pub number: String,
|
|
/// Parent block hash.
|
|
pub parent_hash: String,
|
|
/// Consensus digest logs. The `pow_` PreRuntime log in here carries the
|
|
/// author's reward preimage.
|
|
pub digest: Digest,
|
|
}
|
|
|
|
/// A header's digest logs.
|
|
#[derive(Debug, Clone, Deserialize)]
|
|
pub struct Digest {
|
|
/// Hex-encoded logs, in order.
|
|
pub logs: Vec<String>,
|
|
}
|
|
|
|
impl Header {
|
|
/// The block number, decoded from its hex form.
|
|
pub fn height(&self) -> Option<u64> {
|
|
u64::from_str_radix(self.number.strip_prefix("0x").unwrap_or(&self.number), 16).ok()
|
|
}
|
|
}
|
|
|
|
/// `twox_128("System") ++ twox_128("Events")`, the storage key every FRAME
|
|
/// runtime keeps its per-block events under. Constant across runtimes because
|
|
/// it is a hash of the names, not of anything version-specific.
|
|
const SYSTEM_EVENTS_KEY: &str =
|
|
"0x26aa394eea5630e07c48ae0c9558cef780d41e5e16056765bc8461851072c9d7";
|
|
|
|
/// Which runtime produced a block.
|
|
#[derive(Debug, Clone, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct RuntimeVersion {
|
|
/// e.g. `quantus-runtime`.
|
|
pub spec_name: String,
|
|
/// Bumped on every runtime upgrade. The cache key for metadata.
|
|
pub spec_version: u32,
|
|
/// Bumped when the extrinsic format changes — 2 to 3 to 6 across
|
|
/// Heisenberg's upgrades, and every one of those would break a decoder
|
|
/// written against the previous format.
|
|
pub transaction_version: u32,
|
|
}
|
|
|
|
/// A block body, reduced to what the observer reads from it.
|
|
#[derive(Debug, Clone)]
|
|
pub struct BlockBody {
|
|
/// The block's extrinsics as the node returned them, `0x` hex, in order —
|
|
/// inherents included. Kept raw rather than decoded here: `blackbeard-data`
|
|
/// does I/O and `blackbeard-core` does decoding, and the runtime needed to
|
|
/// read these lives on the other side of that line.
|
|
pub extrinsics: Vec<String>,
|
|
/// The author's timestamp from the `Timestamp::set` inherent.
|
|
pub timestamp_ms: Option<u64>,
|
|
}
|
|
|
|
/// What `system_health` reports.
|
|
#[derive(Debug, Clone, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct Health {
|
|
/// True while the node is importing history rather than following the tip.
|
|
pub is_syncing: bool,
|
|
/// Connected peers.
|
|
pub peers: u32,
|
|
}
|
|
|
|
/// What `system_properties` reports.
|
|
#[derive(Debug, Clone, Default, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
pub struct ChainProperties {
|
|
/// Token ticker, e.g. `PLK`.
|
|
pub token_symbol: Option<String>,
|
|
/// Decimal places in the smallest unit.
|
|
pub token_decimals: Option<u8>,
|
|
/// SS58 address prefix.
|
|
pub ss58_format: Option<u16>,
|
|
}
|
|
|
|
/// The most keys `state_getKeysPaged` will return in one call.
|
|
///
|
|
/// Substrate's own limit, and it *rejects* rather than truncates: asking for
|
|
/// 1,600 is `count exceeds maximum value. value: 1600, max: 1000`. A caller
|
|
/// that turns that error into an empty page has produced a wrong answer that
|
|
/// looks exactly like a right one.
|
|
pub const MAX_KEYS_PER_PAGE: u32 = 1000;
|
|
|
|
/// 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,
|
|
endpoints: Arc<Vec<String>>,
|
|
/// Index into `endpoints` to try first.
|
|
current: Arc<AtomicUsize>,
|
|
/// What each endpoint was found to hold, parallel to `endpoints`.
|
|
depth: Arc<Vec<AtomicU8>>,
|
|
}
|
|
|
|
/// 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<String>, timeout: Duration) -> Result<Self, DataError> {
|
|
Self::with_endpoints(vec![url.into()], timeout)
|
|
}
|
|
|
|
/// A client over several endpoints for the same chain.
|
|
pub fn with_endpoints(endpoints: Vec<String>, timeout: Duration) -> Result<Self, DataError> {
|
|
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)
|
|
// The node is one host away and answers thousands of these per
|
|
// minute; without pooling every call would pay a fresh TCP
|
|
// 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)),
|
|
})
|
|
}
|
|
|
|
/// 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()
|
|
}
|
|
|
|
/// 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<Value, DataError> {
|
|
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<Value, DataError> {
|
|
let body = json!({"jsonrpc": "2.0", "id": 1, "method": method, "params": params});
|
|
let start = self.current.load(Ordering::Relaxed);
|
|
let mut last: Option<DataError> = 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<Value, DataError> {
|
|
let resp: Value = self
|
|
.http
|
|
.post(url)
|
|
.json(body)
|
|
.send()
|
|
.await?
|
|
.error_for_status()?
|
|
.json()
|
|
.await?;
|
|
if let Some(err) = resp.get("error") {
|
|
return Err(DataError::Rpc {
|
|
method: method.to_owned(),
|
|
message: err.to_string(),
|
|
});
|
|
}
|
|
resp.get("result")
|
|
.cloned()
|
|
.ok_or_else(|| DataError::Malformed {
|
|
method: method.to_owned(),
|
|
message: "response carried neither result nor error".into(),
|
|
})
|
|
}
|
|
|
|
/// `system_health`.
|
|
pub async fn health(&self) -> Result<Health, DataError> {
|
|
Ok(serde_json::from_value(
|
|
self.call("system_health", json!([])).await?,
|
|
)?)
|
|
}
|
|
|
|
/// `system_properties` — token symbol, decimals, SS58 prefix.
|
|
pub async fn properties(&self) -> Result<ChainProperties, DataError> {
|
|
Ok(serde_json::from_value(
|
|
self.call("system_properties", json!([])).await?,
|
|
)?)
|
|
}
|
|
|
|
/// The current best header, or the header at `hash`.
|
|
pub async fn header(&self, hash: Option<&str>) -> Result<Option<Header>, DataError> {
|
|
let params = match hash {
|
|
Some(h) => json!([h]),
|
|
None => json!([]),
|
|
};
|
|
let v = self.call("chain_getHeader", params).await?;
|
|
if v.is_null() {
|
|
return Ok(None);
|
|
}
|
|
Ok(Some(serde_json::from_value(v)?))
|
|
}
|
|
|
|
/// The block hash at `height`, if the node has that block.
|
|
pub async fn block_hash(&self, height: u64) -> Result<Option<String>, DataError> {
|
|
let v = self.call("chain_getBlockHash", json!([height])).await?;
|
|
Ok(v.as_str().map(str::to_owned))
|
|
}
|
|
|
|
/// The genesis hash. Discovered rather than configured — one fewer value to
|
|
/// get wrong when a chain launches or is respecced.
|
|
pub async fn genesis(&self) -> Result<Option<String>, DataError> {
|
|
self.block_hash(0).await
|
|
}
|
|
|
|
/// The author's timestamp for a block, in milliseconds.
|
|
///
|
|
/// `None` when the block's first extrinsic is not the timestamp inherent —
|
|
/// a shape we do not recognise, which costs this block its timing data and
|
|
/// nothing else.
|
|
pub async fn block_timestamp_ms(&self, hash: &str) -> Result<Option<u64>, DataError> {
|
|
Ok(self.body(hash).await?.and_then(|b| b.timestamp_ms))
|
|
}
|
|
|
|
/// Every storage key under a prefix, a page at a time.
|
|
///
|
|
/// The only way to ask a node "what is in this map": there is no list, just
|
|
/// keys derived from the things in it, so enumeration walks the prefix.
|
|
///
|
|
/// `start` is the last key of the previous page, exclusive. A page shorter
|
|
/// than `count` is the end.
|
|
///
|
|
/// `count` is clamped to [`MAX_KEYS_PER_PAGE`]. That ceiling is the node's,
|
|
/// not a caller's preference, so it is enforced here rather than in each
|
|
/// caller's constant — asking for more is an RPC *error*, and an error that
|
|
/// a caller turns into an empty page reads exactly like an empty map.
|
|
pub async fn storage_keys_paged(
|
|
&self,
|
|
prefix: &str,
|
|
count: u32,
|
|
start: Option<&str>,
|
|
at: Option<&str>,
|
|
) -> Result<Vec<String>, DataError> {
|
|
// `at` is what makes genesis readable: the endowed set exists only in
|
|
// block zero's state, and enumerating it at the tip returns whatever
|
|
// the map holds now instead. Needs an archive node.
|
|
let count = count.min(MAX_KEYS_PER_PAGE);
|
|
let params = match (start, at) {
|
|
(Some(s), Some(h)) => json!([prefix, count, s, h]),
|
|
(Some(s), None) => json!([prefix, count, s]),
|
|
(None, Some(h)) => json!([prefix, count, null, h]),
|
|
(None, None) => json!([prefix, count]),
|
|
};
|
|
// 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()
|
|
.filter_map(Value::as_str)
|
|
.map(str::to_owned)
|
|
.collect()
|
|
})
|
|
.unwrap_or_default())
|
|
}
|
|
|
|
/// How many keys live under a prefix.
|
|
///
|
|
/// Pages through them and counts, because Substrate offers no `count` and a
|
|
/// map's size is a real fact about a chain — how many accounts exist, how
|
|
/// many balances are locked. At the node's 1,000-key ceiling, mainnet's
|
|
/// 1,914 accounts is two round trips.
|
|
///
|
|
/// `stop_at` bounds the walk so a map nobody expected to be large cannot
|
|
/// turn a page render into a thousand requests; the count is returned with
|
|
/// a flag saying whether it is complete.
|
|
pub async fn count_keys(
|
|
&self,
|
|
prefix: &str,
|
|
stop_at: usize,
|
|
) -> Result<(usize, bool), DataError> {
|
|
let mut total = 0usize;
|
|
let mut start: Option<String> = None;
|
|
loop {
|
|
let page = self
|
|
.storage_keys_paged(prefix, MAX_KEYS_PER_PAGE, start.as_deref(), None)
|
|
.await?;
|
|
let got = page.len();
|
|
total += got;
|
|
if got < MAX_KEYS_PER_PAGE as usize {
|
|
return Ok((total, true));
|
|
}
|
|
if total >= stop_at {
|
|
return Ok((total, false));
|
|
}
|
|
start = page.last().cloned();
|
|
}
|
|
}
|
|
|
|
/// The runtime's own description of itself, as of `hash`.
|
|
///
|
|
/// The node executes `Metadata_metadata` against the runtime code in that
|
|
/// block's state, so this is the WASM answering — the only source that is
|
|
/// right across a runtime upgrade. Needs the state, which a pruning node
|
|
/// will not have for old blocks; the caller caches the result per
|
|
/// `spec_version` so that only matters once.
|
|
pub async fn metadata_at(&self, hash: Option<&str>) -> Result<Vec<u8>, DataError> {
|
|
let params = match hash {
|
|
Some(h) => json!([h]),
|
|
None => json!([]),
|
|
};
|
|
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(),
|
|
})?;
|
|
hex::decode(hex.strip_prefix("0x").unwrap_or(hex)).map_err(|e| DataError::Rpc {
|
|
method: "state_getMetadata".into(),
|
|
message: format!("metadata was not hex: {e}"),
|
|
})
|
|
}
|
|
|
|
/// Which runtime produced a block. `spec_version` changes only on an
|
|
/// upgrade, which is the signal to fetch metadata again.
|
|
pub async fn runtime_version(&self, hash: Option<&str>) -> Result<RuntimeVersion, DataError> {
|
|
let params = match hash {
|
|
Some(h) => json!([h]),
|
|
None => json!([]),
|
|
};
|
|
Ok(serde_json::from_value(
|
|
self.call("state_getRuntimeVersion", params).await?,
|
|
)?)
|
|
}
|
|
|
|
/// The raw `System::Events` blob for a block.
|
|
///
|
|
/// `Ok(None)` where the node has pruned the state it lives in — an answer,
|
|
/// not a failure, and the reason metadata is cached rather than re-fetched.
|
|
pub async fn events_at(&self, hash: &str) -> Result<Option<Vec<u8>>, DataError> {
|
|
self.storage(SYSTEM_EVENTS_KEY, Some(hash)).await
|
|
}
|
|
|
|
/// Read one storage value, by key.
|
|
///
|
|
/// `None` means the chain holds nothing there, which for a `Default` entry
|
|
/// means its default and for an `Optional` entry means nothing — a
|
|
/// distinction only the metadata knows, so it is the caller's to make.
|
|
///
|
|
/// At the tip when `hash` is `None`. Against an old block it needs state
|
|
/// the node may have pruned, which is an absent value rather than an error
|
|
/// and is why nothing here treats absence as authoritative on its own.
|
|
pub async fn storage(
|
|
&self,
|
|
key: &str,
|
|
hash: Option<&str>,
|
|
) -> Result<Option<Vec<u8>>, DataError> {
|
|
let params = match hash {
|
|
Some(h) => json!([key, h]),
|
|
None => json!([key]),
|
|
};
|
|
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);
|
|
};
|
|
Ok(hex::decode(hex.strip_prefix("0x").unwrap_or(hex)).ok())
|
|
}
|
|
|
|
/// A block's body, reduced to what is read from it.
|
|
///
|
|
/// One call for both facts. The ingest path wants only the timestamp and
|
|
/// the block page wants the extrinsic count too; fetching the same body
|
|
/// twice to answer them separately would double the load on a node that is
|
|
/// already mining.
|
|
///
|
|
/// `Ok(None)` for a block the node no longer holds — `blocks-pruning`
|
|
/// defaults to `archive-canonical`, so a losing fork's body goes once
|
|
/// finality passes it. That is an answer, not a failure.
|
|
pub async fn body(&self, hash: &str) -> Result<Option<BlockBody>, DataError> {
|
|
let v = self.call("chain_getBlock", json!([hash])).await?;
|
|
let Some(extrinsics) = v.pointer("/block/extrinsics").and_then(Value::as_array) else {
|
|
return Ok(None);
|
|
};
|
|
Ok(Some(BlockBody {
|
|
// The timestamp inherent is always first. A body whose first
|
|
// extrinsic does not decode as one costs this block its timing and
|
|
// nothing else.
|
|
timestamp_ms: extrinsics
|
|
.first()
|
|
.and_then(Value::as_str)
|
|
.and_then(blackbeard_core::scale::timestamp_inherent_ms),
|
|
extrinsics: extrinsics
|
|
.iter()
|
|
.filter_map(Value::as_str)
|
|
.map(str::to_owned)
|
|
.collect(),
|
|
}))
|
|
}
|
|
|
|
/// Current mining difficulty: expected hashes to win a block.
|
|
///
|
|
/// This is the difficulty the **next** block must meet, not the one the tip
|
|
/// met — see [`Self::difficulty_for_child_of`] for why, and use that
|
|
/// instead whenever the difficulty of a *specific* block is wanted.
|
|
pub async fn difficulty(&self) -> Result<primitive_types::U512, DataError> {
|
|
self.u512_runtime_call("QPoWApi_get_difficulty", None).await
|
|
}
|
|
|
|
/// The difficulty the child of `parent_hash` had to meet.
|
|
///
|
|
/// `pallet_qpow::on_finalize` retargets at the *end* of every block, so the
|
|
/// `CurrentDifficulty` in the state of block *h* is the difficulty for
|
|
/// *h+1*. Asking at the latest state and calling the answer "the difficulty
|
|
/// of the block we just saw" is therefore off by one block — small, because
|
|
/// the retarget is gradual, and invisible, because the number still looks
|
|
/// entirely plausible. To record the difficulty of block *h*, ask at its
|
|
/// parent, which is the same call the miner made when it built it.
|
|
///
|
|
/// Needs the state at `parent_hash`, so a pruning node answers only for the
|
|
/// recent past. The caller must treat an error as "not known" rather than
|
|
/// substituting the current value, which is how a gap fill would otherwise
|
|
/// stamp today's difficulty across a stretch of old heights.
|
|
pub async fn difficulty_for_child_of(
|
|
&self,
|
|
parent_hash: &str,
|
|
) -> Result<primitive_types::U512, DataError> {
|
|
self.u512_runtime_call("QPoWApi_get_difficulty", Some(parent_hash))
|
|
.await
|
|
}
|
|
|
|
/// The ceiling difficulty can reach.
|
|
pub async fn max_difficulty(&self) -> Result<primitive_types::U512, DataError> {
|
|
self.u512_runtime_call("QPoWApi_get_max_difficulty", None)
|
|
.await
|
|
}
|
|
|
|
async fn u512_runtime_call(
|
|
&self,
|
|
api: &str,
|
|
at: Option<&str>,
|
|
) -> Result<primitive_types::U512, DataError> {
|
|
// `state_call` takes the block hash third. Omitting it entirely and
|
|
// passing an explicit null are the same thing to the node, but building
|
|
// the shorter params list keeps the common call identical to what it
|
|
// has always sent.
|
|
let params = match at {
|
|
Some(hash) => json!([api, "0x", hash]),
|
|
None => json!([api, "0x"]),
|
|
};
|
|
// 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(),
|
|
})?;
|
|
blackbeard_core::digest::u512_le(hex).ok_or_else(|| DataError::Rpc {
|
|
method: api.to_owned(),
|
|
message: format!("`{hex}` is not a little-endian U512"),
|
|
})
|
|
}
|
|
}
|
|
|
|
/// How long to wait between reconnection attempts on the head subscription.
|
|
const RECONNECT_DELAY: Duration = Duration::from_secs(5);
|
|
|
|
/// A head, and **when this observer saw it**.
|
|
///
|
|
/// The time is stamped where the frame is read off the socket, not where the
|
|
/// head is finally processed. Those are the same only while the ingest loop
|
|
/// keeps up, and it does not: every head costs two RPC round trips, which on a
|
|
/// remote endpoint is hundreds of milliseconds, so the 64-deep channel backs up
|
|
/// and drains in a burst. Stamping at processing time gave a batch of heads
|
|
/// near-identical observation times while their heights marched on, and
|
|
/// `measured_interval` read that as a chain producing blocks twenty times
|
|
/// faster than it does — 0.052 s/block on Heisenberg.
|
|
///
|
|
/// `observed_at` means "when this observer saw it". This is that moment.
|
|
#[derive(Debug, Clone)]
|
|
pub struct SeenHead {
|
|
/// The header the node announced.
|
|
pub header: Header,
|
|
/// When the frame carrying it arrived.
|
|
pub seen_at: chrono::DateTime<chrono::Utc>,
|
|
}
|
|
|
|
/// Follow a node's new heads over WebSocket, forwarding each to `sink`.
|
|
///
|
|
/// Runs until the channel closes. Reconnects on its own: a node restart, a
|
|
/// dropped mesh link or a proxy timeout are ordinary operating conditions here,
|
|
/// not reasons to stop watching a chain — and a chain configured before it
|
|
/// launches will spend its first weeks failing to connect on every attempt,
|
|
/// which must stay quiet rather than filling the journal.
|
|
///
|
|
/// **Heads are not a complete block record.** `chain_subscribeNewHeads` reports
|
|
/// the *best* head and skips intermediate blocks when several import at once,
|
|
/// so the caller must fill gaps against `chain_getBlockHash`. It is a liveness
|
|
/// signal, not a ledger.
|
|
pub async fn subscribe_new_heads(
|
|
ws_urls: Vec<String>,
|
|
sink: mpsc::Sender<SeenHead>,
|
|
) -> 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(());
|
|
}
|
|
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;
|
|
}
|
|
Err(e) => {
|
|
// First failure at warn, the rest at debug. A chain that has not
|
|
// launched fails every five seconds forever; logging each at warn
|
|
// would bury everything else in the journal.
|
|
if backoff_logged {
|
|
tracing::debug!(url = %ws_url, error = %e, "head subscription still down");
|
|
} else {
|
|
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;
|
|
}
|
|
}
|
|
|
|
async fn follow_once(ws_url: &str, sink: &mpsc::Sender<SeenHead>) -> Result<(), DataError> {
|
|
let (mut socket, _) = tokio_tungstenite::connect_async(ws_url).await?;
|
|
socket
|
|
.send(tokio_tungstenite::tungstenite::Message::Text(
|
|
json!({"jsonrpc": "2.0", "id": 1, "method": "chain_subscribeNewHeads", "params": []})
|
|
.to_string(),
|
|
))
|
|
.await?;
|
|
|
|
while let Some(msg) = socket.next().await {
|
|
let msg = msg?;
|
|
// Text and binary both accepted. Substrate sends text, but the
|
|
// telemetry feed on this same fleet sends binary (see
|
|
// `telemetry::follow_once`), and a client that silently drops the wrong
|
|
// frame type looks perfectly healthy while receiving nothing.
|
|
let payload = match msg {
|
|
tokio_tungstenite::tungstenite::Message::Text(t) => t.as_bytes().to_vec(),
|
|
tokio_tungstenite::tungstenite::Message::Binary(b) => b.to_vec(),
|
|
// The library answers pings itself; close ends the loop and the
|
|
// caller reconnects.
|
|
tokio_tungstenite::tungstenite::Message::Close(_) => break,
|
|
_ => continue,
|
|
};
|
|
let v: Value = match serde_json::from_slice(&payload) {
|
|
Ok(v) => v,
|
|
Err(e) => {
|
|
tracing::debug!(error = %e, "unparsable frame on the head subscription");
|
|
continue;
|
|
}
|
|
};
|
|
// The subscription confirmation carries `result` as a bare id; only
|
|
// notifications carry `params.result`.
|
|
let Some(result) = v.pointer("/params/result") else {
|
|
continue;
|
|
};
|
|
match serde_json::from_value::<Header>(result.clone()) {
|
|
Ok(header) => {
|
|
let seen = SeenHead {
|
|
header,
|
|
seen_at: chrono::Utc::now(),
|
|
};
|
|
if sink.send(seen).await.is_err() {
|
|
return Ok(());
|
|
}
|
|
}
|
|
Err(e) => tracing::warn!(error = %e, "head notification did not decode as a header"),
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
#[cfg(test)]
|
|
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!({
|
|
"number": "0xfe2a8",
|
|
"parentHash": "0x87f4",
|
|
"stateRoot": "0x3398",
|
|
"zkTreeRoot": "0xd382",
|
|
"digest": {"logs": ["0x06706f775f80aa"]}
|
|
}))
|
|
.expect("a Quantus header carries fields upstream Substrate does not; ignore them");
|
|
assert_eq!(h.height(), Some(1_041_064));
|
|
assert_eq!(h.digest.logs.len(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn a_header_missing_its_digest_is_a_decode_error_not_a_panic() {
|
|
let r = serde_json::from_value::<Header>(json!({"number": "0x1", "parentHash": "0x0"}));
|
|
assert!(r.is_err());
|
|
}
|
|
}
|