fix: a bogus length prefix aborted the daemon, and the gap fill never resumed

Two faults, one visible symptom: both testnet leaderboards frozen for a day at
the height their gap opened at, while mainnet was fine.

The daemon had aborted 316 times in ten hours, always identically —
`memory allocation of 82014765760 bytes failed`, 76.4 GiB. That is
1,025,184,572 x 80, and 80 is `size_of::<scale_value::Value<u32>>()`:
scale-value sizes a sequence's `Vec` from the compact length in the blob before
decoding a single item, so a blob that disagrees with the registry asks for an
allocation of any size at all and Rust aborts on the failed one. The step that
would have caught the mismatch is the one that never runs, which is why
`decode_events(&blob).ok()` could not help — there was no `Err`, only SIGABRT.

Every decode now goes through `Runtime::decode_checked`, which walks the bytes
first with scale-decode's `IgnoreVisitor`. That crate has no `with_capacity`
anywhere, so an impossible length runs out of input on the first item and comes
back as an error. The regression test is the exact four-byte blob, reproduced
against the mainnet fixture before the guard went in — the abort message matched
production byte for byte.

`fill_gap` then turned a short process lifetime into no progress whatsoever: it
accumulated the whole gap and wrote once after the loop, so every interruption
discarded everything read. Each block is four RPC round trips, one a historical
state read; 1,696 blocks against a remote endpoint does not fit in the two
minutes the aborts allowed. And this is not only about the aborts —
`blackbeard-api-cert.path` restarts the service several times a day, so any gap
wider than that interval was already unfillable, logging
`filling a gap in the head stream` on every start with the same `from` and
looking like progress. It flushes every 256 blocks now, and says how many it
recorded.

Both traps are in CLAUDE.md, because neither looks broken from the outside: one
presents as a service that is merely restarting, the other as a catch-up that is
merely slow.

Closes #10
Closes #11

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

CLAUDE.md

@@ -173,6 +173,34 @@ decodes it because that is the only way to know where the call begins, then
 keeps the scheme's *name* and the byte count and discards the bytes. At a block
 a second, storing them would be gigabytes a month that nothing renders.
 
+**A length prefix out of a blob must never size an allocation.**
+`scale_value` decodes a sequence by doing `Vec::with_capacity(remaining())`
+*before* it decodes the first item, where `remaining()` is the compact length
+read straight out of the bytes. A `Value<u32>` is 80 bytes, so a blob that
+disagrees with the registry can ask for any allocation at all — and a failed
+allocation aborts the process rather than returning an error. Deployed, that was
+`memory allocation of 82014765760 bytes failed`, 316 times in ten hours, from a
+claimed length of 1,025,184,572. The step that would have caught the mismatch is
+the one that never runs, so `.ok()` at the call site catches nothing: there is no
+`Err`, only SIGABRT. Every decode therefore goes through
+`Runtime::decode_checked`, which walks the bytes first with `scale_decode`'s
+`IgnoreVisitor` — that crate contains no `with_capacity` anywhere, so an
+impossible length runs out of input on the first item and comes back as an
+error. This is what makes "a block decoded against the wrong runtime must fail"
+true in the case where it previously did neither. Do not route a new decode
+around it.
+
+**A catch-up that writes once at the end makes no progress at all.** `fill_gap`
+used to accumulate the whole gap and call `record_blocks` after the loop, so
+anything that ended the process first discarded every block collected. Each
+block is four RPC round trips, one a historical state read, and this service is
+restarted several times a day by `blackbeard-api-cert.path` alone — so a gap
+wider than the interval between routine restarts was permanently unfillable
+while logging `filling a gap in the head stream` on every start, always with the
+same `from`, reading exactly like progress. Both testnets sat a day behind that
+way. It flushes every `GAP_FLUSH_BLOCKS` now; anything that walks history in
+this codebase has to be resumable, because the process is not long-lived.
+
 **Backfill progress is a cursor, not `max(height) from chain_event`.** A block
 with no indexed event and a block never looked at give the same answer, and the
 first is the common case — `INDEXED_EVENTS` is deliberately narrow. Inferring

