diff --git a/CLAUDE.md b/CLAUDE.md index 2125c9b..09cac30 100644 --- a/CLAUDE.md +++ b/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` 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 diff --git a/Cargo.lock b/Cargo.lock index 49cc10a..818ff49 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -291,6 +291,7 @@ dependencies = [ "parity-scale-codec", "primitive-types", "qp-poseidon-core", + "scale-decode", "scale-info", "scale-value", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 5c8dd65..e7dbb7f 100644 --- a/Cargo.toml +++ b/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 } diff --git a/crates/blackbeard-api/src/ingest.rs b/crates/blackbeard-api/src/ingest.rs index 6db3752..43c4a6e 100644 --- a/crates/blackbeard-api/src/ingest.rs +++ b/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, 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, 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, + store: &Store, + batch: &mut Vec, + observed: &mut Vec, +) -> 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. diff --git a/crates/blackbeard-core/Cargo.toml b/crates/blackbeard-core/Cargo.toml index 7b95069..1b07097 100644 --- a/crates/blackbeard-core/Cargo.toml +++ b/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 diff --git a/crates/blackbeard-core/src/runtime.rs b/crates/blackbeard-core/src/runtime.rs index ede5580..d1a2240 100644 --- a/crates/blackbeard-core/src/runtime.rs +++ b/crates/blackbeard-core/src/runtime.rs @@ -567,10 +567,55 @@ impl Runtime { cursor: &mut &[u8], what: &str, ) -> Result, 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` 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, String> { + let mut probe: &[u8] = cursor; + scale_decode::visitor::decode_with_visitor( + &mut probe, + ty, + &self.metadata.types, + scale_decode::visitor::IgnoreVisitor::::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 { 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), 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, 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!(