From 1608535fac705a9806ea06568671438c39efba13 Mon Sep 17 00:00:00 2001 From: emilbob Date: Sat, 25 Jul 2026 16:52:13 +0200 Subject: [PATCH] Classify failures, report block height, bound whole steps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four gaps, one of which contradicted a rule this project had already adopted. 1. Every failure was `ok: false` plus a prose string, so a monitor that wants to page differently for "unreachable" and "too few peers" had to regex it — exactly what #16 avoided for the genesis hash and then didn't do for the failure itself. Errors are now a ProbeError carrying a Failure kind, surfaced as a stable `failure` field: config, connect, timeout, transport, protocol, rpc_error, genesis_mismatch, requirement_unmet. `error` stays for humans and is free to be reworded; `failure` is the part safe to alert on. 2. The report never carried the node's height, so it could not answer "is this node stuck?" — the most common question asked of a node. chain_getHeader joins the pipelined query step and lands as `best_block`. No staleness check: a Substrate header carries no timestamp, so the honest thing is to report the number and let the caller compare across runs or against another node. 3. Timeouts bounded each frame, not each operation. The read loops skip frames they don't recognise, so a node emitting one unrelated frame every 9s would keep a 10s probe alive forever — every wait short, the total unbounded. A Deadline is now set once per step and every read inside draws down the same clock; --follow gets a fresh budget per header, which is the one place a long wait is legitimate. 4. The timeouts were compiled in. --connect-timeout, --rpc-timeout and --head-timeout expose them, rejecting 0 since an instant-fail probe is never what someone means. Verified against wss://rpc.polkadot.io: healthy run reports best_block 32264078 and no failure key; a genesis mismatch, an unmeetable --require-peers, a refused connection, a bad --genesis-hash and a 1s connect timeout each report their own kind. Tests 24 -> 29, including a chattering-node test that would hang on the old per-frame bound. Co-Authored-By: Claude Opus 5 --- README.md | 48 +++++-- src/error.rs | 115 +++++++++++++++++ src/main.rs | 77 ++++++++++-- src/probe.rs | 296 ++++++++++++++++++++++++++++++-------------- src/report.rs | 90 +++++++++++--- src/rpc.rs | 102 +++++++++++---- src/test_support.rs | 1 + 7 files changed, 575 insertions(+), 154 deletions(-) create mode 100644 src/error.rs diff --git a/README.md b/README.md index 4f2b6a0..11391ca 100644 --- a/README.md +++ b/README.md @@ -21,7 +21,7 @@ INFO Node information queried! 1. **Connects** to the node's JSON-RPC endpoint over WebSocket (`ws://` or `wss://`). 2. **Checks chain identity** by asking for the hash of block 0 via `chain_getBlockHash` and comparing it to the `--genesis-hash` you supplied. A node on a different chain is rejected and the client exits non-zero without querying anything further. -3. **Queries node information** — `system_name`, `system_chain`, `system_version` and `system_health` — matching each response to its request by JSON-RPC id, since nodes are free to answer out of order (and do). All four go out before any reply is read, so the step costs one round trip rather than four. +3. **Queries node information** — `system_name`, `system_chain`, `system_version`, `system_health` and `chain_getHeader` — matching each response to its request by JSON-RPC id, since nodes are free to answer out of order (and do). All five go out before any reply is read, so the step costs one round trip rather than five. 4. **Follows new blocks**, optionally — with `--follow N` it subscribes via `chain_subscribeNewHeads`, reports headers as the node pushes them, then unsubscribes. 5. **Reports**, either as logs through `env_logger` (`RUST_LOG=debug` shows the full exchange) or as a single JSON object with `--json`. @@ -36,23 +36,51 @@ $ substrate-node-probe --node-address wss://rpc.polkadot.io --follow 2 --json 2> "ok": true, "genesis_hash": "0x91b171bb158e2d3848fa23a9f1c25182fb8e20313b2c1eb49219da7a70ce90c3", "genesis_verified": true, - "rpc_latency_ms": 63, + "rpc_latency_ms": 58, "name": "Parity Polkadot", "chain": "Polkadot", "version": "1.24.0-660acefe665", - "peers": 65, + "peers": 69, "is_syncing": false, "should_have_peers": true, + "best_block": 32264078, "heads_followed": 2 } ``` -Two properties worth relying on: +Three properties worth relying on: -- **A failed run still prints a report** — `"ok": false` with an `error`, plus whatever was gathered before the failure. A probe that goes silent exactly when the node breaks is no use to whatever is parsing it. A genesis mismatch, for instance, still reports the hash the node actually serves, so you needn't parse it back out of the error string. +- **A failed run still prints a report** — `"ok": false` with a `failure` and an `error`, plus whatever was gathered before the failure. A probe that goes silent exactly when the node breaks is no use to whatever is parsing it. A genesis mismatch, for instance, still reports the hash the node actually serves. - **Fields the node did not answer are omitted, not `null`** — `"peers": 0` means an isolated node; a missing `peers` means the node never said. `system_health` is not exposed everywhere, and a refusal costs only its own fields rather than the whole report. +- **Nothing requires parsing prose.** The `error` string is for humans and is free to be reworded; `failure` is the part that is safe to alert on. -`genesis_verified` is `true` only when `--genesis-hash` was supplied *and* matched; without the flag the hash is reported but nothing about it is proven. `should_have_peers` is what makes `peers: 0` interpretable — it is `false` on a dev chain running alone. +`genesis_verified` is `true` only when `--genesis-hash` was supplied *and* matched; without the flag the hash is reported but nothing about it is proven. `should_have_peers` is what makes `peers: 0` interpretable — it is `false` on a dev chain running alone. `best_block` is the node's current height, which is how you tell a stuck node from a healthy one; the probe reports it but cannot judge staleness on its own, because a Substrate header carries no timestamp — compare it across runs, or against a second node. + +### Failure kinds + +On failure, `failure` carries one of these. They are a stable contract; the prose in `error` is not. + +| `failure` | Meaning | +| --- | --- | +| `config` | The command line asked for something impossible. The node is blameless. | +| `connect` | The connection was never established. | +| `timeout` | A wait expired. The node may be up but is not answering in time. | +| `transport` | The connection failed or closed part-way through. | +| `protocol` | The node answered with something unusable — bad JSON, or a result of the wrong shape. | +| `rpc_error` | The node returned a JSON-RPC error for a call the probe needs. | +| `genesis_mismatch` | The node is serving a different chain than you required. | +| `requirement_unmet` | The node is reachable and correct, but failed a `--require-*` bar. | + +The distinction that motivates this: `connect` and `requirement_unmet` are both "the probe exited 1", but one means your monitoring cannot see the node and the other means the node is up and telling you it is unhealthy. Those usually want different alerts. + +```bash +case "$(substrate-node-probe --node-address "$RPC" --require-peers 1 --json 2>/dev/null | jq -r '.failure // "ok"')" in + ok) page=none ;; + requirement_unmet) page=degraded ;; + connect|timeout) page=unreachable ;; + *) page=investigate ;; +esac +``` ### Why WebSocket @@ -80,8 +108,9 @@ The `HandshakeMessage` struct in `src/scale.rs` is a worked example of SCALE enc | --- | --- | | `src/main.rs` | The CLI flags, and the order the steps run in. | | `src/rpc.rs` | The transport: opening the WebSocket, request ids, timeouts, reading frames. Knows nothing about chains. | -| `src/probe.rs` | The JSON-RPC calls — genesis hash, identity and health, following heads. | +| `src/probe.rs` | The JSON-RPC calls — genesis hash, identity, health and height, following heads. | | `src/report.rs` | The findings, their JSON shape, and the `--require-*` checks. | +| `src/error.rs` | The failure taxonomy. | | `src/scale.rs` | The SCALE codec example. | ## Requirements @@ -103,6 +132,9 @@ cargo run -- [--node-address ] [--genesis-hash ] | `--require-peers ` | *(none)* | Fail unless the node reports at least `N` connected peers. | | `--require-synced` | *(off)* | Fail if the node is still syncing. | | `--json` | *(off)* | Print the findings to stdout as one JSON object. Logs stay on stderr. | +| `--connect-timeout ` | `10` | Seconds to allow for the connection. | +| `--rpc-timeout ` | `10` | Seconds to allow the node to answer a request. | +| `--head-timeout ` | `120` | Seconds to allow for each pushed header under `--follow`. Raise it for a slow parachain. | Exit code is `0` on success and `1` on any failure, so it works as a health check directly: @@ -179,4 +211,4 @@ It is excluded from `cargo test` and never runs on a pull request, because it de ## Notes - TLS uses `rustls` with the `ring` provider, so the build needs no OpenSSL system libraries. -- Both the connect and per-response paths are bounded by a 10-second timeout; a node that accepts the socket and goes silent produces an error rather than a hang. Waiting for a pushed block header uses a separate, much longer bound (120s), since that waits on the chain's block time rather than on the node being responsive. +- Every wait is bounded, so a node that accepts the socket and goes silent produces an error rather than a hang. The bounds apply to a **step as a whole**, not to each frame within it: the read loops skip frames they do not recognise, and a per-frame clock would let a node emitting one unrelated frame just inside the limit keep the probe waiting forever — every individual wait short, the total unbounded. Waiting for a pushed block header gets a fresh budget per header, since that waits on the chain's block time rather than on the node being responsive. diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 0000000..7542e49 --- /dev/null +++ b/src/error.rs @@ -0,0 +1,115 @@ +//! What went wrong, in a form a machine can branch on. +//! +//! The probe's whole output contract is that a consumer should never have to +//! parse prose to act — see the `failure` field on +//! [`ProbeReport`](crate::report::ProbeReport). A `Box` carrying a +//! sentence cannot support that, so every fallible step returns a +//! [`ProbeError`] whose [`Failure`] says which *kind* of thing went wrong while +//! the message stays free to be as specific as it likes. + +use serde::Serialize; +use std::fmt; + +/// The class of failure, stable enough to alert on. +/// +/// A monitor pages differently for "I cannot reach this node" than for "this +/// node is up but has no peers", and those two must be distinguishable without +/// matching on error text that is free to be reworded at any time. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum Failure { + /// The command line asked for something impossible; the node is blameless. + Config, + /// The connection was never established. + Connect, + /// A wait expired. The node may be up but is not answering in time. + Timeout, + /// The connection failed or closed part-way through. + Transport, + /// The node answered, but with something unusable — bad JSON, or a result + /// of the wrong shape. + Protocol, + /// The node returned a JSON-RPC error for a call the probe needs. + RpcError, + /// The node is serving a different chain than the caller required. + GenesisMismatch, + /// The node is reachable and correct, but failed a `--require-*` bar. + RequirementUnmet, +} + +/// A failure, classified. +#[derive(Debug)] +pub(crate) struct ProbeError { + kind: Failure, + message: String, +} + +impl ProbeError { + pub(crate) fn new(kind: Failure, message: impl Into) -> Self { + ProbeError { + kind, + message: message.into(), + } + } + + pub(crate) fn config(message: impl Into) -> Self { + Self::new(Failure::Config, message) + } + + pub(crate) fn connect(message: impl Into) -> Self { + Self::new(Failure::Connect, message) + } + + pub(crate) fn timeout(message: impl Into) -> Self { + Self::new(Failure::Timeout, message) + } + + pub(crate) fn transport(message: impl Into) -> Self { + Self::new(Failure::Transport, message) + } + + pub(crate) fn protocol(message: impl Into) -> Self { + Self::new(Failure::Protocol, message) + } + + pub(crate) fn rpc(message: impl Into) -> Self { + Self::new(Failure::RpcError, message) + } + + pub(crate) fn kind(&self) -> Failure { + self.kind + } + + /// Adds detail without changing the classification. + /// + /// Used where an inner step already knows *what* went wrong and the outer + /// step knows *how far it had got* — a timeout is still a timeout when you + /// note it happened on the second of three headers. + pub(crate) fn context(mut self, extra: impl AsRef) -> Self { + self.message = format!("{} {}", self.message, extra.as_ref()); + self + } +} + +impl fmt::Display for ProbeError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.message) + } +} + +impl std::error::Error for ProbeError {} + +/// A malformed frame is the node's fault, not the transport's. +impl From for ProbeError { + fn from(e: serde_json::Error) -> Self { + ProbeError::protocol(format!("node sent unparseable JSON: {e}")) + } +} + +/// A WebSocket error mid-run means the connection broke under us; failures to +/// *establish* one are classified at the call site in [`crate::rpc::connect`]. +impl From for ProbeError { + fn from(e: tokio_tungstenite::tungstenite::Error) -> Self { + ProbeError::transport(format!("websocket error: {e}")) + } +} diff --git a/src/main.rs b/src/main.rs index f2f4730..7f50647 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,6 +5,7 @@ //! themselves live in [`probe`], the connection in [`rpc`], and the findings in //! [`report`]. +mod error; mod probe; mod report; mod rpc; @@ -15,12 +16,14 @@ mod test_support; use clap::Parser; use env_logger::Env; use log::{error, info}; +use std::time::Duration; +use error::ProbeError; use probe::{ fetch_genesis_hash, follow_new_heads, parse_genesis_hash, query_node_info, verify_genesis_hash, }; use report::{check_requirements, ProbeReport}; -use rpc::Timeouts; +use rpc::{Timeouts, CONNECT_TIMEOUT, RPC_TIMEOUT, SUBSCRIPTION_TIMEOUT}; /// Connect to a Substrate node, verify which chain it serves, and query its /// identity over JSON-RPC. @@ -58,6 +61,32 @@ struct Opt { /// straight into `jq`. #[arg(long)] json: bool, + + /// Seconds to allow for the connection to be established. + #[arg(long, value_name = "SECS", default_value_t = CONNECT_TIMEOUT.as_secs(), value_parser = clap::value_parser!(u64).range(1..))] + connect_timeout: u64, + + /// Seconds to allow the node to answer a request. Bounds each step as a + /// whole, so a node dribbling unrelated frames cannot extend it. + #[arg(long, value_name = "SECS", default_value_t = RPC_TIMEOUT.as_secs(), value_parser = clap::value_parser!(u64).range(1..))] + rpc_timeout: u64, + + /// Seconds to allow for each pushed block header under `--follow`. Far + /// longer than the RPC wait by default: this waits on the chain's block + /// time, not on the node being responsive. Raise it for a slow parachain. + #[arg(long, value_name = "SECS", default_value_t = SUBSCRIPTION_TIMEOUT.as_secs(), value_parser = clap::value_parser!(u64).range(1..))] + head_timeout: u64, +} + +impl Opt { + /// The network waits this invocation asked for. + fn timeouts(&self) -> Timeouts { + Timeouts { + connect: Duration::from_secs(self.connect_timeout), + rpc: Duration::from_secs(self.rpc_timeout), + head: Duration::from_secs(self.head_timeout), + } + } } /// Connects to the node, verifies its chain identity and queries node info. @@ -75,18 +104,15 @@ struct Opt { /// # Returns /// /// A Result indicating the success or failure of the run. -async fn run( - opt: &Opt, - timeouts: Timeouts, - report: &mut ProbeReport, -) -> Result<(), Box> { +async fn run(opt: &Opt, timeouts: Timeouts, report: &mut ProbeReport) -> Result<(), ProbeError> { let expected_genesis = opt .genesis_hash .as_deref() .map(parse_genesis_hash) - .transpose()?; + .transpose() + .map_err(|e| ProbeError::config(format!("--genesis-hash is unusable: {e}")))?; - let mut ws_stream = rpc::connect(&opt.node_address).await?; + let mut ws_stream = rpc::connect(&opt.node_address, timeouts.connect).await?; // Recorded before the comparison, so a mismatch is reported alongside the // hash that caused it rather than only in the error text. @@ -102,6 +128,7 @@ async fn run( report.peers = info.peers; report.is_syncing = info.is_syncing; report.should_have_peers = info.should_have_peers; + report.best_block = info.best_block; info!("Node information queried!"); // Judged before `--follow`, which can block for as long as the chain takes @@ -138,10 +165,13 @@ async fn main() { ..Default::default() }; - let result = run(&opt, Timeouts::default(), &mut report).await; + let result = run(&opt, opt.timeouts(), &mut report).await; match &result { Ok(()) => report.ok = true, - Err(e) => report.error = Some(e.to_string()), + Err(e) => { + report.failure = Some(e.kind()); + report.error = Some(e.to_string()); + } } // Printed on failure too: a probe that emits nothing precisely when the node @@ -192,13 +222,16 @@ mod tests { require_peers: Some(1), require_synced: true, json: false, + connect_timeout: CONNECT_TIMEOUT.as_secs(), + rpc_timeout: RPC_TIMEOUT.as_secs(), + head_timeout: SUBSCRIPTION_TIMEOUT.as_secs(), }; let mut report = ProbeReport { endpoint: opt.node_address.clone(), ..Default::default() }; - let result = run(&opt, Timeouts::default(), &mut report).await; + let result = run(&opt, opt.timeouts(), &mut report).await; assert!( result.is_ok(), @@ -216,6 +249,28 @@ mod tests { "system_health reported no peers: {report:#?}" ); assert_eq!(report.is_syncing, Some(false)); + assert!( + report.best_block.unwrap_or(0) > 0, + "chain_getHeader gave no height: {report:#?}" + ); assert_eq!(report.heads_followed, Some(1), "no header was pushed"); } + + /// The CLI is a contract too. A zero timeout would make every run fail + /// instantly, which is never what someone means by asking for one. + #[test] + fn timeout_flags_reject_zero() { + assert!( + Opt::try_parse_from(["substrate-node-probe", "--rpc-timeout", "0"]).is_err(), + "a zero RPC timeout must not be accepted" + ); + let opt = Opt::try_parse_from(["substrate-node-probe", "--rpc-timeout", "3"]) + .expect("a positive timeout is valid"); + assert_eq!(opt.timeouts().rpc, Duration::from_secs(3)); + assert_eq!( + opt.timeouts().head, + SUBSCRIPTION_TIMEOUT, + "unset flags keep their defaults" + ); + } } diff --git a/src/probe.rs b/src/probe.rs index 9533ad4..3f4d2fd 100644 --- a/src/probe.rs +++ b/src/probe.rs @@ -11,9 +11,10 @@ use std::collections::HashSet; use std::time::{Duration, Instant}; use tokio_tungstenite::tungstenite::protocol::Message; +use crate::error::{Failure, ProbeError}; use crate::rpc::{ - next_text_frame, NodeStream, Timeouts, ID_CHAIN, ID_GENESIS, ID_HEALTH, ID_NAME, ID_SUBSCRIBE, - ID_UNSUBSCRIBE, ID_VERSION, + next_text_frame, Deadline, NodeStream, Timeouts, ID_CHAIN, ID_GENESIS, ID_HEADER, ID_HEALTH, + ID_NAME, ID_SUBSCRIBE, ID_UNSUBSCRIBE, ID_VERSION, }; /// What asking the node for block 0 produced. @@ -36,6 +37,8 @@ pub(crate) struct NodeInfo { pub(crate) peers: Option, pub(crate) is_syncing: Option, pub(crate) should_have_peers: Option, + /// The number of the node's best block, from `chain_getHeader`. + pub(crate) best_block: Option, } /// Parses a hex-encoded 32-byte genesis hash. @@ -46,20 +49,27 @@ pub(crate) struct NodeInfo { /// /// # Returns /// -/// The decoded hash, or an error describing why the input was rejected. -pub(crate) fn parse_genesis_hash(hex_str: &str) -> Result<[u8; 32], Box> { +/// The decoded hash, or a message explaining why the input was rejected. The +/// caller classifies it, because the same bad value means different things +/// coming from the command line and coming from the node. +pub(crate) fn parse_genesis_hash(hex_str: &str) -> Result<[u8; 32], String> { let hex_str = hex_str.strip_prefix("0x").unwrap_or(hex_str); - let bytes = - hex::decode(hex_str).map_err(|e| format!("genesis hash is not valid hex: {}", e))?; + let bytes = hex::decode(hex_str).map_err(|e| format!("genesis hash is not valid hex: {e}"))?; <[u8; 32]>::try_from(bytes.as_slice()).map_err(|_| { format!( "genesis hash must be 32 bytes (64 hex chars), got {} bytes", bytes.len() ) - .into() }) } +/// Reads a `0x`-prefixed hex block number out of a header. +fn parse_block_number(header: &serde_json::Value) -> Option { + header["number"] + .as_str() + .and_then(|n| u64::from_str_radix(n.trim_start_matches("0x"), 16).ok()) +} + /// Asks the node which chain it is serving, by requesting the hash of block 0. /// /// Kept separate from [`verify_genesis_hash`] so that a mismatch can still be @@ -70,6 +80,7 @@ pub(crate) fn parse_genesis_hash(hex_str: &str) -> Result<[u8; 32], Box Result<[u8; 32], Box Result> { +) -> Result { let request = json!({ "jsonrpc": "2.0", "method": "chain_getBlockHash", @@ -91,8 +102,9 @@ pub(crate) async fn fetch_genesis_hash( let sent_at = Instant::now(); ws_stream.send(Message::Text(request.to_string())).await?; + let deadline = Deadline::after(timeouts.rpc); let response: serde_json::Value = loop { - let text = next_text_frame(ws_stream, timeouts.rpc).await?; + let text = next_text_frame(ws_stream, deadline).await?; let value: serde_json::Value = serde_json::from_str(&text)?; if value["id"].as_u64() == Some(ID_GENESIS) { break value; @@ -102,14 +114,17 @@ pub(crate) async fn fetch_genesis_hash( let latency = sent_at.elapsed(); if let Some(error) = response.get("error") { - return Err(format!("node rejected chain_getBlockHash: {error}").into()); + return Err(ProbeError::rpc(format!( + "node rejected chain_getBlockHash: {error}" + ))); } - let reported = response["result"] - .as_str() - .ok_or("chain_getBlockHash returned no block hash — is block 0 available?")?; - let reported = parse_genesis_hash(reported) - .map_err(|e| format!("node reported an unusable genesis hash: {e}"))?; + let reported = response["result"].as_str().ok_or_else(|| { + ProbeError::protocol("chain_getBlockHash returned no block hash — is block 0 available?") + })?; + let reported = parse_genesis_hash(reported).map_err(|e| { + ProbeError::protocol(format!("node reported an unusable genesis hash: {e}")) + })?; Ok(GenesisInfo { hash: reported, @@ -137,14 +152,16 @@ pub(crate) async fn fetch_genesis_hash( pub(crate) fn verify_genesis_hash( reported: [u8; 32], expected: Option<&[u8; 32]>, -) -> Result> { +) -> Result { match expected { - Some(expected) if reported != *expected => Err(format!( - "genesis hash mismatch — expected {}, node reports {}", - hex::encode(expected), - hex::encode(reported) - ) - .into()), + Some(expected) if reported != *expected => Err(ProbeError::new( + Failure::GenesisMismatch, + format!( + "genesis hash mismatch — expected {}, node reports {}", + hex::encode(expected), + hex::encode(reported) + ), + )), Some(_) => { info!("Genesis hash verified: 0x{}", hex::encode(reported)); Ok(true) @@ -159,14 +176,15 @@ pub(crate) fn verify_genesis_hash( } } -/// Queries node identity and health from the Substrate node. +/// Queries node identity, health and current height from the Substrate node. /// -/// All four calls are sent before any reply is read, so the node answers them -/// in parallel and the whole step costs one round trip rather than four. +/// All five calls are sent before any reply is read, so the node answers them +/// in parallel and the whole step costs one round trip rather than five. /// /// # Arguments /// /// * `ws_stream` - The connection to the node. +/// * `timeouts` - The network waits to apply. /// /// # Returns /// @@ -176,7 +194,7 @@ pub(crate) fn verify_genesis_hash( pub(crate) async fn query_node_info( ws_stream: &mut NodeStream, timeouts: Timeouts, -) -> Result> { +) -> Result { let requests = vec![ json!({ "jsonrpc": "2.0", @@ -205,6 +223,14 @@ pub(crate) async fn query_node_info( "params": [], "id": ID_HEALTH, }), + // With no params this returns the *best* header, which is how the probe + // answers "is this node stuck?" without opening a subscription. + json!({ + "jsonrpc": "2.0", + "method": "chain_getHeader", + "params": [], + "id": ID_HEADER, + }), ]; // Track the ids we are still waiting on, so that a response — successful or @@ -221,17 +247,19 @@ pub(crate) async fn query_node_info( } let mut info = NodeInfo::default(); + // One budget for the whole step. The calls are pipelined, so they should all + // land together; a node that dribbles unrelated frames must not be able to + // hold the probe open by resetting a per-frame clock. + let deadline = Deadline::after(timeouts.rpc); while !pending.is_empty() { - let text = next_text_frame(ws_stream, timeouts.rpc) - .await - .map_err(|e| { - format!( - "{e} ({} request(s) unanswered: {:?})", - pending.len(), - pending - ) - })?; + let text = next_text_frame(ws_stream, deadline).await.map_err(|e| { + e.context(format!( + "({} request(s) unanswered: {:?})", + pending.len(), + pending + )) + })?; let response: serde_json::Value = serde_json::from_str(&text)?; match response["id"].as_u64() { @@ -280,6 +308,13 @@ fn record_response(info: &mut NodeInfo, id: u64, result: &serde_json::Value) { info.is_syncing ); } + ID_HEADER => { + info.best_block = parse_block_number(result); + match info.best_block { + Some(number) => info!("Best block: #{number}"), + None => error!("chain_getHeader returned an unreadable block number: {result}"), + } + } _ => error!("No handler for request id {id}"), } } @@ -297,6 +332,7 @@ fn record_response(info: &mut NodeInfo, id: u64, result: &serde_json::Value) { /// /// * `ws_stream` - The connection to the node. /// * `count` - How many headers to observe before unsubscribing. +/// * `timeouts` - The network waits to apply. /// /// # Returns /// @@ -306,7 +342,7 @@ pub(crate) async fn follow_new_heads( ws_stream: &mut NodeStream, count: u64, timeouts: Timeouts, -) -> Result> { +) -> Result { let request = json!({ "jsonrpc": "2.0", "method": "chain_subscribeNewHeads", @@ -319,48 +355,55 @@ pub(crate) async fn follow_new_heads( // The subscription id arrives as the reply to the subscribe call; every // later notification carries it, which is how concurrent subscriptions on // one connection are told apart. + let deadline = Deadline::after(timeouts.rpc); let subscription_id = loop { - let text = next_text_frame(ws_stream, timeouts.rpc).await?; + let text = next_text_frame(ws_stream, deadline).await?; let value: serde_json::Value = serde_json::from_str(&text)?; if value["id"].as_u64() != Some(ID_SUBSCRIBE) { continue; } if let Some(error) = value.get("error") { - return Err(format!("node rejected chain_subscribeNewHeads: {error}").into()); + return Err(ProbeError::rpc(format!( + "node rejected chain_subscribeNewHeads: {error}" + ))); } break value["result"] .as_str() - .ok_or("subscription returned no id")? + .ok_or_else(|| ProbeError::protocol("subscription returned no id"))? .to_string(); }; info!("Subscribed with id {subscription_id}"); let mut seen = 0; while seen < count { - // A block is not a reply, so this waits on the chain's block time - // rather than on the node's responsiveness. - let text = next_text_frame(ws_stream, timeouts.head) - .await - .map_err(|e| format!("{e} (saw {seen} of {count} headers)"))?; - let value: serde_json::Value = serde_json::from_str(&text)?; + // A fresh budget per header: each block legitimately takes up to the + // chain's block time, but no single one may be stretched indefinitely by + // unrelated frames arriving in between. + let deadline = Deadline::after(timeouts.head); - if value["method"].as_str() != Some("chain_newHead") - || value["params"]["subscription"].as_str() != Some(&subscription_id) - { - continue; - } + loop { + let text = next_text_frame(ws_stream, deadline) + .await + .map_err(|e| e.context(format!("(saw {seen} of {count} headers)")))?; + let value: serde_json::Value = serde_json::from_str(&text)?; - let header = &value["params"]["result"]; - let number = header["number"] - .as_str() - .and_then(|n| u64::from_str_radix(n.trim_start_matches("0x"), 16).ok()); - match number { - Some(number) => info!( - "New head #{number} parent={}", - header["parentHash"].as_str().unwrap_or("?") - ), - None => error!("New head with unreadable block number: {header}"), + if value["method"].as_str() != Some("chain_newHead") + || value["params"]["subscription"].as_str() != Some(&subscription_id) + { + continue; + } + + let header = &value["params"]["result"]; + match parse_block_number(header) { + Some(number) => info!( + "New head #{number} parent={}", + header["parentHash"].as_str().unwrap_or("?") + ), + None => error!("New head with unreadable block number: {header}"), + } + break; } + seen += 1; } @@ -390,7 +433,7 @@ mod tests { ws_stream: &mut NodeStream, expected: Option<&[u8; 32]>, timeouts: Timeouts, - ) -> Result> { + ) -> Result { let genesis = fetch_genesis_hash(ws_stream, timeouts).await?; verify_genesis_hash(genesis.hash, expected) } @@ -439,9 +482,12 @@ mod tests { let err = check_genesis_hash(&mut ws, Some(&expected), fast()) .await - .expect_err("a different chain must not be accepted") - .to_string(); - assert!(err.contains("mismatch"), "unhelpful error: {err}"); + .expect_err("a different chain must not be accepted"); + assert_eq!(err.kind(), Failure::GenesisMismatch); + assert!( + err.to_string().contains("mismatch"), + "unhelpful error: {err}" + ); } /// Without `--genesis-hash` there is nothing to enforce, so any chain is @@ -466,9 +512,10 @@ mod tests { .await; let expected = parse_genesis_hash(VALID_HASH).unwrap(); - assert!(check_genesis_hash(&mut ws, Some(&expected), fast()) + let err = check_genesis_hash(&mut ws, Some(&expected), fast()) .await - .is_err()); + .expect_err("a refused call is not a pass"); + assert_eq!(err.kind(), Failure::RpcError); } /// A node that accepts the socket and then says nothing must not hang the @@ -485,9 +532,45 @@ mod tests { let expected = parse_genesis_hash(VALID_HASH).unwrap(); let err = check_genesis_hash(&mut ws, Some(&expected), fast()) .await - .expect_err("a silent node must time out") - .to_string(); - assert!(err.contains("no response"), "unhelpful error: {err}"); + .expect_err("a silent node must time out"); + assert_eq!(err.kind(), Failure::Timeout); + } + + /// The reason reads are bounded by a shared deadline rather than one clock + /// per frame: a node that chatters just fast enough would otherwise reset + /// the wait forever — every individual read inside the limit, the total + /// unbounded. Wrapped in an outer timeout so a regression fails the test + /// instead of hanging the suite. + #[tokio::test] + async fn unrelated_frames_cannot_extend_the_wait() { + let mut ws = mock_node(|mut ws| async move { + ws.next().await; + // Never the response we asked for, arriving faster than the budget. + loop { + if ws + .send(Message::Text(r#"{"jsonrpc":"2.0","id":999}"#.into())) + .await + .is_err() + { + return; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await; + + let expected = parse_genesis_hash(VALID_HASH).unwrap(); + let outcome = tokio::time::timeout( + Duration::from_secs(5), + check_genesis_hash(&mut ws, Some(&expected), fast()), + ) + .await + .expect("a chattering node must not hold the probe open indefinitely"); + + assert_eq!( + outcome.expect_err("noise is not an answer").kind(), + Failure::Timeout + ); } /// Serves a new-heads subscription: confirms it, pushes `heads` headers @@ -514,7 +597,7 @@ mod tests { #[tokio::test] async fn follows_the_requested_number_of_heads() { let mut ws = mock_node(|ws| serve_new_heads(ws, 3)).await; - assert!(follow_new_heads(&mut ws, 3, fast()).await.is_ok()); + assert_eq!(follow_new_heads(&mut ws, 3, fast()).await.unwrap(), 3); } /// Notifications for a different subscription must not count — one @@ -564,9 +647,12 @@ mod tests { let err = follow_new_heads(&mut ws, 1, fast()) .await - .expect_err("a refused subscription must not look like success") - .to_string(); - assert!(err.contains("rejected"), "unhelpful error: {err}"); + .expect_err("a refused subscription must not look like success"); + assert_eq!(err.kind(), Failure::RpcError); + assert!( + err.to_string().contains("rejected"), + "unhelpful error: {err}" + ); } /// A chain that stalls mid-subscription must time out, and the error must @@ -577,17 +663,20 @@ mod tests { let err = follow_new_heads(&mut ws, 3, fast()) .await - .expect_err("a stalled chain must time out") - .to_string(); - assert!(err.contains("saw 1 of 3"), "unhelpful error: {err}"); + .expect_err("a stalled chain must time out"); + assert_eq!(err.kind(), Failure::Timeout); + assert!( + err.to_string().contains("saw 1 of 3"), + "unhelpful error: {err}" + ); } - /// The node hangs up after answering only one of the four requests. The + /// The node hangs up after answering only one of the five requests. The /// terminated stream must surface an error, not spin on `None` forever. #[tokio::test] async fn hangup_errors_instead_of_spinning() { let mut ws = mock_node(|mut ws| async move { - take_requests(&mut ws, 4).await; + take_requests(&mut ws, 5).await; ws.send(Message::Text( r#"{"jsonrpc":"2.0","id":1,"result":"node"}"#.into(), )) @@ -597,10 +686,10 @@ mod tests { }) .await; - assert!( - query_node_info(&mut ws, fast()).await.is_err(), - "early hangup should surface an error" - ); + let err = query_node_info(&mut ws, fast()) + .await + .expect_err("early hangup should surface an error"); + assert_eq!(err.kind(), Failure::Transport); } /// Every request is answered with a JSON-RPC error. Errors still retire the @@ -608,7 +697,10 @@ mod tests { #[tokio::test] async fn all_error_responses_still_terminate() { let mut ws = mock_node(|mut ws| async move { - for id in 1..=4 { + // Keyed off the constants rather than a range: the query ids are not + // contiguous, and a mock that answers the wrong ones would look like + // a client bug. + for id in [ID_NAME, ID_CHAIN, ID_VERSION, ID_HEALTH, ID_HEADER] { ws.next().await; ws.send(Message::Text(format!( r#"{{"jsonrpc":"2.0","id":{id},"error":{{"code":-32601}}}}"# @@ -625,12 +717,13 @@ mod tests { assert_eq!(info, NodeInfo::default(), "nothing was actually reported"); } - /// Answers all four identity/health calls, keyed by request id so the + /// Answers all five identity/health/height calls, keyed by request id so the /// out-of-order case the client is built for is exercised (health first). - async fn serve_node_info(mut ws: WebSocketStream, health: &str) { + async fn serve_node_info(mut ws: WebSocketStream, health: &str, header: &str) { let replies = [ (ID_HEALTH, health.to_string()), (ID_VERSION, r#""1.24.0""#.to_string()), + (ID_HEADER, header.to_string()), (ID_NAME, r#""Parity Polkadot""#.to_string()), (ID_CHAIN, r#""Polkadot""#.to_string()), ]; @@ -645,15 +738,13 @@ mod tests { ws.next().await; } + /// A healthy node, for tests that care about one field in isolation. + const HEALTHY: &str = r#"{"peers":42,"isSyncing":false,"shouldHavePeers":true}"#; + const AT_BLOCK_100: &str = r#"{"number":"0x64","parentHash":"0xdead"}"#; + #[tokio::test] - async fn query_reports_identity_and_health() { - let mut ws = mock_node(|ws| { - serve_node_info( - ws, - r#"{"peers":42,"isSyncing":false,"shouldHavePeers":true}"#, - ) - }) - .await; + async fn query_reports_identity_health_and_height() { + let mut ws = mock_node(|ws| serve_node_info(ws, HEALTHY, AT_BLOCK_100)).await; let info = query_node_info(&mut ws, fast()).await.unwrap(); assert_eq!( @@ -665,20 +756,33 @@ mod tests { peers: Some(42), is_syncing: Some(false), should_have_peers: Some(true), + best_block: Some(100), } ); } + /// The height is what answers "is this node stuck?", so a header the probe + /// cannot read must leave it empty rather than guess at zero. + #[tokio::test] + async fn unreadable_header_leaves_the_height_empty() { + let mut ws = mock_node(|ws| serve_node_info(ws, HEALTHY, r#"{"parentHash":"0x0"}"#)).await; + + let info = query_node_info(&mut ws, fast()).await.unwrap(); + assert_eq!(info.best_block, None); + assert_eq!(info.peers, Some(42), "the rest of the report survived"); + } + /// `system_health` is not exposed everywhere. A node that refuses it is /// still worth reporting on, so the refusal must cost only that field. #[tokio::test] async fn refused_health_call_leaves_identity_intact() { let mut ws = mock_node(|mut ws| async move { - take_requests(&mut ws, 4).await; + take_requests(&mut ws, 5).await; for (id, body) in [ (ID_NAME, r#""result":"Parity Polkadot""#), (ID_CHAIN, r#""result":"Polkadot""#), (ID_VERSION, r#""result":"1.24.0""#), + (ID_HEADER, r#""result":{"number":"0x64"}"#), (ID_HEALTH, r#""error":{"code":-32601}"#), ] { ws.send(Message::Text(format!( @@ -693,6 +797,7 @@ mod tests { let info = query_node_info(&mut ws, fast()).await.unwrap(); assert_eq!(info.chain.as_deref(), Some("Polkadot"), "identity survived"); + assert_eq!(info.best_block, Some(100), "height survived"); assert_eq!(info.peers, None, "a refused call reports nothing"); assert_eq!(info.is_syncing, None); } @@ -701,7 +806,8 @@ mod tests { /// node did not say" and "the node has no peers" mean opposite things. #[tokio::test] async fn partial_health_result_does_not_invent_values() { - let mut ws = mock_node(|ws| serve_node_info(ws, r#"{"isSyncing":true}"#)).await; + let mut ws = + mock_node(|ws| serve_node_info(ws, r#"{"isSyncing":true}"#, AT_BLOCK_100)).await; let info = query_node_info(&mut ws, fast()).await.unwrap(); assert_eq!(info.is_syncing, Some(true)); diff --git a/src/report.rs b/src/report.rs index 9ad0b2c..09a6529 100644 --- a/src/report.rs +++ b/src/report.rs @@ -3,6 +3,8 @@ use log::info; use serde::Serialize; +use crate::error::{Failure, ProbeError}; + /// Everything the probe learned about the node, and the exact shape `--json` /// prints to stdout. /// @@ -16,6 +18,11 @@ pub(crate) struct ProbeReport { pub(crate) endpoint: String, /// Whether every step succeeded. `false` means `error` is set. pub(crate) ok: bool, + /// What kind of thing went wrong, for a consumer that needs to branch. + /// `error` below says the same thing in prose and is free to be reworded; + /// this is the part that is safe to alert on. + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) failure: Option, #[serde(skip_serializing_if = "Option::is_none")] pub(crate) error: Option, /// The genesis hash the node reported, `0x`-prefixed. @@ -44,6 +51,11 @@ pub(crate) struct ProbeReport { /// alarming when this is true. #[serde(skip_serializing_if = "Option::is_none")] pub(crate) should_have_peers: Option, + /// The node's best block number. Compare it across runs, or against another + /// node, to tell a stuck node from a healthy one — the header carries no + /// timestamp, so the probe cannot judge staleness on its own. + #[serde(skip_serializing_if = "Option::is_none")] + pub(crate) best_block: Option, /// How many headers `--follow` observed. #[serde(skip_serializing_if = "Option::is_none")] pub(crate) heads_followed: Option, @@ -73,19 +85,21 @@ pub(crate) fn check_requirements( report: &ProbeReport, min_peers: Option, require_synced: bool, -) -> Result<(), Box> { +) -> Result<(), ProbeError> { + let unmet = |message: String| ProbeError::new(Failure::RequirementUnmet, message); + if let Some(min) = min_peers { match report.peers { Some(peers) if peers < min => { - return Err( - format!("node has {peers} peer(s), --require-peers demands {min}").into(), - ) + return Err(unmet(format!( + "node has {peers} peer(s), --require-peers demands {min}" + ))) } None => { - return Err( + return Err(unmet( "node did not report a peer count, so --require-peers cannot be satisfied" .into(), - ) + )) } Some(peers) => info!("Peer requirement met: {peers} >= {min}"), } @@ -93,12 +107,12 @@ pub(crate) fn check_requirements( if require_synced { match report.is_syncing { - Some(true) => return Err("node is still syncing".into()), + Some(true) => return Err(unmet("node is still syncing".into())), None => { - return Err( + return Err(unmet( "node did not report its sync state, so --require-synced cannot be satisfied" .into(), - ) + )) } Some(false) => info!("Sync requirement met: node is not syncing"), } @@ -135,9 +149,16 @@ mod tests { fn peer_requirement_rejects_an_isolated_node() { let report = report_with_health(Some(0), Some(false)); let err = check_requirements(&report, Some(1), false) - .expect_err("a node with no peers must not pass") - .to_string(); - assert!(err.contains("0 peer(s)"), "unhelpful error: {err}"); + .expect_err("a node with no peers must not pass"); + assert_eq!( + err.kind(), + Failure::RequirementUnmet, + "a reachable node that fails a bar is not a connection problem" + ); + assert!( + err.to_string().contains("0 peer(s)"), + "unhelpful error: {err}" + ); } /// The trap this whole flag exists to avoid: a node that will not say how @@ -196,20 +217,59 @@ mod tests { } /// A failed run must still produce a report, or whatever is parsing stdout - /// gets nothing exactly when the node is broken. + /// gets nothing exactly when the node is broken. The classification is the + /// part a monitor branches on, so it has to be there and it has to be + /// stable — never make a consumer regex the prose. #[test] fn json_report_carries_the_failure() { let report = ProbeReport { endpoint: "ws://127.0.0.1:9944".into(), ok: false, - error: Some("genesis hash mismatch".into()), + failure: Some(Failure::GenesisMismatch), + error: Some("genesis hash mismatch — expected …".into()), ..Default::default() }; let json: serde_json::Value = serde_json::from_str(&serde_json::to_string(&report).unwrap()).unwrap(); assert_eq!(json["ok"], false); - assert_eq!(json["error"], "genesis hash mismatch"); + assert_eq!(json["failure"], "genesis_mismatch"); assert_eq!(json["endpoint"], "ws://127.0.0.1:9944"); + assert!(json["error"].is_string()); + } + + /// The names are the machine-facing contract — renaming a variant silently + /// breaks every alert built on it, so pin the wire strings. + #[test] + fn failure_kinds_serialise_to_stable_names() { + let cases = [ + (Failure::Config, "config"), + (Failure::Connect, "connect"), + (Failure::Timeout, "timeout"), + (Failure::Transport, "transport"), + (Failure::Protocol, "protocol"), + (Failure::RpcError, "rpc_error"), + (Failure::GenesisMismatch, "genesis_mismatch"), + (Failure::RequirementUnmet, "requirement_unmet"), + ]; + for (kind, expected) in cases { + assert_eq!(serde_json::to_value(kind).unwrap(), expected); + } + } + + /// A clean run must not carry a failure key at all. + #[test] + fn a_successful_report_has_no_failure() { + let report = ProbeReport { + ok: true, + best_block: Some(32_263_282), + ..Default::default() + }; + + let json: serde_json::Value = + serde_json::from_str(&serde_json::to_string(&report).unwrap()).unwrap(); + assert!(json.get("failure").is_none()); + assert!(json.get("error").is_none()); + assert_eq!(json["best_block"], 32_263_282u64); } } diff --git a/src/rpc.rs b/src/rpc.rs index 28770fe..d2bb564 100644 --- a/src/rpc.rs +++ b/src/rpc.rs @@ -6,11 +6,13 @@ use futures_util::StreamExt; use log::info; -use std::time::Duration; +use std::time::{Duration, Instant}; use tokio::net::TcpStream; use tokio_tungstenite::tungstenite::protocol::Message; use tokio_tungstenite::{connect_async, MaybeTlsStream, WebSocketStream}; +use crate::error::ProbeError; + /// The client's end of the WebSocket connection to the node. pub(crate) type NodeStream = WebSocketStream>; @@ -24,22 +26,26 @@ pub(crate) const ID_VERSION: u64 = 3; pub(crate) const ID_HEALTH: u64 = 4; pub(crate) const ID_SUBSCRIBE: u64 = 5; pub(crate) const ID_UNSUBSCRIBE: u64 = 6; +pub(crate) const ID_HEADER: u64 = 7; -/// How long to wait for the WebSocket connection to be established. -const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +/// Default wait for the WebSocket connection to be established. +pub(crate) const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); -/// How long to wait for any single response frame from the node. Without this a -/// node that accepts the socket and then goes quiet would hang the client. -const RPC_TIMEOUT: Duration = Duration::from_secs(10); +/// Default wait for a node to answer a request. Without a bound, a node that +/// accepts the socket and then goes quiet would hang the client. +pub(crate) const RPC_TIMEOUT: Duration = Duration::from_secs(10); -/// How long to wait for a pushed block header. Deliberately far longer than +/// Default wait for a pushed block header. Deliberately far longer than /// `RPC_TIMEOUT`: this waits on the chain's block time, not on the node being /// responsive. Polkadot targets ~6s, but parachains and dev chains vary widely. -const SUBSCRIPTION_TIMEOUT: Duration = Duration::from_secs(120); +pub(crate) const SUBSCRIPTION_TIMEOUT: Duration = Duration::from_secs(120); -/// The network waits, grouped so they can be shortened in tests. +/// The network waits, grouped so they can be shortened in tests and overridden +/// from the command line. #[derive(Debug, Clone, Copy)] pub(crate) struct Timeouts { + /// Waiting for the connection to be established. + pub(crate) connect: Duration, /// Waiting for the node to answer a request. pub(crate) rpc: Duration, /// Waiting for the chain to produce a block. @@ -49,31 +55,75 @@ pub(crate) struct Timeouts { impl Default for Timeouts { fn default() -> Self { Timeouts { + connect: CONNECT_TIMEOUT, rpc: RPC_TIMEOUT, head: SUBSCRIPTION_TIMEOUT, } } } -/// Opens a WebSocket connection to the node. +/// A budget for a whole operation, rather than for one frame of it. /// -/// Bounded by `CONNECT_TIMEOUT`, so an address that accepts TCP but never -/// completes the WebSocket handshake fails rather than hanging. +/// The distinction matters because the read loops skip frames they do not +/// recognise. Bounding each individual read would let a node that emits one +/// unrelated frame just inside the limit keep the probe waiting forever — every +/// single wait short, the total unbounded. A deadline is set once when the +/// operation starts and every read inside it draws down the same clock. +#[derive(Debug, Clone, Copy)] +pub(crate) struct Deadline { + at: Instant, + /// The budget it was created with, kept only so timeout messages can name + /// the limit the caller actually set. + budget: Duration, +} + +impl Deadline { + /// Starts a budget of `budget` from now. + pub(crate) fn after(budget: Duration) -> Self { + Deadline { + at: Instant::now() + budget, + budget, + } + } + + /// How much of the budget is left, or a timeout error if none is. + pub(crate) fn remaining(&self) -> Result { + self.at + .checked_duration_since(Instant::now()) + .filter(|left| !left.is_zero()) + .ok_or_else(|| self.expired()) + } + + /// The error to report when this budget runs out. + pub(crate) fn expired(&self) -> ProbeError { + ProbeError::timeout(format!( + "node sent no usable response within {:?}", + self.budget + )) + } +} + +/// Opens a WebSocket connection to the node. /// /// # Arguments /// /// * `address` - The node's endpoint, `ws://` or `wss://`. +/// * `timeout` - How long to allow for the connection to be established. /// /// # Returns /// /// The open connection, or an error naming the address that failed. -pub(crate) async fn connect(address: &str) -> Result> { +pub(crate) async fn connect(address: &str, timeout: Duration) -> Result { info!("Connecting to node at {address}"); - let (ws_stream, response) = tokio::time::timeout(CONNECT_TIMEOUT, connect_async(address)) + let (ws_stream, response) = tokio::time::timeout(timeout, connect_async(address)) .await - .map_err(|_| format!("timed out connecting to {address} after {CONNECT_TIMEOUT:?}"))? - .map_err(|e| format!("failed to connect to {address}: {e}"))?; + .map_err(|_| { + ProbeError::timeout(format!( + "timed out connecting to {address} after {timeout:?}" + )) + })? + .map_err(|e| ProbeError::connect(format!("failed to connect to {address}: {e}")))?; info!("Connected to the node with response: {response:?}"); Ok(ws_stream) @@ -83,30 +133,32 @@ pub(crate) async fn connect(address: &str) -> Result Result> { + deadline: Deadline, +) -> Result { loop { - let msg = tokio::time::timeout(timeout, ws_stream.next()) + let msg = tokio::time::timeout(deadline.remaining()?, ws_stream.next()) .await - .map_err(|_| format!("node sent no response within {timeout:?}"))? - .ok_or("connection closed by the node")??; + .map_err(|_| deadline.expired())? + .ok_or_else(|| ProbeError::transport("connection closed by the node"))??; match msg { Message::Text(text) => return Ok(text), - Message::Close(_) => return Err("connection closed by the node".into()), + Message::Close(_) => { + return Err(ProbeError::transport("connection closed by the node")) + } _ => continue, } } diff --git a/src/test_support.rs b/src/test_support.rs index efc6a5a..e33a8e8 100644 --- a/src/test_support.rs +++ b/src/test_support.rs @@ -39,6 +39,7 @@ where /// while the mock's reply is still in flight, and the test fails at random. pub(crate) fn fast() -> Timeouts { Timeouts { + connect: Duration::from_millis(250), rpc: Duration::from_millis(250), head: Duration::from_millis(250), }