feat: decode events against the runtime's own metadata

A header says who mined a block, not what they were paid: the amount is
remaining supply over an emission divisor plus fees, quantized with dust
carried forward, so only the `MinerRewarded` event knows it. Reading it means
decoding SCALE against the registry of the runtime that produced the block, and
this chain's shapes differ from vanilla Substrate because of the post-quantum
signatures.

So the chain describes itself. `state_getMetadata` at a block hash executes
`Metadata_metadata` against the runtime WASM in that block's state; the v14
registry that comes back drives a decoder that knows nothing about mining,
rewards or transfers. `INDEXED_EVENTS` is the only place a name appears —
adding transfers or anything the chain grows next is one line and a query, not
a migration, a struct and a decoder.

Metadata is cached per spec_version, because a node that starts pruning would
otherwise make history undecodable. Blocks are tried against the tip's runtime
first, which costs no extra call; `decode_events` refuses a partial read, so a
block from before an upgrade fails loudly rather than decoding into plausible
nonsense, and that failure is what asks which runtime actually produced it.

`backfill` walks outward from the tip — to the tip first, then toward genesis —
because live indexing alone leaves holes an account page would show: a restart,
an RPC blip, a gap fill that ran before the metadata was cached. Progress is a
cursor rather than `max(height)`, since a block with no indexed event and a
block never read are otherwise the same answer.