Cargo.lock

@@ -291,6 +291,7 @@ dependencies = [
  "parity-scale-codec",
  "primitive-types",
  "qp-poseidon-core",
+ "scale-decode",
  "scale-info",
  "scale-value",
  "serde_json",

Cargo.toml

@@ -29,6 +29,10 @@ frame-metadata = { version = "23", default-features = false, features = ["curren
 parity-scale-codec = { version = "3", default-features = false, features = ["derive"] }
 scale-info = { version = "2", default-features = false }
 scale-value = { version = "0.18", default-features = false }
+# Only for `IgnoreVisitor`: the pre-flight walk in `Runtime::decode_checked`
+# that keeps a bogus length prefix from sizing an allocation. Pinned to the
+# version `scale-value` itself resolves, so both see one type registry.
+scale-decode = { version = "0.16", default-features = false }
 # The chain's own reward-address derivation, so the observer computes the same
 # account `pallets/mining-rewards` pays rather than guessing or asking.
 qp-poseidon-core = { version = "3.1.0", default-features = false }

crates/blackbeard-api/src/ingest.rs

@@ -342,6 +342,30 @@ async fn ingest(
     tracing::warn!(chain = %chain.id(), "head stream ended; ingest stopping");
 }
 
+/// How many gap-filled blocks to hold before writing them.
+///
+/// The batch used to be the whole gap, written once after the loop, and that
+/// made the fill **all or nothing**: anything that ended the process first threw
+/// away every block collected. Which is not the rare case it sounds like. Each
+/// block here is four RPC round trips, one of them a historical state read, so
+/// a few thousand blocks against a remote endpoint takes minutes — while
+/// `blackbeard-api-cert.path` restarts this service on every certificate
+/// rotation, several times a day, and every deploy restarts it too.
+///
+/// A gap wider than the interval between routine restarts was therefore
+/// permanently unfillable, and said nothing about it: the same
+/// `filling a gap in the head stream` line on every start, always the same
+/// `from`, reading exactly like progress. Both testnets sat frozen at the
+/// height their gap opened at for a day — planck 1,696 blocks behind,
+/// heisenberg 2,792 — while a decoder abort held process lifetime at under two
+/// minutes.
+///
+/// Flushing in chunks makes the work resumable and monotonic: whatever a pass
+/// finishes is durable, and the next pass starts higher. It also bounds the
+/// `Vec`, which `max_gap_fill_blocks` never did — that limit exists to say how
+/// far back to reach, not how much to hold.
+const GAP_FLUSH_BLOCKS: usize = 256;
+
 /// Fetch and record the blocks between two heads.
 ///
 /// Best effort: a node that has pruned, or that fails midway, costs the window
@@ -350,6 +374,7 @@ async fn fill_gap(chain: &Arc<ChainRuntime>, store: &Store, from: u64, to: u64)
     tracing::info!(chain = %chain.id(), from, to, "filling a gap in the head stream");
     let mut batch = Vec::new();
     let mut observed = Vec::new();
+    let mut recorded = 0usize;
     for height in from..to {
         let Ok(Some(hash)) = chain.rpc.block_hash(height).await else {
             continue;
@@ -398,17 +423,55 @@ async fn fill_gap(chain: &Arc<ChainRuntime>, store: &Store, from: u64, to: u64)
             authored_at,
             difficulty: magnitude,
         });
+
+        if batch.len() >= GAP_FLUSH_BLOCKS {
+            recorded += flush_gap(chain, store, &mut batch, &mut observed).await;
+        }
     }
 
-    if let Err(e) = store.record_blocks(&batch).await {
-        tracing::warn!(chain = %chain.id(), error = %e, "gap blocks were not persisted");
+    recorded += flush_gap(chain, store, &mut batch, &mut observed).await;
+    // Counted, because a pass that recorded nothing and a pass that recorded
+    // everything used to produce the same single line.
+    tracing::info!(chain = %chain.id(), from, to, recorded, "gap fill finished");
+}
+
+/// Write one chunk of gap-filled blocks and push them into the window.
+///
+/// Returns how many were written, so the caller can say what a pass achieved.
+/// The lock is taken after the await and released before returning: everything
+/// under `ChainRuntime::inner` is CPU work on in-memory collections, and it is
+/// a `std::sync::RwLock` precisely so that holding it across an await is not
+/// possible to do by accident.
+async fn flush_gap(
+    chain: &Arc<ChainRuntime>,
+    store: &Store,
+    batch: &mut Vec<BlockRecord>,
+    observed: &mut Vec<Observed>,
+) -> usize {
+    if batch.is_empty() {
+        return 0;
     }
+    let count = batch.len();
+    if let Err(e) = store.record_blocks(batch).await {
+        // The window still gets them: they were read from the chain, and a
+        // failed write is a reason to serve them from memory, not to forget
+        // them. The next pass will try this stretch again.
+        tracing::warn!(chain = %chain.id(), error = %e, count, "gap blocks were not persisted");
+        batch.clear();
+        let mut inner = chain.write();
+        for o in observed.drain(..) {
+            inner.window.push(o, false);
+        }
+        return 0;
+    }
+    batch.clear();
     let mut inner = chain.write();
-    for o in observed {
+    for o in observed.drain(..) {
         // Not at the tip: these are being caught up on, and their observation
         // times are all "now" rather than when they were authored.
         inner.window.push(o, false);
     }
+    count
 }
 
 /// Record one head.

crates/blackbeard-core/Cargo.toml

@@ -16,6 +16,7 @@ twox-hash.workspace = true
 frame-metadata.workspace = true
 hex.workspace = true
 parity-scale-codec.workspace = true
+scale-decode.workspace = true
 scale-info.workspace = true
 scale-value.workspace = true
 serde_json.workspace = true

crates/blackbeard-core/src/runtime.rs

@@ -567,10 +567,55 @@ impl Runtime {
         cursor: &mut &[u8],
         what: &str,
     ) -> Result<scale_value::Value<u32>, RuntimeError> {
-        scale_value::scale::decode_as_type(cursor, ty, &self.metadata.types)
+        self.decode_checked(ty, cursor)
             .map_err(|e| RuntimeError::Decode(format!("{what}: {e}")))
     }
 
+    /// Decode one registry type, walking the bytes first without building
+    /// anything from them.
+    ///
+    /// **`scale_value` sizes a sequence's `Vec` from the length prefix before
+    /// it decodes a single item** — `Vec::with_capacity($value.remaining())` in
+    /// its `to_unnamed_composite!`, where `remaining()` is the compact length
+    /// read straight out of the blob. A `Value<u32>` is 80 bytes, so a blob
+    /// that disagrees with the registry can ask for an allocation of any size
+    /// at all, and Rust aborts the process on a failed one. Deployed, that was
+    /// `memory allocation of 82014765760 bytes failed` — 76 GiB, from a length
+    /// of 1,025,184,572 — three hundred times over, every two minutes, and it
+    /// took the whole daemon with it each time.
+    ///
+    /// The step that would have caught the mismatch is the one that never
+    /// runs: the allocation happens *before* the first item is decoded. So
+    /// `.ok()` at a call site cannot help either — there is no `Err` to catch.
+    ///
+    /// `scale_decode`'s `IgnoreVisitor` walks the same bytes against the same
+    /// type and allocates nothing at all — the crate contains no
+    /// `with_capacity` anywhere. A length that cannot be satisfied runs out of
+    /// input on the first item and comes back as an error, which is what this
+    /// decoder is supposed to produce. The second pass then costs one more walk
+    /// of a few hundred kilobytes, against the RPC round trip that fetched it.
+    ///
+    /// This is what makes the guarantee in `CLAUDE.md` — a block decoded
+    /// against the wrong runtime must *fail*, not succeed — true in the case
+    /// where it previously did neither.
+    fn decode_checked(
+        &self,
+        ty: u32,
+        cursor: &mut &[u8],
+    ) -> Result<scale_value::Value<u32>, String> {
+        let mut probe: &[u8] = cursor;
+        scale_decode::visitor::decode_with_visitor(
+            &mut probe,
+            ty,
+            &self.metadata.types,
+            scale_decode::visitor::IgnoreVisitor::<scale_info::PortableRegistry>::new(),
+        )
+        .map_err(|e| e.to_string())?;
+
+        scale_value::scale::decode_as_type(cursor, ty, &self.metadata.types)
+            .map_err(|e| e.to_string())
+    }
+
     /// The account behind a `MultiAddress`, when it is a plain id.
     ///
     /// The other forms — an index, a raw 20 bytes — name an account the chain
@@ -803,9 +848,7 @@ impl Runtime {
     ) -> Option<serde_json::Value> {
         let offset = 32 + map.key_offset?;
         let mut cursor = storage_key.get(offset..)?;
-        let value =
-            scale_value::scale::decode_as_type(&mut cursor, map.key_ty, &self.metadata.types)
-                .ok()?;
+        let value = self.decode_checked(map.key_ty, &mut cursor).ok()?;
         let mut accounts = BTreeSet::new();
         Some(self.normalise(&value, &mut accounts))
     }
@@ -846,8 +889,9 @@ impl Runtime {
         raw: &[u8],
     ) -> Result<(serde_json::Value, Vec<String>), RuntimeError> {
         let mut cursor = raw;
-        let value = scale_value::scale::decode_as_type(&mut cursor, ty, &self.metadata.types)
-            .map_err(|e| RuntimeError::Decode(e.to_string()))?;
+        let value = self
+            .decode_checked(ty, &mut cursor)
+            .map_err(RuntimeError::Decode)?;
         if !cursor.is_empty() {
             return Err(RuntimeError::Decode(format!(
                 "{} trailing bytes in storage value",
@@ -1122,9 +1166,9 @@ impl Runtime {
     /// truth.
     pub fn decode_events(&self, blob: &[u8]) -> Result<Vec<DecodedEvent>, RuntimeError> {
         let mut cursor = blob;
-        let value =
-            scale_value::scale::decode_as_type(&mut cursor, self.events_ty, &self.metadata.types)
-                .map_err(|e| RuntimeError::Decode(e.to_string()))?;
+        let value = self
+            .decode_checked(self.events_ty, &mut cursor)
+            .map_err(RuntimeError::Decode)?;
         if !cursor.is_empty() {
             return Err(RuntimeError::Decode(format!(
                 "{} trailing bytes; metadata does not match this block",
@@ -1447,6 +1491,31 @@ mod tests {
         assert!(names.contains(&"System"));
     }
 
+    /// A length prefix nothing could satisfy is an error, not an allocation.
+    ///
+    /// This blob is four bytes: the SCALE compact encoding of 1,025,184,572,
+    /// and nothing after it. `scale_value` would size the sequence's `Vec` from
+    /// that length before decoding an item — 1,025,184,572 × 80 bytes, which is
+    /// the `memory allocation of 82014765760 bytes failed` the deployed daemon
+    /// aborted on three hundred times in ten hours. Reproduced exactly here
+    /// before the guard went in, so the number is not a coincidence.
+    ///
+    /// **Do not "simplify" this by calling `decode_as_type` directly.** The
+    /// unguarded call does not return an error to assert on; it takes the test
+    /// binary down with SIGABRT, and would do the same to CI.
+    #[test]
+    fn a_length_prefix_that_cannot_fit_is_refused_rather_than_allocated() {
+        let rt = Runtime::from_metadata(&metadata()).expect("parses");
+        let claimed: u32 = 1_025_184_572;
+        let blob = ((((claimed as u64) << 2) as u32) | 0b10).to_le_bytes();
+
+        let err = rt.decode_events(&blob).expect_err("must not decode");
+        assert!(
+            matches!(err, RuntimeError::Decode(_)),
+            "expected a decode error, got {err:?}"
+        );
+    }
+
     #[test]
     fn nonsense_is_refused_rather_than_guessed_at() {
         assert_eq!(