diff --git a/crates/blackbeard-api/src/intents.rs b/crates/blackbeard-api/src/intents.rs index 6049b4f..005ad5b 100644 --- a/crates/blackbeard-api/src/intents.rs +++ b/crates/blackbeard-api/src/intents.rs @@ -139,18 +139,44 @@ async fn scan(chain: Arc, store: Store, near: NearClient, config: // Per batch, not once: resharding renumbers shards, and a layout read // at startup would follow the wrong one after it while every call // kept succeeding. - let shard = match near.shard_layout(heights[0]).await { - Ok(layout) => core::shard_for( - &config.contract, - &layout.boundary_accounts, - &layout.shard_ids, - ), - Err(e) => { - tracing::warn!(chain = %id, error = %e, "near shard layout unreadable"); - tokio::time::sleep(RETRY).await; - continue; + // + // From the first height in the batch that exists. NEAR skips heights, + // and every endpoint answers UNKNOWN_BLOCK for one that was never + // produced — asking at the batch's first height alone stalled the scan + // for good the first time that height was a skipped one, because each + // retry asked the same question (blackbeard/observer#31). + let mut layout = None; + let mut failed = None; + for &height in &heights { + match near.shard_layout(height).await { + Ok(l) => { + layout = Some(l); + break; + } + Err(e) if blackbeard_data::near::unknown(&e) => continue, + Err(e) => { + failed = Some(e); + break; + } } + } + if let Some(e) = failed { + tracing::warn!(chain = %id, error = %e, "near shard layout unreadable"); + tokio::time::sleep(RETRY).await; + continue; + } + let Some(layout) = layout else { + // Every height in the batch is unknown everywhere. A whole batch of + // skips is implausible; say so and retry rather than claim them. + tracing::warn!(chain = %id, from = heights[0], "no height in the batch is known to any endpoint"); + tokio::time::sleep(RETRY).await; + continue; }; + let shard = core::shard_for( + &config.contract, + &layout.boundary_accounts, + &layout.shard_ids, + ); let Some(shard) = shard else { tracing::error!(chain = %id, contract = %config.contract, "contract maps to no shard"); tokio::time::sleep(RETRY).await; diff --git a/crates/blackbeard-data/src/near.rs b/crates/blackbeard-data/src/near.rs index 31daf56..e1a99b6 100644 --- a/crates/blackbeard-data/src/near.rs +++ b/crates/blackbeard-data/src/near.rs @@ -102,7 +102,7 @@ pub struct NearClient { /// Said both for a height that was never produced — NEAR skips heights — and /// for one the node has pruned. Only the endpoint's retention tells them apart, /// which is why a live node saying it is followed by asking an archive. -fn unknown(e: &DataError) -> bool { +pub fn unknown(e: &DataError) -> bool { matches!(e, DataError::Rpc { message, .. } if message.contains("UNKNOWN_BLOCK") || message.contains("UNKNOWN_CHUNK") diff --git a/crates/blackbeard-data/src/rpc.rs b/crates/blackbeard-data/src/rpc.rs index 2c8079b..e9c0d4e 100644 --- a/crates/blackbeard-data/src/rpc.rs +++ b/crates/blackbeard-data/src/rpc.rs @@ -405,27 +405,10 @@ impl RpcClient { /// unhealthy endpoint: pruning is a legitimate answer, and failing over on /// it would walk the whole list asking a question none of them can answer. async fn call_one(&self, url: &str, body: &Value, method: &str) -> Result { - let resp: Value = self - .http - .post(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(), - }) + let resp = self.http.post(url).json(body).send().await?; + let status = resp.status().as_u16(); + let bytes = resp.bytes().await?; + parse_response(method, status, &bytes) } /// `system_health`. @@ -916,8 +899,72 @@ async fn follow_once(ws_url: &str, sink: &mpsc::Sender) -> Result<(), Ok(()) } +/// Classify one HTTP response to a JSON-RPC call. +/// +/// The body is read **whatever the status**. A JSON-RPC `error` object is the +/// node answering, and stays [`DataError::Rpc`] even when it arrives with a +/// 4xx — which is how neard sends them: `UNKNOWN_BLOCK` comes back as HTTP 422, +/// where FastNEAR's proxy sent the same body with a 200. Calling +/// `error_for_status()` first turned that answer into a transport failure, so a +/// skipped height could never be recognised as one, the archive was never +/// asked, and a multi-endpoint chain would have failed over on a node's +/// perfectly valid "no". blackbeard/observer#31. +/// +/// A non-success status *without* a JSON-RPC body — a proxy's 502, an HTML +/// error page — is still the host not answering, and is +/// [`DataError::Malformed`], which `call` fails over on. +fn parse_response(method: &str, status: u16, body: &[u8]) -> Result { + let malformed = |message: String| DataError::Malformed { + method: method.to_owned(), + message, + }; + let parsed: Option = serde_json::from_slice(body).ok(); + if let Some(err) = parsed.as_ref().and_then(|v| v.get("error")) { + return Err(DataError::Rpc { + method: method.to_owned(), + message: err.to_string(), + }); + } + if !(200..300).contains(&status) { + return Err(malformed(format!("HTTP {status} without a JSON-RPC error"))); + } + let parsed = parsed.ok_or_else(|| malformed("response is not JSON".into()))?; + parsed + .get("result") + .cloned() + .ok_or_else(|| malformed("response carried neither result nor error".into())) +} + #[cfg(test)] mod tests { + + #[test] + fn a_json_rpc_error_is_the_node_answering_whatever_the_status() { + // neard's own body for a skipped height, as served with HTTP 422. + let body = br#"{"jsonrpc":"2.0","error":{"name":"HANDLER_ERROR","cause":{"info":{},"name":"UNKNOWN_BLOCK"},"code":-32000,"message":"Server error","data":"Block Not Found: BLOCK HEIGHT: 218363444"},"id":1}"#; + for status in [200, 422, 500] { + match parse_response("chunk", status, body) { + Err(DataError::Rpc { message, .. }) => assert!(message.contains("UNKNOWN_BLOCK")), + other => panic!("HTTP {status}: {other:?}"), + } + } + } + + #[test] + fn a_failure_without_a_json_rpc_body_is_the_host_not_answering() { + assert!(matches!( + parse_response("status", 502, b"Bad Gateway"), + Err(DataError::Malformed { .. }) + )); + assert!(matches!( + parse_response("status", 200, br#"{"jsonrpc":"2.0","id":1}"#), + Err(DataError::Malformed { .. }) + )); + assert_eq!( + parse_response("status", 200, br#"{"jsonrpc":"2.0","result":7,"id":1}"#).unwrap(), + serde_json::json!(7) + ); + } use super::*; /// The probe key is derived, not typed. A wrong key is absent everywhere,