Verified on mainnet: block 13160's preimage derives, through the same Poseidon2
the chain runs, to exactly the `miner` field of its own reward event — so the
header author and the payee join with no lookup table.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Jp6a8EDar9ueEhAxzep4V5
This commit is contained in:
2026-09-09 16:47:10 +03:00
co-authored by Claude Opus 5
parent 6a29250c88
commit eb26a7bb3c
23 changed files with 1902 additions and 6 deletions
@@ -0,0 +1,34 @@
{
"db_name": "PostgreSQL",
"query": "\n select spec_version, spec_name, first_seen_height\n from runtime_metadata where chain = $1 order by spec_version asc\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "spec_version",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "spec_name",
"type_info": "Text"
},
{
"ordinal": 2,
"name": "first_seen_height",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false,
false
]
},
"hash": "4777ff33fa3c39f2bad9d5c4182460c0d32eb157d92bdb6f836ac4f097cc7a01"
}
@@ -0,0 +1,18 @@
{
"db_name": "PostgreSQL",
"query": "\n insert into runtime_metadata\n (chain, spec_version, spec_name, metadata, first_seen_height)\n values ($1, $2, $3, $4, $5)\n on conflict (chain, spec_version) do nothing\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Int8",
"Text",
"Bytea",
"Int8"
]
},
"nullable": []
},
"hash": "487dfea1cd62fec4a9a2676318505d9ae3f5039a036e186fc928975860c86356"
}
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "select metadata from runtime_metadata where chain = $1 and spec_version = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "metadata",
"type_info": "Bytea"
}
],
"parameters": {
"Left": [
"Text",
"Int8"
]
},
"nullable": [
false
]
},
"hash": "5619f644cab76e92eedd55f03d914773c4ca90656d99a377fe212905e4c1da90"
}
@@ -0,0 +1,62 @@
{
"db_name": "PostgreSQL",
"query": "\n select height, event_index, pallet, variant, phase, extrinsic_index,\n fields as \"fields!\"\n from chain_event\n where chain = $1 and pallet = $2 and variant = $3 and fields @> $4\n order by height desc, event_index desc\n limit $5\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "height",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "event_index",
"type_info": "Int4"
},
{
"ordinal": 2,
"name": "pallet",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "variant",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "phase",
"type_info": "Text"
},
{
"ordinal": 5,
"name": "extrinsic_index",
"type_info": "Int4"
},
{
"ordinal": 6,
"name": "fields!",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"Text",
"Text",
"Jsonb",
"Int8"
]
},
"nullable": [
false,
false,
false,
false,
false,
true,
false
]
},
"hash": "89288c55f511f60e817045e0fb579b2592236c9dedf86a0e816dbb598b8a6166"
}
@@ -0,0 +1,28 @@
{
"db_name": "PostgreSQL",
"query": "select low, high from event_scan where chain = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "low",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "high",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
false,
false
]
},
"hash": "9d9c16dd7fe8cd30305d731494298960b100e0c09f407118d534e4e796f970d0"
}
@@ -0,0 +1,21 @@
{
"db_name": "PostgreSQL",
"query": "\n insert into chain_event\n (chain, height, event_index, pallet, variant, phase, extrinsic_index, fields)\n select * from unnest(\n $1::text[], $2::bigint[], $3::int[], $4::text[],\n $5::text[], $6::text[], $7::int[], $8::jsonb[]\n )\n on conflict (chain, height, event_index) do update set\n pallet = excluded.pallet,\n variant = excluded.variant,\n phase = excluded.phase,\n extrinsic_index = excluded.extrinsic_index,\n fields = excluded.fields\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"TextArray",
"Int8Array",
"Int4Array",
"TextArray",
"TextArray",
"TextArray",
"Int4Array",
"JsonbArray"
]
},
"nullable": []
},
"hash": "f4954eb7baea34b5121dc516522a29637c663f85c8d9aacc96889492174c0505"
}
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "\n insert into event_scan (chain, low, high) values ($1, $2, $3)\n on conflict (chain) do update\n set low = least(event_scan.low, excluded.low),\n high = greatest(event_scan.high, excluded.high),\n updated_at = now()\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Int8",
"Int8"
]
},
"nullable": []
},
"hash": "ff1b7318c8148ad65900865dc4ac4eab80c191bed5bbd4bab95ef5f4f9c75709"
}
+20
View File
@@ -106,6 +106,26 @@ halves are one statement because every data-modifying CTE sees the same
snapshot, so the copy reads the pre-update row; splitting them would need an
explicit transaction to be equivalent.
**A block decoded against the wrong runtime must fail, not succeed.** The whole
metadata-oracle approach rests on `Runtime::decode_events` refusing a partial
read. Events are a `Vec<EventRecord>` and SCALE is self-describing only against
a registry, so a nearly-right registry decodes a nearly-right answer and leaves
bytes over. That leftover is the *only* signal an upgrade boundary was crossed —
it is what makes trying the tip's runtime speculatively safe, and what triggers
`runtime_at` to ask which runtime actually produced the block. Relaxing that
check to tolerate trailing bytes would turn a loud failure into silently wrong
rewards.
**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
progress from the rows makes the backfill re-read the same empty stretch
forever. `event_scan` holds one contiguous interval, and each batch is walked in
the direction that keeps it contiguous if it stops early: upward when extending
`high`, downward when extending `low`. Walking a downward batch upward leaves the
unread remainder *between* what was just read and `low`, which the cursor cannot
express and nothing would ever notice.
## Facts established by measurement
Taken from the live **Planck** chain, 2026-09-04. Don't re-derive or contradict
Generated
+267 -4
View File
@@ -88,6 +88,12 @@ version = "1.0.104"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470"
[[package]]
name = "arrayvec"
version = "0.7.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56"
[[package]]
name = "async-compression"
version = "0.4.44"
@@ -278,9 +284,14 @@ dependencies = [
"blackbeard-entities",
"blake2",
"chrono",
"frame-metadata",
"hex",
"parity-scale-codec",
"primitive-types",
"qp-poseidon-core",
"scale-info",
"scale-value",
"serde_json",
"thiserror 2.0.20",
"tracing",
]
@@ -294,6 +305,7 @@ dependencies = [
"blackbeard-entities",
"chrono",
"futures-util",
"hex",
"primitive-types",
"reqwest",
"serde",
@@ -341,6 +353,12 @@ version = "3.20.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649"
[[package]]
name = "byte-slice-cast"
version = "1.2.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7575182f7272186991736b70173b0ea045398f984bf5ebbb3804736ce1330c9d"
[[package]]
name = "bytemuck"
version = "1.25.2"
@@ -475,6 +493,27 @@ version = "0.9.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8"
[[package]]
name = "const_format"
version = "0.2.36"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4481a617ad9a412be3b97c5d403fef8ed023103368908b9c50af598ff467cc1e"
dependencies = [
"const_format_proc_macros",
"konst",
]
[[package]]
name = "const_format_proc_macros"
version = "0.2.34"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d57c2eccfb16dbac1f4e61e206105db5820c9d26c3c472bc17c774259ef7744"
dependencies = [
"proc-macro2",
"quote",
"unicode-xid",
]
[[package]]
name = "core-foundation"
version = "0.10.1"
@@ -581,6 +620,26 @@ dependencies = [
"zeroize",
]
[[package]]
name = "derive_more"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4a9b99b9cbbe49445b21764dc0625032a89b145a2642e67603e1c936f5458d05"
dependencies = [
"derive_more-impl",
]
[[package]]
name = "derive_more-impl"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cb7330aeadfbe296029522e6c40f315320aba36fc43a5b3632f3795348f3bd22"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.119",
]
[[package]]
name = "digest"
version = "0.10.7"
@@ -722,6 +781,17 @@ dependencies = [
"percent-encoding",
]
[[package]]
name = "frame-metadata"
version = "23.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9ba5be0edbdb824843a0f9c6f0906ecfc66c5316218d74457003218b24909ed0"
dependencies = [
"cfg-if",
"parity-scale-codec",
"scale-info",
]
[[package]]
name = "futures-channel"
version = "0.3.34"
@@ -1138,6 +1208,17 @@ dependencies = [
"icu_properties",
]
[[package]]
name = "impl-trait-for-tuples"
version = "0.2.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a0eb5a3343abf848c0984fe4604b2b105da9539376e24fc0a3b0007411ae4fd9"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.119",
]
[[package]]
name = "indexmap"
version = "2.14.1"
@@ -1183,6 +1264,21 @@ dependencies = [
"wasm-bindgen",
]
[[package]]
name = "konst"
version = "0.2.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "128133ed7824fcd73d6e7b17957c5eb7bacb885649bd8c69708b2331a10bcefb"
dependencies = [
"konst_macro_rules",
]
[[package]]
name = "konst_macro_rules"
version = "0.2.19"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a4933f3f57a8e9d9da04db23fb153356ecaf00cbd14aee46279c33dc80925c37"
[[package]]
name = "lazy_static"
version = "1.5.0"
@@ -1393,6 +1489,32 @@ version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe"
[[package]]
name = "parity-scale-codec"
version = "3.7.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "799781ae679d79a948e13d4824a40970bfa500058d245760dd857301059810fa"
dependencies = [
"arrayvec",
"byte-slice-cast",
"const_format",
"impl-trait-for-tuples",
"parity-scale-codec-derive",
"rustversion",
]
[[package]]
name = "parity-scale-codec-derive"
version = "3.7.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "34b4653168b563151153c9e4c08ebed57fb8262bebfa79711552fa983c623e7a"
dependencies = [
"proc-macro-crate",
"proc-macro2",
"quote",
"syn 2.0.119",
]
[[package]]
name = "parking"
version = "2.2.1"
@@ -1527,6 +1649,15 @@ dependencies = [
"uint",
]
[[package]]
name = "proc-macro-crate"
version = "3.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e67ba7e9b2b56446f1d419b1d807906278ffa1a658a8a5d8a39dcb1f5a78614f"
dependencies = [
"toml_edit 0.25.13+spec-1.1.0",
]
[[package]]
name = "proc-macro2"
version = "1.0.107"
@@ -1889,6 +2020,93 @@ version = "1.0.23"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
[[package]]
name = "scale-bits"
version = "0.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "27243ab0d2d6235072b017839c5f0cd1a3b1ce45c0f7a715363b0c7d36c76c94"
dependencies = [
"parity-scale-codec",
"scale-info",
"scale-type-resolver",
"serde",
]
[[package]]
name = "scale-decode"
version = "0.16.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8d6ed61699ad4d54101ab5a817169259b5b0efc08152f8632e61482d8a27ca3d"
dependencies = [
"parity-scale-codec",
"scale-bits",
"scale-type-resolver",
"smallvec",
"thiserror 2.0.20",
]
[[package]]
name = "scale-encode"
version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f2a976d73564a59e482b74fd5d95f7518b79ca8c8ca5865398a4d629dd15ee50"
dependencies = [
"parity-scale-codec",
"scale-bits",
"scale-type-resolver",
"smallvec",
"thiserror 2.0.20",
]
[[package]]
name = "scale-info"
version = "2.11.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "346a3b32eba2640d17a9cb5927056b08f3de90f65b72fe09402c2ad07d684d0b"
dependencies = [
"cfg-if",
"derive_more",
"parity-scale-codec",
"scale-info-derive",
]
[[package]]
name = "scale-info-derive"
version = "2.11.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c6630024bf739e2179b91fb424b28898baf819414262c5d376677dbff1fe7ebf"
dependencies = [
"proc-macro-crate",
"proc-macro2",
"quote",
"syn 2.0.119",
]
[[package]]
name = "scale-type-resolver"
version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0cded6518aa0bd6c1be2b88ac81bf7044992f0f154bfbabd5ad34f43512abcb"
dependencies = [
"scale-info",
"smallvec",
]
[[package]]
name = "scale-value"
version = "0.18.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b3b64809a541e8d5a59f7a9d67cc700cdf5d7f907932a83a0afdedc90db07ccb"
dependencies = [
"either",
"parity-scale-codec",
"scale-bits",
"scale-decode",
"scale-encode",
"scale-type-resolver",
"thiserror 2.0.20",
]
[[package]]
name = "schannel"
version = "0.1.29"
@@ -2565,8 +2783,8 @@ checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362"
dependencies = [
"serde",
"serde_spanned",
"toml_datetime",
"toml_edit",
"toml_datetime 0.6.11",
"toml_edit 0.22.27",
]
[[package]]
@@ -2578,6 +2796,15 @@ dependencies = [
"serde",
]
[[package]]
name = "toml_datetime"
version = "1.1.1+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7"
dependencies = [
"serde_core",
]
[[package]]
name = "toml_edit"
version = "0.22.27"
@@ -2587,9 +2814,30 @@ dependencies = [
"indexmap",
"serde",
"serde_spanned",
"toml_datetime",
"toml_datetime 0.6.11",
"toml_write",
"winnow",
"winnow 0.7.15",
]
[[package]]
name = "toml_edit"
version = "0.25.13+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6975367e4d2ef766d86af01ffad14b622fecc8d4357a998fbc4deb6e9bacaf9b"
dependencies = [
"indexmap",
"toml_datetime 1.1.1+spec-1.1.0",
"toml_parser",
"winnow 1.0.4",
]
[[package]]
name = "toml_parser"
version = "1.1.3+spec-1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d38ac1cf9b95face32296c0a3ede1fdc270627c9d9c02a7274dd6d960dc4d56"
dependencies = [
"winnow 1.0.4",
]
[[package]]
@@ -2844,6 +3092,12 @@ version = "0.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d"
[[package]]
name = "unicode-xid"
version = "0.2.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853"
[[package]]
name = "untrusted"
version = "0.9.0"
@@ -3265,6 +3519,15 @@ dependencies = [
"memchr",
]
[[package]]
name = "winnow"
version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "23b97319f7b8343df12cc98938e5c3eb436064524c8d2b4e30a1d3a36eecdf81"
dependencies = [
"memchr",
]
[[package]]
name = "wit-bindgen"
version = "0.46.0"
+8 -1
View File
@@ -22,6 +22,13 @@ clap = { version = "4", features = ["derive", "env"] }
figment = { version = "0.10", features = ["toml", "env"] }
futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
hex = "0.4"
# The chain's own description of itself, and a decoder driven by it. This is the
# alternative to hand-written SCALE readers that rot on every runtime upgrade,
# and to pulling in subxt or the runtime crate to avoid writing them.
frame-metadata = { version = "23", default-features = false, features = ["current", "decode"] }
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 }
# 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 }
@@ -34,7 +41,7 @@ serde_json = "1"
sqlx = { version = "0.8", default-features = false, features = [
"postgres",
"runtime-tokio-rustls",
"macros",
"macros", "json",
"migrate",
"chrono",
"bigdecimal",
+367 -1
View File
@@ -63,6 +63,53 @@ const ATTRIBUTION_GIVE_UP: Duration = Duration::from_secs(180);
/// Interval between difficulty and sync-state polls.
const POLL_INTERVAL: Duration = Duration::from_secs(4);
/// Events worth keeping, as `(pallet, variant)`.
///
/// The decoder in `blackbeard-core::runtime` reads *everything* the runtime
/// emits — it knows nothing about any particular event and needs no change to
/// handle a new one. This list is only about storage: most blocks carry an
/// `ExtrinsicSuccess` and a `DifficultyAdjusted` that nothing on the site asks
/// about, and at a one-second block time keeping them all is millions of rows a
/// month for no reader.
///
/// So adding transfers, or multisig, or anything the chain grows next, is an
/// entry here and a query — not a migration, a decoder, or a struct. `"*"`
/// takes every variant a pallet emits.
const INDEXED_EVENTS: &[(&str, &str)] = &[
// What a miner was actually paid, which is the one number the block header
// cannot tell us: the amount is remaining-supply over an emission divisor
// plus fees, quantized with dust carried between blocks, so it is only
// knowable from the event.
("MiningRewards", "*"),
// The other half of an account page. Indexed from the start because
// backfilling later costs the same as doing it now and history only grows.
("Balances", "Transfer"),
];
/// Whether an event is one we keep.
fn is_indexed(pallet: &str, variant: &str) -> bool {
INDEXED_EVENTS
.iter()
.any(|(p, v)| *p == pallet && (*v == "*" || *v == variant))
}
/// Heights read per backfill pass.
///
/// Each height is two RPC calls — `chain_getBlockHash` and one storage read —
/// so a pass is a bounded burst against the node rather than a flood, and the
/// pause between passes leaves ingest's own calls unimpeded. At this size a
/// chain a million blocks deep is read in about a day of background work, and
/// the newest blocks — the ones anything on the site actually shows — are
/// present within the first pass.
const BACKFILL_BATCH: u64 = 64;
/// Pause between backfill passes that did work.
const BACKFILL_TICK: Duration = Duration::from_millis(500);
/// Pause when there is nothing to do, or nothing can be done: history is fully
/// read, no runtime is known yet, or the node is not answering.
const BACKFILL_IDLE: Duration = Duration::from_secs(30);
/// Interval between attribution persists.
const PERSIST_INTERVAL: Duration = Duration::from_secs(30);
@@ -79,6 +126,7 @@ pub fn spawn(chain: Arc<ChainRuntime>, store: Store, config: &crate::config::Con
store.clone(),
config.telemetry.url.clone(),
));
tokio::spawn(backfill(Arc::clone(&chain), store.clone()));
tokio::spawn(housekeeping(chain, store, refresh));
}
@@ -443,6 +491,8 @@ async fn record(
None,
);
let _ = index_events(chain, store, height, &hash).await;
// Queue the attribution join. The feed needs a few seconds to collect
// reports from around the network before "who was first" means anything.
if chain.telemetry.connected() {
@@ -555,7 +605,16 @@ async fn poll(chain: Arc<ChainRuntime>, store: Store, telemetry_url: String) {
target_block_time: chain.spec.target_block_time_seconds,
}
};
if let Err(e) = store.upsert_chain(&record).await {
// After the chain row, never before: `runtime_metadata` references it,
// and on the very first poll the row does not exist yet. Caching then
// fails the foreign key while decoding still works from the RPC —
// which is the worst combination, because `spec_version` is set and
// the cache is never retried.
let chain_row_written = store.upsert_chain(&record).await;
if chain_row_written.is_ok() {
ensure_runtime(&chain, &store).await;
}
if let Err(e) = chain_row_written {
tracing::warn!(chain = %chain.id(), error = %e, "chain row not updated");
}
@@ -627,6 +686,313 @@ async fn housekeeping(chain: Arc<ChainRuntime>, store: Store, refresh: Duration)
}
}
/// What one attempt to index a block's events came to.
enum Indexed {
/// Read and stored, with the number of events kept.
Ok(usize),
/// The node has no events for this block and never will — pruned state, or
/// a runtime whose metadata cannot be obtained. Skipping it is permanent.
Unreadable,
/// A transient failure: an RPC that did not answer, a database that
/// refused. The caller must not treat this height as done.
Failed,
}
/// Decode a block's events and keep the ones anything asks for.
///
/// Best effort by design: a block whose events cannot be read costs the site
/// that block's rewards and nothing else. It must never stop ingest, because
/// the leaderboard is built from headers and does not depend on this at all.
async fn index_events(
chain: &Arc<ChainRuntime>,
store: &Store,
height: u64,
hash: &str,
) -> Indexed {
let blob = match chain.rpc.events_at(hash).await {
Ok(Some(b)) => b,
// Pruned, or a node that will not answer with state this old.
Ok(None) => return Indexed::Unreadable,
Err(e) => {
tracing::debug!(chain = %chain.id(), height, error = %e, "events unavailable");
return Indexed::Failed;
}
};
// The tip's runtime first, because it decodes all but a handful of blocks
// and costs no RPC call to identify. `decode_events` refuses a partial
// read, so a block from before an upgrade fails here rather than decoding
// into plausible nonsense — which is what makes trying it speculatively
// safe, and what the fallback below is for.
let decoded = match chain.read().current_runtime() {
Some(runtime) => runtime.decode_events(&blob).ok(),
None => None,
};
let decoded = match decoded {
Some(d) => d,
None => {
let Some(runtime) = runtime_at(chain, store, hash).await else {
return Indexed::Unreadable;
};
match runtime.decode_events(&blob) {
Ok(d) => d,
Err(e) => {
// The runtime the chain itself says produced this block
// cannot read it. That is this decoder's limitation, not a
// transient fault, so retrying forever would be a loop.
tracing::warn!(chain = %chain.id(), height, error = %e, "events did not decode");
return Indexed::Unreadable;
}
}
}
};
let keep: Vec<blackbeard_data::store::EventRecord> = decoded
.into_iter()
.filter(|e| is_indexed(&e.pallet, &e.variant))
.map(|e| blackbeard_data::store::EventRecord {
chain: chain.id(),
height,
event_index: e.index,
pallet: e.pallet,
variant: e.variant,
phase: e.phase,
extrinsic_index: e.extrinsic_index,
fields: e.fields,
})
.collect();
if keep.is_empty() {
return Indexed::Ok(0);
}
match store.record_events(&keep).await {
Ok(_) => Indexed::Ok(keep.len()),
Err(e) => {
tracing::warn!(chain = %chain.id(), height, error = %e, "events not persisted");
Indexed::Failed
}
}
}
/// Read every block's events from the tip back to genesis, and keep up.
///
/// Live ingest indexes each block as it arrives, which is enough for a feed and
/// not enough for an account page: the observer is restarted, the node is
/// briefly unreachable, a gap fill runs before the metadata is cached. Any of
/// those leaves a hole that nothing would ever return to.
///
/// The cursor is one contiguous interval (see the migration). Reaching the tip
/// takes priority over reaching genesis, because a stale tip is visible on the
/// site and a missing year of history is not — and because the tip keeps
/// moving, while genesis does not.
async fn backfill(chain: Arc<ChainRuntime>, store: Store) {
let id = chain.id();
loop {
let Some(tip) = chain.read().height else {
tokio::time::sleep(BACKFILL_IDLE).await;
continue;
};
// Nothing can be decoded before the first `ensure_runtime`, and the
// chain row the scan cursor references does not exist either.
if chain.read().current_runtime().is_none() {
tokio::time::sleep(BACKFILL_IDLE).await;
continue;
}
let scan = match store.event_scan(&id).await {
Ok(s) => s,
Err(e) => {
tracing::warn!(chain = %id, error = %e, "event scan cursor unreadable");
tokio::time::sleep(BACKFILL_IDLE).await;
continue;
}
};
// An empty cursor starts at the tip: the newest history is what the
// site shows first, and working outward from there means the useful
// half is present within seconds of a fresh database.
let (low, high) = scan.unwrap_or((tip, tip.saturating_sub(1)));
// Reaching the tip first, then genesis. Each batch is walked in the
// direction that keeps the interval contiguous if it stops early:
// upward when extending `high`, downward when extending `low`. Walking
// a downward batch upward would leave the unread remainder *between*
// what was just read and `low`, which the cursor cannot express.
let (from, to, descending) = if high < tip {
(high + 1, tip.min(high + BACKFILL_BATCH), false)
} else if low > 1 {
(low - 1, low.saturating_sub(BACKFILL_BATCH).max(1), true)
} else {
// Genesis reached and the tip is current: there is nothing left to
// read until the chain produces more.
tokio::time::sleep(BACKFILL_IDLE).await;
continue;
};
let heights: Vec<u64> = if descending {
(to..=from).rev().collect()
} else {
(from..=to).collect()
};
let mut reached = None;
let mut kept = 0usize;
for height in heights {
let hash = match chain.rpc.block_hash(height).await {
Ok(Some(h)) => h,
// A height the node does not have is not going to appear, and
// stopping here would wedge the scan against it forever.
Ok(None) => {
reached = Some(height);
continue;
}
Err(e) => {
tracing::debug!(chain = %id, height, error = %e, "block hash unavailable");
break;
}
};
match index_events(&chain, &store, height, &hash).await {
Indexed::Ok(n) => kept += n,
// Permanently unreadable: claim it, or the scan stops here for
// good against a block that will never decode.
Indexed::Unreadable => {}
// Transient. Leave it unclaimed and come back to it.
Indexed::Failed => break,
}
reached = Some(height);
}
let Some(reached) = reached else {
// Not one height moved. Backing off keeps a node that is down or
// still syncing from being asked once per block interval.
tokio::time::sleep(BACKFILL_IDLE).await;
continue;
};
// Only the stretch actually read is claimed; `set_event_scan` widens
// and never narrows, so passing the unchanged end is a no-op.
let (low, high) = if descending {
(reached, high)
} else {
(low.min(from), reached)
};
if let Err(e) = store.set_event_scan(&id, low, high).await {
tracing::warn!(chain = %id, error = %e, "event scan cursor not saved");
}
if kept > 0 {
tracing::debug!(chain = %id, low, high, kept, "events backfilled");
}
tokio::time::sleep(BACKFILL_TICK).await;
}
}
/// Make sure the metadata for the runtime now in force is cached and parsed.
///
/// Called from the poll loop rather than per block: `spec_version` changes only
/// on a runtime upgrade — four times in a million blocks on Heisenberg — so
/// asking every four seconds is already far more often than it can change,
/// and asking per block would double the RPC traffic of ingest.
async fn ensure_runtime(chain: &Arc<ChainRuntime>, store: &Store) {
let Ok(version) = chain.rpc.runtime_version(None).await else {
return;
};
if chain.read().spec_version == Some(version.spec_version) {
return;
}
if load_runtime(chain, store, &version, None).await.is_some() {
chain.write().spec_version = Some(version.spec_version);
tracing::info!(
chain = %chain.id(), spec = version.spec_version, spec_name = %version.spec_name,
"decoding against runtime"
);
}
}
/// The runtime that produced a particular block.
///
/// One extra RPC call to learn which version that is, so it is only reached
/// when the tip's runtime has already failed to read the block — which happens
/// once per upgrade boundary the backfill crosses, not once per block.
async fn runtime_at(
chain: &Arc<ChainRuntime>,
store: &Store,
hash: &str,
) -> Option<Arc<blackbeard_core::runtime::Runtime>> {
let version = chain.rpc.runtime_version(Some(hash)).await.ok()?;
if let Some(runtime) = chain.read().runtimes.get(&version.spec_version).cloned() {
return Some(runtime);
}
load_runtime(chain, store, &version, Some(hash)).await
}
/// Obtain and parse one runtime's metadata, preferring the cache.
///
/// Metadata for a given `spec_version` never changes, and the node may not have
/// the state to regenerate it once it has pruned — so a hit here is not merely
/// faster, it is sometimes the only way the answer still exists.
async fn load_runtime(
chain: &Arc<ChainRuntime>,
store: &Store,
version: &blackbeard_data::rpc::RuntimeVersion,
at: Option<&str>,
) -> Option<Arc<blackbeard_core::runtime::Runtime>> {
let id = chain.id();
let cached = store
.runtime_metadata(&id, version.spec_version)
.await
.ok()
.flatten();
let raw = match cached {
Some(bytes) => bytes,
None => match chain.rpc.metadata_at(at).await {
Ok(bytes) => {
let height = chain.read().height.unwrap_or(0);
match store
.cache_runtime_metadata(
&id,
version.spec_version,
&version.spec_name,
&bytes,
height,
)
.await
{
Ok(true) => tracing::info!(
chain = %id, spec = version.spec_version, bytes = bytes.len(),
"runtime metadata cached"
),
Ok(false) => {}
Err(e) => {
tracing::warn!(chain = %id, error = %e, "runtime metadata not cached")
}
}
bytes
}
Err(e) => {
tracing::warn!(chain = %id, error = %e, "runtime metadata unavailable");
return None;
}
},
};
match blackbeard_core::runtime::Runtime::from_metadata(&raw) {
Ok(runtime) => {
let runtime = Arc::new(runtime);
chain
.write()
.runtimes
.insert(version.spec_version, Arc::clone(&runtime));
Some(runtime)
}
Err(e) => {
tracing::warn!(chain = %id, error = %e, "runtime metadata did not parse");
None
}
}
}
/// Join settled blocks with the feed's first-import reports.
fn resolve_attributions(chain: &Arc<ChainRuntime>) {
let now = tokio::time::Instant::now();
+18
View File
@@ -21,6 +21,7 @@ use std::sync::{Arc, RwLock};
use blackbeard_core::attribution::Attributor;
use blackbeard_core::hashrate::{self, Interval};
use blackbeard_core::runtime::Runtime;
use blackbeard_core::window::RollingWindow;
use blackbeard_data::rpc::RpcClient;
use blackbeard_data::store::Store;
@@ -80,6 +81,12 @@ pub struct ChainInner {
pub token_symbol: Option<String>,
/// Token decimals.
pub token_decimals: Option<u8>,
/// Parsed metadata per `spec_version`, so a block is decoded against the
/// runtime that produced it without re-parsing 100 KiB every time. Small:
/// one entry per runtime upgrade the observer has seen.
pub runtimes: HashMap<u32, Arc<Runtime>>,
/// The runtime the chain is on now, as of the last poll.
pub spec_version: Option<u32>,
/// Current difficulty.
pub difficulty: Option<U512>,
/// Ceiling difficulty.
@@ -125,6 +132,15 @@ pub struct PendingAttribution {
pub give_up: tokio::time::Instant,
}
impl ChainInner {
/// The parsed runtime for the version now in force, if its metadata has
/// been fetched. `None` before the first poll, and after a decode failure
/// clears `spec_version` to force a re-check.
pub fn current_runtime(&self) -> Option<Arc<Runtime>> {
self.runtimes.get(&self.spec_version?).cloned()
}
}
/// One chain, live.
#[derive(Debug)]
pub struct ChainRuntime {
@@ -165,6 +181,8 @@ impl ChainRuntime {
status: ChainStatus::Awaiting,
token_symbol: None,
token_decimals: None,
runtimes: HashMap::new(),
spec_version: None,
difficulty: None,
max_difficulty: None,
syncing: false,
+5
View File
@@ -12,7 +12,12 @@ repository.workspace = true
blackbeard-entities.workspace = true
chrono.workspace = true
blake2.workspace = true
frame-metadata.workspace = true
hex.workspace = true
parity-scale-codec.workspace = true
scale-info.workspace = true
scale-value.workspace = true
serde_json.workspace = true
qp-poseidon-core.workspace = true
primitive-types.workspace = true
thiserror.workspace = true
+1
View File
@@ -16,6 +16,7 @@
pub mod attribution;
pub mod digest;
pub mod hashrate;
pub mod runtime;
pub mod scale;
pub mod series;
pub mod window;
+449
View File
@@ -0,0 +1,449 @@
//! Decoding a chain's own data using the chain's own description of itself.
//!
//! ## Why this exists rather than a hand-written decoder
//!
//! An earlier attempt at this problem (`qsafe.af`) generated pallet and event
//! *index* tables from the runtime and then hand-wrote a SCALE reader for every
//! event's fields. It worked and it was miserable: every new event was a code
//! change, and Quantus had not settled its encodings, so the hand-written half
//! kept rotting underneath. On the chain we can watch that happen —
//! `transactionVersion` went 2 → 3 → 6 across four runtime upgrades, and every
//! one of those is an extrinsic-format change that silently breaks a decoder
//! written against the previous one.
//!
//! Metadata v14 removes the need for the hand-written half entirely: it carries
//! a complete `scale-info` type registry describing the *structure* of every
//! event, call and storage item, not merely their names. Given the metadata for
//! a block, anything in that block can be decoded without knowing anything about
//! it in advance. Nothing here mentions `MinerRewarded`, or any other event.
//!
//! ## Where the metadata comes from
//!
//! `state_getMetadata` takes a block hash and makes the node run
//! `Metadata_metadata` against the runtime code in *that block's* state. So the
//! runtime WASM is the oracle, and the node executes it for us. Verified across
//! four upgrade boundaries on Heisenberg: v126, v131, v136 and v148 each return
//! a different blob.
//!
//! That does depend on the node still holding state for the block, which is why
//! `blackbeard-data` caches each blob by `spec_version` — one fetch per runtime
//! version, ever, and decoding stays possible after the node has pruned.
use std::collections::BTreeMap;
use frame_metadata::v14::RuntimeMetadataV14;
use frame_metadata::{RuntimeMetadata, RuntimeMetadataPrefixed};
use parity_scale_codec::Decode;
use scale_info::form::PortableForm;
/// Failures decoding a runtime's description or its data.
#[derive(Debug, thiserror::Error, PartialEq)]
pub enum RuntimeError {
/// The blob is not SCALE-encoded prefixed metadata.
#[error("metadata did not decode")]
Malformed,
/// Metadata this decoder does not read. v14 is what every Quantus runtime
/// observed so far emits; a chain that moved to v15/v16 needs this widened
/// deliberately rather than silently mis-read.
#[error("unsupported metadata version {0}")]
UnsupportedVersion(u8),
/// The runtime does not describe where its events live.
#[error("runtime declares no System::Events storage entry")]
NoEventsEntry,
/// A value did not match the type the registry said it would.
#[error("decoding against the registry failed: {0}")]
Decode(String),
}
/// One decoded event, named as the runtime names it.
#[derive(Debug, Clone, PartialEq)]
pub struct DecodedEvent {
/// Position within the block's event list.
pub index: u32,
/// Emitting pallet, e.g. `MiningRewards`.
pub pallet: String,
/// Event variant, e.g. `MinerRewarded`.
pub variant: String,
/// When in the block it fired: `ApplyExtrinsic(n)`, `Finalization`,
/// `Initialization`. Kept because "which extrinsic caused this" is the
/// join a transaction view needs, and it is free here.
pub phase: String,
/// The extrinsic index, when the phase is `ApplyExtrinsic`.
pub extrinsic_index: Option<u32>,
/// Fields, by name, normalised for storage — see [`normalise`].
pub fields: serde_json::Value,
}
/// A runtime's description of itself, ready to decode with.
pub struct Runtime {
metadata: RuntimeMetadataV14,
events_ty: u32,
/// Type ids whose registry path ends in a name worth rendering specially.
/// Resolved once here rather than looked up per value.
account_tys: BTreeMap<u32, ()>,
}
impl std::fmt::Debug for Runtime {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Runtime")
.field("pallets", &self.metadata.pallets.len())
.field("types", &self.metadata.types.types.len())
.finish()
}
}
impl Runtime {
/// Parse metadata as `state_getMetadata` returns it.
pub fn from_metadata(raw: &[u8]) -> Result<Self, RuntimeError> {
let prefixed =
RuntimeMetadataPrefixed::decode(&mut &raw[..]).map_err(|_| RuntimeError::Malformed)?;
let metadata = match prefixed.1 {
RuntimeMetadata::V14(v) => v,
other => return Err(RuntimeError::UnsupportedVersion(version_of(&other))),
};
// Where the events live, asked of the runtime rather than assumed. The
// storage entry's declared value type is
// `Vec<EventRecord<RuntimeEvent, Hash>>`, so this one id is enough to
// read the whole blob.
let events_ty = metadata
.pallets
.iter()
.find(|p| p.name == "System")
.and_then(|p| p.storage.as_ref())
.and_then(|s| s.entries.iter().find(|e| e.name == "Events"))
.and_then(|e| match &e.ty {
frame_metadata::v14::StorageEntryType::Plain(t) => Some(t.id),
// A map would mean the runtime keeps events per key, which no
// FRAME runtime does; refuse rather than guess at a shape.
frame_metadata::v14::StorageEntryType::Map { .. } => None,
})
.ok_or(RuntimeError::NoEventsEntry)?;
// Account ids are 32 raw bytes in the registry and unreadable as such.
// They are found by their registry path, not by length: a block hash is
// also 32 bytes, and rendering one as an address would be a lie.
let account_tys = metadata
.types
.types
.iter()
.filter(|t| {
t.ty.path
.segments
.last()
.is_some_and(|s| s == "AccountId32")
})
.map(|t| (t.id, ()))
.collect();
Ok(Self {
metadata,
events_ty,
account_tys,
})
}
/// Pallets this runtime has, by index.
pub fn pallets(&self) -> impl Iterator<Item = (u8, &str)> {
self.metadata
.pallets
.iter()
.map(|p| (p.index, p.name.as_str()))
}
/// Decode a `System::Events` blob into named events.
///
/// Takes the whole blob and refuses a partial read: bytes left over mean
/// the registry and the data disagree, which is exactly the situation —
/// decoding one runtime's block against another's metadata — that this
/// module exists to make impossible. Better to fail than to store half a
/// 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()))?;
if !cursor.is_empty() {
return Err(RuntimeError::Decode(format!(
"{} trailing bytes; metadata does not match this block",
cursor.len()
)));
}
let records = match value.value {
scale_value::ValueDef::Composite(scale_value::Composite::Unnamed(v)) => v,
other => {
return Err(RuntimeError::Decode(format!(
"events blob is not a sequence: {other:?}"
)));
}
};
let mut out = Vec::with_capacity(records.len());
for (index, record) in records.into_iter().enumerate() {
let Some(event) = self.event_of(&record) else {
continue;
};
let (phase, extrinsic_index) = phase_of(&record);
out.push(DecodedEvent {
index: index as u32,
pallet: event.0,
variant: event.1,
phase,
extrinsic_index,
fields: event.2,
});
}
Ok(out)
}
/// Pull `(pallet, variant, fields)` out of one `EventRecord`.
fn event_of(
&self,
record: &scale_value::Value<u32>,
) -> Option<(String, String, serde_json::Value)> {
let event = field(record, "event")?;
// RuntimeEvent is an enum of pallets; each pallet's is an enum of its
// events. Two variants deep, both named by the runtime.
let scale_value::ValueDef::Variant(pallet) = &event.value else {
return None;
};
let inner = match &pallet.values {
scale_value::Composite::Unnamed(v) => v.first()?,
scale_value::Composite::Named(v) => v.first().map(|(_, v)| v)?,
};
let scale_value::ValueDef::Variant(variant) = &inner.value else {
return None;
};
Some((
pallet.name.clone(),
variant.name.clone(),
self.normalise(&scale_value::Value {
value: scale_value::ValueDef::Composite(variant.values.clone()),
context: inner.context,
}),
))
}
/// Turn a decoded value into JSON fit for storage and a browser.
///
/// Two things are not left as the registry produces them:
///
/// - **Account ids** come through as thirty-two separate numbers. They are
/// rendered as `0x…` hex, identified by their registry path so a block
/// hash of the same length is not mistaken for one. Callers that want an
/// SS58 address derive it from the hex; this module does not assume a
/// network prefix.
/// - **Integers** beyond 2^53 are rendered as strings. Balances on this
/// chain are `u128` and routinely past it, and `serde_json` would put
/// them on the wire as JSON numbers for a browser to silently round —
/// the same trap `BigUintDec` exists to avoid for difficulty.
fn normalise(&self, value: &scale_value::Value<u32>) -> serde_json::Value {
use scale_value::{Composite, Primitive, ValueDef};
match &value.value {
ValueDef::Composite(c) => {
if self.account_tys.contains_key(&value.context)
&& let Some(bytes) = as_bytes(c)
{
return serde_json::Value::String(format!("0x{}", hex::encode(bytes)));
}
match c {
Composite::Named(fields) => serde_json::Value::Object(
fields
.iter()
.map(|(k, v)| (k.clone(), self.normalise(v)))
.collect(),
),
Composite::Unnamed(items) => {
// A one-element tuple is a newtype wrapper and carries
// no information; unwrap it so `{"reward": 300}` does
// not become `{"reward": [300]}`.
if items.len() == 1 {
self.normalise(&items[0])
} else if let Some(bytes) = as_bytes(c) {
serde_json::Value::String(format!("0x{}", hex::encode(bytes)))
} else {
serde_json::Value::Array(
items.iter().map(|v| self.normalise(v)).collect(),
)
}
}
}
}
ValueDef::Variant(v) => {
if matches!(&v.values, Composite::Unnamed(u) if u.is_empty())
|| matches!(&v.values, Composite::Named(n) if n.is_empty())
{
// A fieldless variant is just its name.
return serde_json::Value::String(v.name.clone());
}
serde_json::json!({
v.name.clone(): self.normalise(&scale_value::Value {
value: ValueDef::Composite(v.values.clone()),
context: value.context,
})
})
}
ValueDef::BitSequence(bits) => {
serde_json::Value::String(bits.iter().map(|b| if b { '1' } else { '0' }).collect())
}
ValueDef::Primitive(p) => match p {
Primitive::Bool(b) => serde_json::Value::Bool(*b),
Primitive::Char(c) => serde_json::Value::String(c.to_string()),
Primitive::String(s) => serde_json::Value::String(s.clone()),
Primitive::U128(n) => big(*n),
Primitive::I128(n) => {
if n.unsigned_abs() > SAFE_INTEGER {
serde_json::Value::String(n.to_string())
} else {
serde_json::Value::Number((*n as i64).into())
}
}
Primitive::U256(b) | Primitive::I256(b) => {
serde_json::Value::String(format!("0x{}", hex::encode(b)))
}
},
}
}
}
/// Above this, a JSON number is not safely representable in a browser.
const SAFE_INTEGER: u128 = 9_007_199_254_740_991;
fn big(n: u128) -> serde_json::Value {
if n > SAFE_INTEGER {
serde_json::Value::String(n.to_string())
} else {
serde_json::Value::Number((n as u64).into())
}
}
/// A composite that is really a byte string.
fn as_bytes(c: &scale_value::Composite<u32>) -> Option<Vec<u8>> {
let items = match c {
scale_value::Composite::Unnamed(v) => v,
scale_value::Composite::Named(_) => return None,
};
// A single unnamed field is a newtype; look inside it.
if items.len() == 1
&& let scale_value::ValueDef::Composite(inner) = &items[0].value
{
return as_bytes(inner);
}
if items.is_empty() {
return None;
}
items
.iter()
.map(|v| match &v.value {
scale_value::ValueDef::Primitive(scale_value::Primitive::U128(n)) if *n < 256 => {
Some(*n as u8)
}
_ => None,
})
.collect()
}
/// A named field of a composite value.
fn field<'a>(
value: &'a scale_value::Value<u32>,
name: &str,
) -> Option<&'a scale_value::Value<u32>> {
match &value.value {
scale_value::ValueDef::Composite(scale_value::Composite::Named(fields)) => {
fields.iter().find(|(k, _)| k == name).map(|(_, v)| v)
}
_ => None,
}
}
/// `(phase name, extrinsic index)` for an event record.
fn phase_of(record: &scale_value::Value<u32>) -> (String, Option<u32>) {
let Some(phase) = field(record, "phase") else {
return ("Unknown".into(), None);
};
let scale_value::ValueDef::Variant(v) = &phase.value else {
return ("Unknown".into(), None);
};
let index = match &v.values {
scale_value::Composite::Unnamed(items) => items.first().and_then(|i| match &i.value {
scale_value::ValueDef::Primitive(scale_value::Primitive::U128(n)) => Some(*n as u32),
_ => None,
}),
scale_value::Composite::Named(_) => None,
};
(v.name.clone(), index)
}
fn version_of(md: &RuntimeMetadata) -> u8 {
match md {
RuntimeMetadata::V14(_) => 14,
RuntimeMetadata::V15(_) => 15,
_ => 0,
}
}
/// Marker so the unused-import lint does not fire on the form import, which
/// documents that every type here is the portable (self-describing) form.
type _Portable = PortableForm;
#[cfg(test)]
mod tests {
use super::*;
/// Metadata captured from mainnet at spec_version 152.
fn metadata() -> Vec<u8> {
let hex = include_str!("../tests/mainnet-v152.metadata.hex");
hex::decode(hex.trim().trim_start_matches("0x")).expect("fixture is hex")
}
#[test]
fn a_runtime_describes_its_own_pallets() {
let rt = Runtime::from_metadata(&metadata()).expect("parses");
let names: Vec<&str> = rt.pallets().map(|(_, n)| n).collect();
assert!(names.contains(&"MiningRewards"), "{names:?}");
assert!(names.contains(&"System"));
}
#[test]
fn nonsense_is_refused_rather_than_guessed_at() {
assert_eq!(
Runtime::from_metadata(&[1, 2, 3]).unwrap_err(),
RuntimeError::Malformed
);
}
/// The whole point: an event nothing in this crate has heard of decodes
/// into its own names, straight from the registry.
#[test]
fn events_decode_with_their_runtime_given_names() {
let rt = Runtime::from_metadata(&metadata()).expect("parses");
let blob = hex::decode(
include_str!("../tests/mainnet-events.hex")
.trim()
.trim_start_matches("0x"),
)
.expect("fixture is hex");
let events = rt.decode_events(&blob).expect("decodes");
assert!(!events.is_empty());
let reward = events
.iter()
.find(|e| e.pallet == "MiningRewards" && e.variant == "MinerRewarded")
.expect("the fixture block paid its author");
// An account renders as hex, not as thirty-two numbers.
let miner = reward.fields["miner"].as_str().expect("miner is a string");
assert!(miner.starts_with("0x") && miner.len() == 66, "{miner}");
// A balance survives as an exact value rather than a rounded float.
assert!(reward.fields["reward"].is_number() || reward.fields["reward"].is_string());
}
#[test]
fn a_block_decoded_against_the_wrong_runtime_fails_loudly() {
// Trailing bytes mean the registry and the data disagree. Storing a
// partial read would be worse than storing nothing.
let rt = Runtime::from_metadata(&metadata()).expect("parses");
let err = rt.decode_events(&[0xff; 16]).unwrap_err();
assert!(matches!(err, RuntimeError::Decode(_)), "{err:?}");
}
}
+91
View File
@@ -104,6 +104,66 @@ fn base58(input: &[u8]) -> String {
out
}
/// The account id behind an SS58 address, as `0x` hex.
///
/// The inverse of [`ss58`], and the one function an account route needs: the
/// URL carries the address a person recognises, while every stored field —
/// `chain_event.fields`, a transfer's `from` and `to` — carries the raw 32
/// bytes. Doing the conversion here rather than at the query means a
/// mistyped address is rejected once, at the edge, instead of silently
/// matching nothing.
///
/// `None` unless the string is a well-formed address *for this network*: the
/// checksum must verify and the prefix must be [`SS58_PREFIX`]. Accepting a
/// foreign prefix would let a Polkadot address address a Quantus account,
/// which is the same account id wearing someone else's network, and would show
/// a stranger a page of holdings that are not theirs.
pub fn account_id(address: &str) -> Option<String> {
let raw = base58_decode(address)?;
// Two prefix bytes for a prefix above 63, then 32 payload and 2 checksum.
if raw.len() != 36 || (raw[0] & 0b0100_0000) == 0 {
return None;
}
let prefix = ((raw[0] as u16 & 0b0011_1111) << 2) | ((raw[1] as u16 & 0b1100_0000) >> 6);
let prefix = prefix | ((raw[1] as u16 & 0b0011_1111) << 8);
if prefix != SS58_PREFIX {
return None;
}
use blake2::digest::consts::U64;
use blake2::{Blake2b, Digest};
let mut hasher = Blake2b::<U64>::new();
hasher.update(b"SS58PRE");
hasher.update(&raw[..34]);
if hasher.finalize()[..2] != raw[34..] {
return None;
}
Some(format!("0x{}", hex::encode(&raw[2..34])))
}
/// Base58 decode: the inverse of [`base58`], and equally little arithmetic.
fn base58_decode(input: &str) -> Option<Vec<u8>> {
let mut bytes: Vec<u8> = Vec::with_capacity(input.len());
for c in input.chars() {
let value = ALPHABET.iter().position(|a| *a as char == c)? as u32;
let mut carry = value;
for byte in bytes.iter_mut() {
carry += *byte as u32 * 58;
*byte = (carry & 0xFF) as u8;
carry >>= 8;
}
while carry > 0 {
bytes.push((carry & 0xFF) as u8);
carry >>= 8;
}
}
// Each leading '1' is one leading zero byte, which the arithmetic drops.
let leading = input.chars().take_while(|c| *c == '1').count();
bytes.extend(std::iter::repeat_n(0u8, leading));
bytes.reverse();
Some(bytes)
}
/// The address in the form a table cell can hold, keeping both ends so two
/// addresses cannot be confused by their prefix alone.
pub fn abbreviated_address(address: &str) -> String {
@@ -151,6 +211,37 @@ mod tests {
}
}
#[test]
fn an_address_decodes_back_to_the_account_it_encodes() {
let miner =
MinerId("0x134e73f06fa9bdb1dbfa909e149c563f5860ceb71a0e7307918f7033970edf59".into());
let address = reward_address(&miner).unwrap();
let account = account_id(&address).unwrap();
assert_eq!(account.len(), 66);
assert_eq!(ss58_roundtrip(&account), address);
}
fn ss58_roundtrip(account_hex: &str) -> String {
let mut account = [0u8; 32];
hex::decode_to_slice(account_hex.trim_start_matches("0x"), &mut account).unwrap();
ss58(&account, SS58_PREFIX)
}
/// The address must be rejected unless it is an address *here*. The last
/// case is a real Polkadot address: the same 32 bytes, a different network,
/// and accepting it would show one account's history under another's name.
#[test]
fn a_foreign_or_damaged_address_is_declined() {
for bad in [
"",
"not base58 at all 0OIl",
"qzp2AxZw8szXc5qDe1aU8P5kt1dkf1JFYktvV7N6cVbFZrs8M",
"15oF4uVJwmo4TdGW7VfQxNLavjCXviqxT9S1MgbjMNHr6Sp5",
] {
assert_eq!(account_id(bad), None, "{bad}");
}
}
#[test]
fn abbreviation_keeps_both_ends() {
let a = abbreviated_address("qzp2AxZw8szXc5qDe1aU8P5kt1dkf1JFYktvV7N6cVbFZrs8L");
@@ -0,0 +1 @@
0x1800000000000000a2469cdb551702000001050113730827172e000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000002154ede91c2e000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000002f050000000000000001020ac6801725b054b06a3d5030f25caa3f2e9094d7722efe4f9eddbce1227798ec7c00b864d945000000000000000000000000011500563c000000000000000114000101010101010101010101010101010101010101010101010101010101010101c6801725b054b06a3d5030f25caa3f2e9094d7722efe4f9eddbce1227798ec7c00b864d9450000000000000000000000ba2e000000000000563c00000000000000010600c6801725b054b06a3d5030f25caa3f2e9094d7722efe4f9eddbce1227798ec7c00b864d945000000000000000000000000
File diff suppressed because one or more lines are too long
+1
View File
@@ -15,6 +15,7 @@ blackbeard-entities.workspace = true
bigdecimal.workspace = true
chrono.workspace = true
futures-util.workspace = true
hex.workspace = true
primitive-types.workspace = true
reqwest.workspace = true
serde.workspace = true
@@ -0,0 +1,96 @@
-- The chain's description of itself, and what it said happened.
--
-- ## Why the metadata is stored rather than fetched when needed
--
-- `state_getMetadata` takes a block hash and makes the node execute
-- `Metadata_metadata` against the runtime code in that block's state, so the
-- runtime WASM is the oracle and the node does the executing. That is exactly
-- what is wanted — decoding a block against the runtime that produced it — but
-- it only works while the node still holds state for the block. On this fleet
-- that is currently every block back to genesis; it is not a guarantee, and a
-- node that starts pruning would make every historical block undecodable.
--
-- One row per runtime version removes that dependency permanently. Metadata is
-- ~100-150 KiB and changes only on a runtime upgrade — Heisenberg has had four
-- in a million blocks — so this table stays in single figures for years while
-- making decoding reproducible long after the state behind it is gone.
create table runtime_metadata (
chain text not null references chain (id) on delete cascade,
spec_version bigint not null,
spec_name text not null,
-- Exactly the bytes `state_getMetadata` returned, unparsed. Storing the
-- parsed form would bake in this decoder's understanding of v14; storing
-- the blob means a later version of the code can reinterpret history.
metadata bytea not null,
first_seen_height bigint not null,
fetched_at timestamptz not null default now(),
primary key (chain, spec_version)
);
-- Decoded events, in the runtime's own vocabulary.
--
-- ## Why one table rather than one per thing we care about
--
-- A `miner_reward` table would need a sibling for transfers, another for
-- multisig, another for whatever the chain grows next — each a migration, a
-- struct and a query. The runtime already names everything it emits, and
-- metadata v14 describes the shape of every field, so nothing here needs to
-- know what a reward *is*: `pallet`, `variant` and the fields as jsonb hold any
-- event the chain has or will have. Adding transfers to the site is a query
-- and an interest-list entry, not a schema change.
--
-- Fields are normalised on the way in (see `blackbeard-core::runtime`):
-- account ids as `0x` hex rather than thirty-two separate numbers, and integers
-- past 2^53 as strings, because balances here are u128 and a JSON number would
-- be silently rounded by the browser — the same trap `BigUintDec` exists for.
create table chain_event (
chain text not null references chain (id) on delete cascade,
height bigint not null,
-- Position within the block's event list. With height it is the natural
-- key, and it makes re-indexing a block idempotent.
event_index integer not null,
pallet text not null,
variant text not null,
-- `ApplyExtrinsic`, `Finalization`, `Initialization`. Kept because "which
-- extrinsic caused this" is the join a transaction view needs, and the
-- decoder gets it for free.
phase text not null,
extrinsic_index integer,
fields jsonb not null,
primary key (chain, height, event_index)
);
-- "Every MinerRewarded, newest first" — the shape every feed on the site wants.
create index chain_event_kind_idx
on chain_event (chain, pallet, variant, height desc);
-- "Everything involving this account". `jsonb_path_ops` rather than the default
-- operator class: it indexes only containment, which is the single operator
-- these queries use, and is markedly smaller and faster for it. This is what
-- lets an account page work without a per-event-type table.
create index chain_event_fields_idx
on chain_event using gin (fields jsonb_path_ops);
-- How much of a chain's history has actually been read.
--
-- ## Why a cursor rather than inferring it from `chain_event`
--
-- "The highest height with an event" is not the same question. A block with no
-- indexed event is indistinguishable from one never looked at, and both are
-- ordinary: the interest list is deliberately narrow, so most blocks on most
-- chains will contribute nothing. Inferring progress from the rows would make
-- the backfill re-read the same empty stretch forever.
--
-- One contiguous interval rather than a set of ranges: the backfill extends
-- `high` toward the tip and `low` toward genesis, one batch at a time, and a
-- single interval cannot develop the holes a per-batch record would need
-- coalescing to avoid. Blocks indexed opportunistically by live ingest outside
-- the interval are simply re-read when the interval reaches them, which costs
-- one RPC call and writes nothing new — `chain_event` is keyed on
-- `(chain, height, event_index)`, so re-indexing a block is idempotent.
create table event_scan (
chain text primary key references chain (id) on delete cascade,
low bigint not null,
high bigint not null,
updated_at timestamptz not null default now()
);
+69
View File
@@ -57,6 +57,26 @@ impl Header {
}
}
/// `twox_128("System") ++ twox_128("Events")`, the storage key every FRAME
/// runtime keeps its per-block events under. Constant across runtimes because
/// it is a hash of the names, not of anything version-specific.
const SYSTEM_EVENTS_KEY: &str =
"0x26aa394eea5630e07c48ae0c9558cef780d41e5e16056765bc8461851072c9d7";
/// Which runtime produced a block.
#[derive(Debug, Clone, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RuntimeVersion {
/// e.g. `quantus-runtime`.
pub spec_name: String,
/// Bumped on every runtime upgrade. The cache key for metadata.
pub spec_version: u32,
/// Bumped when the extrinsic format changes — 2 to 3 to 6 across
/// Heisenberg's upgrades, and every one of those would break a decoder
/// written against the previous format.
pub transaction_version: u32,
}
/// A block body, reduced to what the observer reads from it.
#[derive(Debug, Clone, Copy)]
pub struct BlockBody {
@@ -182,6 +202,55 @@ impl RpcClient {
Ok(self.body(hash).await?.and_then(|b| b.timestamp_ms))
}
/// The runtime's own description of itself, as of `hash`.
///
/// The node executes `Metadata_metadata` against the runtime code in that
/// block's state, so this is the WASM answering — the only source that is
/// right across a runtime upgrade. Needs the state, which a pruning node
/// will not have for old blocks; the caller caches the result per
/// `spec_version` so that only matters once.
pub async fn metadata_at(&self, hash: Option<&str>) -> Result<Vec<u8>, DataError> {
let params = match hash {
Some(h) => json!([h]),
None => json!([]),
};
let v = self.call("state_getMetadata", params).await?;
let hex = v.as_str().ok_or_else(|| DataError::Rpc {
method: "state_getMetadata".into(),
message: "metadata was not a hex string".into(),
})?;
hex::decode(hex.strip_prefix("0x").unwrap_or(hex)).map_err(|e| DataError::Rpc {
method: "state_getMetadata".into(),
message: format!("metadata was not hex: {e}"),
})
}
/// Which runtime produced a block. `spec_version` changes only on an
/// upgrade, which is the signal to fetch metadata again.
pub async fn runtime_version(&self, hash: Option<&str>) -> Result<RuntimeVersion, DataError> {
let params = match hash {
Some(h) => json!([h]),
None => json!([]),
};
Ok(serde_json::from_value(
self.call("state_getRuntimeVersion", params).await?,
)?)
}
/// The raw `System::Events` blob for a block.
///
/// `Ok(None)` where the node has pruned the state it lives in — an answer,
/// not a failure, and the reason metadata is cached rather than re-fetched.
pub async fn events_at(&self, hash: &str) -> Result<Option<Vec<u8>>, DataError> {
let v = self
.call("state_getStorage", json!([SYSTEM_EVENTS_KEY, hash]))
.await?;
let Some(hex) = v.as_str() else {
return Ok(None);
};
Ok(hex::decode(hex.strip_prefix("0x").unwrap_or(hex)).ok())
}
/// A block's body, reduced to what is read from it.
///
/// One call for both facts. The ingest path wants only the timestamp and
+257
View File
@@ -131,6 +131,46 @@ pub struct RecordedBlock {
pub at_tip: bool,
}
/// One decoded event on its way into the database.
#[derive(Debug, Clone)]
pub struct EventRecord {
/// Which chain.
pub chain: ChainId,
/// Block it fired in.
pub height: u64,
/// Position within that block's event list.
pub event_index: u32,
/// Emitting pallet, as the runtime names it.
pub pallet: String,
/// Event variant, as the runtime names it.
pub variant: String,
/// `ApplyExtrinsic`, `Finalization` or `Initialization`.
pub phase: String,
/// Which extrinsic, when the phase is `ApplyExtrinsic`.
pub extrinsic_index: Option<u32>,
/// Normalised fields.
pub fields: serde_json::Value,
}
/// A decoded event read back.
#[derive(Debug, Clone)]
pub struct StoredEvent {
/// Block it fired in.
pub height: u64,
/// Position within that block's event list.
pub event_index: u32,
/// Emitting pallet.
pub pallet: String,
/// Event variant.
pub variant: String,
/// Dispatch phase.
pub phase: String,
/// Which extrinsic, when applicable.
pub extrinsic_index: Option<u32>,
/// Normalised fields.
pub fields: serde_json::Value,
}
/// A held attribution loaded back from the database.
#[derive(Debug, Clone)]
pub struct StoredAttribution {
@@ -698,6 +738,223 @@ impl Store {
.collect())
}
/// The metadata for a runtime version, if it has been cached.
pub async fn runtime_metadata(
&self,
chain: &ChainId,
spec_version: u32,
) -> Result<Option<Vec<u8>>, DataError> {
let row = sqlx::query!(
"select metadata from runtime_metadata where chain = $1 and spec_version = $2",
chain.as_str(),
i64::from(spec_version),
)
.fetch_optional(&self.pool)
.await?;
Ok(row.map(|r| r.metadata))
}
/// Cache a runtime's metadata, once.
///
/// `do nothing` on conflict: a given `spec_version` has exactly one
/// metadata blob by definition, so a second sighting is not new
/// information, and overwriting would discard the height at which the
/// version was first seen.
pub async fn cache_runtime_metadata(
&self,
chain: &ChainId,
spec_version: u32,
spec_name: &str,
metadata: &[u8],
first_seen_height: u64,
) -> Result<bool, DataError> {
let result = sqlx::query!(
r#"
insert into runtime_metadata
(chain, spec_version, spec_name, metadata, first_seen_height)
values ($1, $2, $3, $4, $5)
on conflict (chain, spec_version) do nothing
"#,
chain.as_str(),
i64::from(spec_version),
spec_name,
metadata,
first_seen_height as i64,
)
.execute(&self.pool)
.await?;
Ok(result.rows_affected() > 0)
}
/// Every runtime version cached for a chain, oldest first.
pub async fn cached_runtimes(
&self,
chain: &ChainId,
) -> Result<Vec<(u32, String, u64)>, DataError> {
let rows = sqlx::query!(
r#"
select spec_version, spec_name, first_seen_height
from runtime_metadata where chain = $1 order by spec_version asc
"#,
chain.as_str(),
)
.fetch_all(&self.pool)
.await?;
Ok(rows
.into_iter()
.map(|r| {
(
r.spec_version.max(0) as u32,
r.spec_name,
r.first_seen_height.max(0) as u64,
)
})
.collect())
}
/// Store a block's decoded events.
///
/// Idempotent on `(chain, height, event_index)`, so re-indexing a block —
/// after a reorg, or a backfill overlapping live ingest — replaces rather
/// than duplicates. One statement with array parameters for the same reason
/// `record_blocks` uses one: a backfill does this per block for thousands
/// of blocks.
pub async fn record_events(&self, events: &[EventRecord]) -> Result<u64, DataError> {
if events.is_empty() {
return Ok(0);
}
let chains: Vec<String> = events.iter().map(|e| e.chain.0.clone()).collect();
let heights: Vec<i64> = events.iter().map(|e| e.height as i64).collect();
let indexes: Vec<i32> = events.iter().map(|e| e.event_index as i32).collect();
let pallets: Vec<String> = events.iter().map(|e| e.pallet.clone()).collect();
let variants: Vec<String> = events.iter().map(|e| e.variant.clone()).collect();
let phases: Vec<String> = events.iter().map(|e| e.phase.clone()).collect();
let extrinsics: Vec<Option<i32>> = events
.iter()
.map(|e| e.extrinsic_index.map(|i| i as i32))
.collect();
let fields: Vec<serde_json::Value> = events.iter().map(|e| e.fields.clone()).collect();
let result = sqlx::query!(
r#"
insert into chain_event
(chain, height, event_index, pallet, variant, phase, extrinsic_index, fields)
select * from unnest(
$1::text[], $2::bigint[], $3::int[], $4::text[],
$5::text[], $6::text[], $7::int[], $8::jsonb[]
)
on conflict (chain, height, event_index) do update set
pallet = excluded.pallet,
variant = excluded.variant,
phase = excluded.phase,
extrinsic_index = excluded.extrinsic_index,
fields = excluded.fields
"#,
&chains,
&heights,
&indexes,
&pallets,
&variants,
&phases,
&extrinsics as &[Option<i32>],
&fields,
)
.execute(&self.pool)
.await?;
Ok(result.rows_affected())
}
/// The stretch of a chain's history whose events have been read, if any.
///
/// Deliberately not `max(height) from chain_event`: a block with no indexed
/// event and a block never looked at produce the same answer, and the first
/// is the common case. See the migration.
pub async fn event_scan(&self, chain: &ChainId) -> Result<Option<(u64, u64)>, DataError> {
let row = sqlx::query!(
"select low, high from event_scan where chain = $1",
chain.as_str()
)
.fetch_optional(&self.pool)
.await?;
Ok(row.map(|r| (r.low.max(0) as u64, r.high.max(0) as u64)))
}
/// Record how far the scan has reached.
///
/// The interval only ever grows, and `least`/`greatest` enforce that in the
/// statement rather than trusting the caller: an ingest that restarts mid
/// backfill would otherwise be able to narrow it back to the tip and lose
/// everything already read.
pub async fn set_event_scan(
&self,
chain: &ChainId,
low: u64,
high: u64,
) -> Result<(), DataError> {
sqlx::query!(
r#"
insert into event_scan (chain, low, high) values ($1, $2, $3)
on conflict (chain) do update
set low = least(event_scan.low, excluded.low),
high = greatest(event_scan.high, excluded.high),
updated_at = now()
"#,
chain.as_str(),
low as i64,
high as i64,
)
.execute(&self.pool)
.await?;
Ok(())
}
/// Events of one kind involving one account, newest first.
///
/// The generic query behind every per-account view. `fields @> …` is
/// answered by the `jsonb_path_ops` GIN index, so rewards, transfers and
/// whatever the chain grows next are all this one method with different
/// arguments — no table and no query per event type.
pub async fn events_for_account(
&self,
chain: &ChainId,
pallet: &str,
variant: &str,
field: &str,
account_hex: &str,
limit: i64,
) -> Result<Vec<StoredEvent>, DataError> {
let filter = serde_json::json!({ field: account_hex });
let rows = sqlx::query!(
r#"
select height, event_index, pallet, variant, phase, extrinsic_index,
fields as "fields!"
from chain_event
where chain = $1 and pallet = $2 and variant = $3 and fields @> $4
order by height desc, event_index desc
limit $5
"#,
chain.as_str(),
pallet,
variant,
filter,
limit,
)
.fetch_all(&self.pool)
.await?;
Ok(rows
.into_iter()
.map(|r| StoredEvent {
height: r.height.max(0) as u64,
event_index: r.event_index.max(0) as u32,
pallet: r.pallet,
variant: r.variant,
phase: r.phase,
extrinsic_index: r.extrinsic_index.map(|i| i.max(0) as u32),
fields: r.fields,
})
.collect())
}
/// Persist the attributions currently held.
pub async fn save_attributions(
&self,
+49
View File
@@ -173,6 +173,55 @@ did I do last week", a hashrate line older than the process, and a leaderboard
that survives a deploy. On start the daemon replays the window and the held
attribution names out of it, so a restart does not serve an empty site.
## The runtime is the oracle
A header tells you who mined a block. It does not tell you what they were paid:
the amount is remaining supply over an emission divisor plus collected fees,
quantized with the dust carried into the next block, so it is only knowable from
the `MiningRewards::MinerRewarded` event the runtime emits. Reading that means
decoding SCALE against the type registry of the runtime that produced the block
— and this chain's encodings differ from vanilla Substrate, because the
post-quantum signature scheme changes the shapes.
Rather than transcribe those shapes into Rust and re-transcribe them after every
upgrade, the observer asks the chain to describe itself. `state_getMetadata` at a
block hash makes the node execute `Metadata_metadata` against the runtime WASM in
that block's state, which returns metadata v14: every pallet, every event, every
field, as a portable `scale-info` registry. `blackbeard-core::runtime` decodes
against that registry and knows nothing about mining, rewards or transfers —
`INDEXED_EVENTS` in `ingest.rs` is the only place a name appears, and adding
transfers or multisig or whatever the chain grows next is one line there and a
query, not a decoder.
Metadata is cached per `spec_version` in `runtime_metadata`, because a node that
starts pruning would otherwise make every historical block undecodable. Blocks
are decoded against the tip's runtime first — which costs no extra RPC call and
is right for all but a handful of blocks — and `decode_events` refuses a partial
read, so a block from before an upgrade *fails* rather than decoding into
plausible nonsense. That failure is what triggers asking which runtime actually
produced it. It is the property the whole approach rests on.
Events land in one table, `chain_event`, in the runtime's own vocabulary:
`pallet`, `variant`, and the fields as jsonb. Account ids are normalised to `0x`
hex on the way in, and integers past 2^53 to strings — balances here are u128 and
a JSON number would be silently rounded by the browser. A `jsonb_path_ops` GIN
index over the fields is what makes "everything involving this account" a query
rather than a table per event type.
`ingest::backfill` walks the chain outward from the tip: to the tip first, then
toward genesis, one contiguous interval recorded in `event_scan`. Live indexing
alone is not enough for an account page — a restart, a brief RPC failure or a gap
fill that ran before the metadata was cached each leave a hole that nothing would
return to. Progress is a cursor rather than `max(height) from chain_event`,
because a block with no indexed event and a block never looked at are otherwise
indistinguishable, and the first is the common case.
A miner's `pow_` preimage and their reward event agree without a lookup table:
`wormhole::reward_address` runs the same Poseidon2 derivation the chain does, and
its output is exactly the `miner` field of `MinerRewarded`. Verified on mainnet
block 13160 — preimage `0x02dacf59…`, address `qzowWAgb…`, account
`0xc6801725…`, which is the event's own field.
## Build and run
```sh