fix(data): a skipped NEAR height no longer stalls the intents scan

NEAR does not produce a block at every height, and 218,363,444 was one it
skipped. The backfill asked for its shard layout at the batch's first height,
got UNKNOWN_BLOCK from every endpoint, and asked the same question every
15 s from then on: production's indexed_from stopped at 218,363,445. The tip
side runs the same code and would have frozen the first time a batch began on
a skip. The layout now comes from the first height in the batch that exists.

The second half of it: neard sends JSON-RPC errors with HTTP 422 where
FastNEAR's proxy sent them with 200, and call_one checked the status before
reading the body, so the node's own "no such block" arrived as a transport
failure. chunk() could not see a skip as a skip, deep() never asked the
archive, and on a chain with several endpoints a node's valid answer would
have been failed over. parse_response reads the body whatever the status: a
JSON-RPC error is the node answering; a non-success status without one is the
host not answering, as before.

Checked against near.internal with the scan seeded to begin a backfill batch
on the skipped height: it walked past it with no warnings.

Closes #31

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015pkwBsR54NRmZyU4QVhvfj

crates/blackbeard-api/src/intents.rs

@@ -139,18 +139,44 @@ async fn scan(chain: Arc<ChainRuntime>, 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;

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")

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<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(),
-            })
+        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<SeenHead>) -> 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<Value, DataError> {
+    let malformed = |message: String| DataError::Malformed {
+        method: method.to_owned(),
+        message,
+    };
+    let parsed: Option<Value> = 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"<html>Bad Gateway</html>"),
+            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,