diff --git a/lean_client/Cargo.lock b/lean_client/Cargo.lock index 077f2b77..de4d8b86 100644 --- a/lean_client/Cargo.lock +++ b/lean_client/Cargo.lock @@ -46,16 +46,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "air" -version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" -dependencies = [ - "field", - "koala-bear", - "poly", -] - [[package]] name = "alloy-primitives" version = "1.5.7" @@ -584,24 +574,6 @@ dependencies = [ "syn 2.0.117", ] -[[package]] -name = "backend" -version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" -dependencies = [ - "air", - "fiat-shamir", - "field", - "koala-bear", - "parallel", - "poly", - "sumcheck", - "symetric", - "utils", - "whir", - "zk-alloc", -] - [[package]] name = "base-x" version = "0.2.11" @@ -636,6 +608,15 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" +[[package]] +name = "bincode" +version = "1.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f45e9417d87227c7a56d22e471c6206462cba514c7590c09aff4cf6d1ddcad" +dependencies = [ + "serde", +] + [[package]] name = "bit-set" version = "0.8.0" @@ -1793,27 +1774,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" [[package]] -name = "fiat-shamir" +name = "fiat_shamir" version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" dependencies = [ - "field", - "koala-bear", "parallel", + "primitives", "serde", - "symetric", - "utils", -] - -[[package]] -name = "field" -version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" -dependencies = [ - "paste", - "rand 0.10.1", - "serde", - "utils", ] [[package]] @@ -1834,6 +1801,18 @@ dependencies = [ "static_assertions", ] +[[package]] +name = "flock" +version = "0.1.0" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" +dependencies = [ + "fiat_shamir", + "parallel", + "pcs", + "primitives", + "zk_alloc", +] + [[package]] name = "fnv" version = "1.0.7" @@ -2715,25 +2694,6 @@ dependencies = [ "syn 2.0.117", ] -[[package]] -name = "include_dir" -version = "0.7.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "923d117408f1e49d914f1a379a309cffe4f18c05cf4e3d12e613a15fc81bd0dd" -dependencies = [ - "include_dir_macros", -] - -[[package]] -name = "include_dir_macros" -version = "0.7.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7cab85a7ed0bd5f0e76d93846e0147172bed2e2d3f859bcc33a8d9699cad1a75" -dependencies = [ - "proc-macro2 1.0.106", - "quote 1.0.45", -] - [[package]] name = "indexmap" version = "1.9.3" @@ -2879,40 +2839,12 @@ dependencies = [ "sha3-asm", ] -[[package]] -name = "koala-bear" -version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" -dependencies = [ - "field", - "paste", - "rand 0.10.1", - "serde", - "utils", -] - [[package]] name = "lazy_static" version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" -[[package]] -name = "lean-multisig" -version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" -dependencies = [ - "backend", - "clap", - "lean_vm", - "rec_aggregation", - "serde_json", - "sub_protocols", - "system-info", - "xmss 0.1.0", - "zk-alloc", -] - [[package]] name = "lean_client" version = "0.1.0" @@ -2947,40 +2879,53 @@ dependencies = [ [[package]] name = "lean_compiler" version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" dependencies = [ - "backend", - "include_dir", "lean_vm", - "pest", - "pest_derive", - "sub_protocols", - "xmss 0.1.0", + "primitives", ] [[package]] -name = "lean_prover" +name = "lean_da" version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" dependencies = [ - "backend", - "lean_compiler", - "lean_vm", - "rand 0.10.1", + "fiat_shamir", + "parallel", + "pcs", + "primitives", "serde", - "sub_protocols", "tracing", - "xmss 0.1.0", ] [[package]] name = "lean_vm" version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" dependencies = [ - "backend", + "fiat_shamir", + "flock", + "parallel", + "pcs", + "primitives", "tracing", + "zk_alloc", +] + +[[package]] +name = "leanvm" +version = "0.1.0" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" +dependencies = [ + "clap", + "lean_da", + "lean_vm", + "primitives", + "rand 0.9.4", + "rec_aggregation", + "sphincs", "xmss 0.1.0", + "zk_alloc", ] [[package]] @@ -3717,31 +3662,6 @@ dependencies = [ "libc", ] -[[package]] -name = "objc2" -version = "0.6.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a12a8ed07aefc768292f076dc3ac8c48f3781c8f2d5851dd3d98950e8c5a89f" -dependencies = [ - "objc2-encode", -] - -[[package]] -name = "objc2-encode" -version = "4.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef25abbcd74fb2609453eb695bd2f860d389e457f67dc17cafc8b8cbc89d0c33" - -[[package]] -name = "objc2-foundation" -version = "0.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e3e0adef53c21f888deb4fa59fc59f7eb17404926ee8a6f59f5df0fd7f9f3272" -dependencies = [ - "bitflags", - "objc2", -] - [[package]] name = "oid-registry" version = "0.8.1" @@ -3777,9 +3697,9 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "parallel" version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" dependencies = [ - "system-info", + "libc", ] [[package]] @@ -3870,6 +3790,19 @@ version = "1.0.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" +[[package]] +name = "pcs" +version = "0.1.0" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" +dependencies = [ + "fiat_shamir", + "parallel", + "primitives", + "serde", + "tracing", + "zk_alloc", +] + [[package]] name = "pem" version = "3.0.6" @@ -3896,39 +3829,6 @@ dependencies = [ "ucd-trie", ] -[[package]] -name = "pest_derive" -version = "2.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "11f486f1ea21e6c10ed15d5a7c77165d0ee443402f0780849d1768e7d9d6fe77" -dependencies = [ - "pest", - "pest_generator", -] - -[[package]] -name = "pest_generator" -version = "2.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8040c4647b13b210a963c1ed407c1ff4fdfa01c31d6d2a098218702e6664f94f" -dependencies = [ - "pest", - "pest_meta", - "proc-macro2 1.0.106", - "quote 1.0.45", - "syn 2.0.117", -] - -[[package]] -name = "pest_meta" -version = "2.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89815c69d36021a140146f26659a81d6c2afa33d216d736dd4be5381a7362220" -dependencies = [ - "pest", - "sha2 0.10.9 (registry+https://github.com/rust-lang/crates.io-index)", -] - [[package]] name = "pin-project" version = "1.1.11" @@ -3979,21 +3879,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "poly" -version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" -dependencies = [ - "field", - "koala-bear", - "parallel", - "rand 0.10.1", - "serde", - "system-info", - "utils", - "zk-alloc", -] - [[package]] name = "poly1305" version = "0.8.0" @@ -4093,6 +3978,19 @@ dependencies = [ "uint 0.9.5", ] +[[package]] +name = "primitives" +version = "0.1.0" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" +dependencies = [ + "libc", + "parallel", + "serde", + "tracing-forest", + "tracing-subscriber", + "zk_alloc", +] + [[package]] name = "proc-macro-crate" version = "3.5.0" @@ -4520,19 +4418,19 @@ dependencies = [ [[package]] name = "rec_aggregation" version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" dependencies = [ - "backend", - "include_dir", + "bincode", + "flock", "lean_compiler", - "lean_prover", + "lean_da", "lean_vm", - "objc2", - "objc2-foundation", "parallel", - "postcard", + "pcs", + "primitives", + "rand 0.9.4", "serde", - "sub_protocols", + "sphincs", "tracing", "xmss 0.1.0", ] @@ -5252,6 +5150,17 @@ dependencies = [ "xmss 0.0.0", ] +[[package]] +name = "sphincs" +version = "0.1.0" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" +dependencies = [ + "parallel", + "primitives", + "rand 0.9.4", + "serde", +] + [[package]] name = "spin" version = "0.9.8" @@ -5388,48 +5297,12 @@ dependencies = [ "syn 2.0.117", ] -[[package]] -name = "sub_protocols" -version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" -dependencies = [ - "backend", - "lean_vm", - "tracing", -] - [[package]] name = "subtle" version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" -[[package]] -name = "sumcheck" -version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" -dependencies = [ - "air", - "fiat-shamir", - "field", - "koala-bear", - "parallel", - "poly", - "tracing", - "zk-alloc", -] - -[[package]] -name = "symetric" -version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" -dependencies = [ - "field", - "koala-bear", - "parallel", - "zk-alloc", -] - [[package]] name = "syn" version = "0.15.44" @@ -5504,14 +5377,6 @@ dependencies = [ "libc", ] -[[package]] -name = "system-info" -version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" -dependencies = [ - "libc", -] - [[package]] name = "tagptr" version = "0.2.0" @@ -6084,17 +5949,6 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" -[[package]] -name = "utils" -version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" -dependencies = [ - "parallel", - "serde", - "tracing-forest", - "tracing-subscriber", -] - [[package]] name = "uuid" version = "1.23.0" @@ -6312,25 +6166,6 @@ dependencies = [ "rustls-pki-types", ] -[[package]] -name = "whir" -version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" -dependencies = [ - "fiat-shamir", - "field", - "koala-bear", - "parallel", - "poly", - "rand 0.10.1", - "sumcheck", - "symetric", - "system-info", - "tracing", - "utils", - "zk-alloc", -] - [[package]] name = "widestring" version = "1.2.1" @@ -6728,7 +6563,7 @@ dependencies = [ "ethereum-types", "ethereum_ssz", "hex", - "lean-multisig", + "leanvm", "metrics", "postcard", "rand 0.10.1", @@ -6741,12 +6576,12 @@ dependencies = [ [[package]] name = "xmss" version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" dependencies = [ - "backend", "ethereum_ssz", - "postcard", - "rand 0.10.1", + "parallel", + "primitives", + "rand 0.9.4", "serde", ] @@ -6931,13 +6766,11 @@ dependencies = [ ] [[package]] -name = "zk-alloc" +name = "zk_alloc" version = "0.1.0" -source = "git+https://github.com/leanEthereum/leanVM.git?rev=a5909d18647de6aed38640c098d9177fab2bf36a#a5909d18647de6aed38640c098d9177fab2bf36a" +source = "git+https://github.com/leanEthereum/leanVM.git?rev=48a904208d682848dac0e18ef8b01ebfc40df9ad#48a904208d682848dac0e18ef8b01ebfc40df9ad" dependencies = [ "libc", - "parallel", - "system-info", ] [[package]] diff --git a/lean_client/Cargo.toml b/lean_client/Cargo.toml index ef2e7312..f6d9db01 100644 --- a/lean_client/Cargo.toml +++ b/lean_client/Cargo.toml @@ -253,7 +253,7 @@ indexmap = "2" http-body-util = "0.1" http_api_utils = { git = "https://github.com/grandinetech/grandine", rev = "c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" } k256 = "0.13" -lean-multisig = { git = "https://github.com/leanEthereum/leanVM.git", rev = "a5909d18647de6aed38640c098d9177fab2bf36a" } +leanvm = { git = "https://github.com/leanEthereum/leanVM.git", rev = "48a904208d682848dac0e18ef8b01ebfc40df9ad" } postcard = { version = "1.1.3", features = ["alloc"] } libp2p = { git = "https://github.com/libp2p/rust-libp2p.git", rev = "91e8931e275bcd1c72791d18b09fea8b77209baf", default-features = false, features = [ 'dns', diff --git a/lean_client/fork_choice/src/block_cache.rs b/lean_client/fork_choice/src/block_cache.rs index db29c913..0789ea65 100644 --- a/lean_client/fork_choice/src/block_cache.rs +++ b/lean_client/fork_choice/src/block_cache.rs @@ -157,6 +157,18 @@ impl BlockCache { results.into_iter().collect() } + pub fn prune_finalized(&mut self, finalized_slot: Slot) { + let stale: Vec = self + .blocks + .iter() + .filter(|(_, pending)| pending.slot <= finalized_slot) + .map(|(root, _)| *root) + .collect(); + for root in stale { + self.remove(&root); + } + } + pub fn clear(&mut self) { self.blocks.clear(); self.insertion_order.clear(); diff --git a/lean_client/fork_choice/src/handlers.rs b/lean_client/fork_choice/src/handlers.rs index 778c38d4..f1908f64 100644 --- a/lean_client/fork_choice/src/handlers.rs +++ b/lean_client/fork_choice/src/handlers.rs @@ -489,12 +489,7 @@ pub fn on_aggregated_attestation( } } - // Store the verified proof in latest_new_aggregated_payloads, keyed by data_root - store - .latest_new_aggregated_payloads - .entry(data_root) - .or_default() - .push(proof); + store.add_new_aggregated_payload(data_root, proof); METRICS.get().map(|metrics| { metrics @@ -765,6 +760,9 @@ pub fn apply_verified_block( ); store.blocks.insert(block_root, block.clone()); + store + .proto + .on_block(block.slot.0, block_root, block.parent_root); store.states.insert(block_root, Arc::clone(&new_state)); METRICS.get().map(|m| { diff --git a/lean_client/fork_choice/src/lib.rs b/lean_client/fork_choice/src/lib.rs index 7432afe9..0123635f 100644 --- a/lean_client/fork_choice/src/lib.rs +++ b/lean_client/fork_choice/src/lib.rs @@ -1,5 +1,6 @@ pub mod block_cache; pub mod handlers; +pub mod proto; pub mod reaggregate; pub mod store; pub mod sync_state; diff --git a/lean_client/fork_choice/src/proto.rs b/lean_client/fork_choice/src/proto.rs new file mode 100644 index 00000000..eed75210 --- /dev/null +++ b/lean_client/fork_choice/src/proto.rs @@ -0,0 +1,637 @@ +use std::collections::HashMap; + +use containers::{AggregatedSignatureProof, AttestationData}; +use ssz::H256; +use tracing::warn; + +#[derive(Debug, Clone)] +pub struct ProtoNode { + pub slot: u64, + pub root: H256, + pub parent: Option, + pub weight: u64, + pub best_child: Option, + pub best_descendant: Option, +} + +#[derive(Debug, Clone, Default)] +pub struct ProtoArray { + nodes: Vec, + indices: HashMap, +} + +impl ProtoArray { + pub fn len(&self) -> usize { + self.nodes.len() + } + + pub fn is_empty(&self) -> bool { + self.nodes.is_empty() + } + + pub fn index_of(&self, root: &H256) -> Option { + self.indices.get(root).copied() + } + + pub fn root_at(&self, index: usize) -> Option { + self.nodes.get(index).map(|n| n.root) + } + + pub fn on_block(&mut self, slot: u64, root: H256, parent_root: H256) -> bool { + if self.indices.contains_key(&root) { + return true; + } + let parent = self.indices.get(&parent_root).copied(); + if self.nodes.is_empty() { + self.nodes.push(ProtoNode { + slot, + root, + parent: None, + weight: 0, + best_child: None, + best_descendant: None, + }); + self.indices.insert(root, 0); + return true; + } + let Some(parent_index) = parent else { + return false; + }; + if slot <= self.nodes[parent_index].slot { + return false; + } + let index = self.nodes.len(); + self.nodes.push(ProtoNode { + slot, + root, + parent: Some(parent_index), + weight: 0, + best_child: None, + best_descendant: None, + }); + self.indices.insert(root, index); + true + } + + pub fn apply_score_changes(&mut self, mut deltas: Vec, cutoff: u64) { + if deltas.len() != self.nodes.len() { + return; + } + for i in (0..self.nodes.len()).rev() { + let delta = deltas[i]; + let parent = self.nodes[i].parent; + let node = &mut self.nodes[i]; + node.weight = if delta >= 0 { + node.weight.saturating_add(delta as u64) + } else { + node.weight.saturating_sub(delta.unsigned_abs()) + }; + if let Some(p) = parent { + deltas[p] += delta; + } + } + for node in &mut self.nodes { + node.best_child = None; + node.best_descendant = None; + } + for i in (1..self.nodes.len()).rev() { + let Some(p) = self.nodes[i].parent else { + continue; + }; + if self.nodes[i].weight < cutoff { + continue; + } + let candidate_descendant = self.nodes[i].best_descendant.unwrap_or(i); + let better = match self.nodes[p].best_child { + None => true, + Some(b) => { + (self.nodes[i].weight, self.nodes[i].root) + > (self.nodes[b].weight, self.nodes[b].root) + } + }; + if better { + self.nodes[p].best_child = Some(i); + self.nodes[p].best_descendant = Some(candidate_descendant); + } + } + } + + pub fn find_head(&self, root: &H256) -> Option { + let index = self.index_of(root)?; + Some(match self.nodes[index].best_descendant { + Some(d) => self.nodes[d].root, + None => self.nodes[index].root, + }) + } + + pub fn prune(&mut self, finalized_root: &H256) -> Option> { + let finalized_index = self.index_of(finalized_root)?; + if finalized_index == 0 { + return None; + } + let mut keep = vec![false; self.nodes.len()]; + let mut index_map = HashMap::new(); + let mut new_nodes: Vec = Vec::with_capacity(self.nodes.len() - finalized_index); + for i in 0..self.nodes.len() { + let is_finalized = i == finalized_index; + let parent_kept = self.nodes[i].parent.map(|p| keep[p]).unwrap_or(false); + if !(is_finalized || parent_kept) { + continue; + } + keep[i] = true; + index_map.insert(i, new_nodes.len()); + let mut node = self.nodes[i].clone(); + node.parent = if is_finalized { + None + } else { + node.parent.and_then(|p| index_map.get(&p).copied()) + }; + node.best_child = None; + node.best_descendant = None; + new_nodes.push(node); + } + self.indices = new_nodes + .iter() + .enumerate() + .map(|(i, n)| (n.root, i)) + .collect(); + self.nodes = new_nodes; + Some(index_map) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct VoteTarget { + pub index: usize, + pub slot: u64, + pub data_root: H256, +} + +fn stronger(a: &VoteTarget, b: &VoteTarget) -> bool { + (a.slot, a.data_root) > (b.slot, b.data_root) +} + +#[derive(Debug, Clone, Default)] +pub struct VoteTracker { + pub applied: Option, + pub known: Option, + pub new: Option, +} + +#[derive(Debug, Clone, Default)] +pub struct VoteStore { + trackers: Vec, +} + +impl VoteStore { + fn tracker_mut(&mut self, validator: u64) -> &mut VoteTracker { + let index = validator as usize; + if index >= self.trackers.len() { + self.trackers.resize_with(index + 1, VoteTracker::default); + } + &mut self.trackers[index] + } + + pub fn set_new(&mut self, validator: u64, target: VoteTarget) { + let tracker = self.tracker_mut(validator); + if tracker.new.map_or(true, |cur| stronger(&target, &cur)) { + tracker.new = Some(target); + } + } + + pub fn set_known(&mut self, validator: u64, target: VoteTarget) { + let tracker = self.tracker_mut(validator); + if tracker.known.map_or(true, |cur| stronger(&target, &cur)) { + tracker.known = Some(target); + } + } + + pub fn promote_new_to_known(&mut self) { + for tracker in &mut self.trackers { + if let Some(new) = tracker.new.take() { + if tracker.known.map_or(true, |cur| stronger(&new, &cur)) { + tracker.known = Some(new); + } + } + } + } + + pub fn compute_deltas(&mut self, num_nodes: usize, from_known: bool) -> Vec { + let mut deltas = vec![0i64; num_nodes]; + for tracker in &mut self.trackers { + if let Some(applied) = tracker.applied.take() { + if applied < num_nodes { + deltas[applied] -= 1; + } + } + let target = if from_known { + tracker.known + } else { + tracker.new + }; + if let Some(target) = target { + if target.index < num_nodes { + deltas[target.index] += 1; + tracker.applied = Some(target.index); + } + } + } + deltas + } + + pub fn remap(&mut self, index_map: &HashMap) { + for tracker in &mut self.trackers { + tracker.applied = tracker.applied.and_then(|i| index_map.get(&i).copied()); + tracker.known = tracker.known.and_then(|mut t| { + index_map.get(&t.index).map(|&i| { + t.index = i; + t + }) + }); + tracker.new = tracker.new.and_then(|mut t| { + index_map.get(&t.index).map(|&i| { + t.index = i; + t + }) + }); + } + } +} + +#[derive(Debug, Clone, Default)] +pub struct ProtoForkChoice { + array: ProtoArray, + votes: VoteStore, +} + +impl ProtoForkChoice { + pub fn on_block(&mut self, slot: u64, root: H256, parent_root: H256) { + if !self.array.on_block(slot, root, parent_root) { + warn!( + slot, + %root, + %parent_root, + "proto fork choice rejected block: parent unknown or slot not above parent" + ); + } + } + + pub fn contains_block(&self, root: &H256) -> bool { + self.array.index_of(root).is_some() + } + + pub fn num_nodes(&self) -> usize { + self.array.len() + } + + pub fn ingest_payload( + &mut self, + data: &AttestationData, + data_root: H256, + proof: &AggregatedSignatureProof, + known: bool, + ) { + let Some(index) = self.array.index_of(&data.head.root) else { + return; + }; + let target = VoteTarget { + index, + slot: data.slot.0, + data_root, + }; + for validator in proof.get_participant_indices() { + if known { + self.votes.set_known(validator, target); + } else { + self.votes.set_new(validator, target); + } + } + } + + pub fn promote_new_to_known(&mut self) { + self.votes.promote_new_to_known(); + } + + pub fn update_head(&mut self, justified_root: &H256) -> Option { + let root = self.resolve_root(justified_root)?; + let deltas = self.votes.compute_deltas(self.array.len(), true); + self.array.apply_score_changes(deltas, 0); + self.array.find_head(&root) + } + + pub fn update_safe_target(&mut self, justified_root: &H256, min_score: u64) -> Option { + let root = self.resolve_root(justified_root)?; + let deltas = self.votes.compute_deltas(self.array.len(), false); + self.array.apply_score_changes(deltas, min_score); + self.array.find_head(&root) + } + + fn resolve_root(&self, justified_root: &H256) -> Option { + if justified_root.is_zero() { + return self.array.root_at(0); + } + Some(*justified_root) + } + + pub fn prune(&mut self, finalized_root: &H256) { + if let Some(index_map) = self.array.prune(finalized_root) { + self.votes.remap(&index_map); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + struct Lcg(u64); + + impl Lcg { + fn next(&mut self) -> u64 { + self.0 = self.0.wrapping_mul(6364136223846793005).wrapping_add(1442695040888963407); + self.0 >> 16 + } + } + + fn h(byte: u64) -> H256 { + let mut out = [0u8; 32]; + out[..8].copy_from_slice(&byte.to_be_bytes()); + H256(out) + } + + fn oracle_head( + blocks: &HashMap, + mut root: H256, + votes: &HashMap, + min_votes: usize, + ) -> H256 { + if root.is_zero() { + root = *blocks + .iter() + .min_by_key(|(_, (slot, _))| *slot) + .map(|(r, _)| r) + .unwrap(); + } + let root_slot = match blocks.get(&root) { + Some((slot, _)) => *slot, + None => return root, + }; + let mut vote_weights: HashMap = HashMap::new(); + for (head_root, _) in votes.values() { + let mut curr = *head_root; + if let Some((slot, _)) = blocks.get(&curr) { + let mut curr_slot = *slot; + while curr_slot > root_slot { + *vote_weights.entry(curr).or_insert(0) += 1; + if let Some((_, parent)) = blocks.get(&curr) { + curr = *parent; + if curr.is_zero() { + break; + } + if let Some((next_slot, _)) = blocks.get(&curr) { + curr_slot = *next_slot; + } else { + break; + } + } else { + break; + } + } + } + } + let mut child_map: HashMap> = HashMap::new(); + for (block_hash, (_, parent)) in blocks { + if !parent.is_zero() + && vote_weights.get(block_hash).copied().unwrap_or(0) >= min_votes + { + child_map.entry(*parent).or_default().push(*block_hash); + } + } + let mut curr = root; + loop { + let children = match child_map.get(&curr) { + Some(list) if !list.is_empty() => list, + _ => return curr, + }; + curr = *children + .iter() + .max_by(|&&a, &&b| { + let wa = vote_weights.get(&a).copied().unwrap_or(0); + let wb = vote_weights.get(&b).copied().unwrap_or(0); + wa.cmp(&wb).then_with(|| a.cmp(&b)) + }) + .unwrap(); + } + } + + struct Scenario { + blocks: HashMap, + proto: ProtoForkChoice, + anchor: H256, + roots: Vec, + } + + fn random_scenario(rng: &mut Lcg, num_blocks: usize) -> Scenario { + let anchor = h(1); + let mut blocks = HashMap::new(); + blocks.insert(anchor, (0u64, H256::zero())); + let mut proto = ProtoForkChoice::default(); + proto.on_block(0, anchor, H256::zero()); + let mut roots = vec![anchor]; + for i in 1..num_blocks { + let parent = roots[(rng.next() as usize) % roots.len()]; + let parent_slot = blocks[&parent].0; + let slot = parent_slot + 1 + rng.next() % 3; + let root = h(0x1000 + i as u64 * 7 + rng.next() % 5); + if blocks.contains_key(&root) { + continue; + } + blocks.insert(root, (slot, parent)); + proto.on_block(slot, root, parent); + roots.push(root); + } + Scenario { + blocks, + proto, + anchor, + roots, + } + } + + fn random_votes( + rng: &mut Lcg, + scenario: &mut Scenario, + num_validators: u64, + known: bool, + ) -> HashMap { + let mut per_validator: HashMap> = HashMap::new(); + for validator in 0..num_validators { + let num_votes = 1 + rng.next() % 3; + for _ in 0..num_votes { + let head = scenario.roots[(rng.next() as usize) % scenario.roots.len()]; + let vote_slot = scenario.blocks[&head].0 + rng.next() % 4; + let data_root = h(0x9000_0000 + vote_slot * 131 + (head.0[7] as u64)); + per_validator + .entry(validator) + .or_default() + .push((vote_slot, data_root, head)); + } + } + let mut expected = HashMap::new(); + for (validator, mut entries) in per_validator { + for (vote_slot, data_root, head) in &entries { + let index = scenario.proto.array.index_of(head).unwrap(); + let target = VoteTarget { + index, + slot: *vote_slot, + data_root: *data_root, + }; + if known { + scenario.proto.votes.set_known(validator, target); + } else { + scenario.proto.votes.set_new(validator, target); + } + } + entries.sort_by(|a, b| (b.0, b.1).cmp(&(a.0, a.1))); + let (slot, _, head) = entries[0]; + expected.insert(validator, (head, slot)); + } + expected + } + + #[test] + fn head_matches_oracle_randomized() { + let mut rng = Lcg(42); + for round in 0..200 { + let mut scenario = random_scenario(&mut rng, 2 + (round % 40)); + let votes = random_votes(&mut rng, &mut scenario, 1 + round as u64 % 32, true); + let proto_head = scenario.proto.update_head(&scenario.anchor).unwrap(); + let oracle = oracle_head(&scenario.blocks, scenario.anchor, &votes, 0); + assert_eq!(proto_head, oracle, "round {round}"); + } + } + + #[test] + fn safe_target_matches_oracle_randomized() { + let mut rng = Lcg(1337); + for round in 0..200 { + let mut scenario = random_scenario(&mut rng, 2 + (round % 40)); + let num_validators = 1 + round as u64 % 32; + let votes = random_votes(&mut rng, &mut scenario, num_validators, false); + let min_score = (num_validators as usize * 2 + 2) / 3; + let proto_target = scenario + .proto + .update_safe_target(&scenario.anchor, min_score as u64) + .unwrap(); + let oracle = oracle_head(&scenario.blocks, scenario.anchor, &votes, min_score); + assert_eq!(proto_target, oracle, "round {round}"); + } + } + + #[test] + fn alternating_pools_keep_weights_consistent() { + let mut rng = Lcg(7); + let mut scenario = random_scenario(&mut rng, 30); + let known = random_votes(&mut rng, &mut scenario, 16, true); + let newer = random_votes(&mut rng, &mut scenario, 16, false); + for _ in 0..5 { + let target = scenario.proto.update_safe_target(&scenario.anchor, 11); + let head = scenario.proto.update_head(&scenario.anchor).unwrap(); + let oracle = oracle_head(&scenario.blocks, scenario.anchor, &known, 0); + assert_eq!(head, oracle); + let oracle_target = oracle_head(&scenario.blocks, scenario.anchor, &newer, 11); + assert_eq!(target.unwrap(), oracle_target); + } + } + + #[test] + fn promote_keeps_strongest_vote() { + let mut proto = ProtoForkChoice::default(); + let a = h(1); + let b = h(2); + proto.on_block(0, a, H256::zero()); + proto.on_block(1, b, a); + let strong = VoteTarget { + index: 1, + slot: 9, + data_root: h(0xAA), + }; + let weak = VoteTarget { + index: 0, + slot: 3, + data_root: h(0xBB), + }; + proto.votes.set_known(0, strong); + proto.votes.set_new(0, weak); + proto.promote_new_to_known(); + assert_eq!(proto.votes.trackers[0].known, Some(strong)); + assert_eq!(proto.votes.trackers[0].new, None); + } + + #[test] + fn set_vote_keeps_max_slot_and_root() { + let mut votes = VoteStore::default(); + let first = VoteTarget { + index: 1, + slot: 5, + data_root: h(0x20), + }; + let older = VoteTarget { + index: 2, + slot: 4, + data_root: h(0xFF), + }; + let same_slot_higher_root = VoteTarget { + index: 3, + slot: 5, + data_root: h(0x30), + }; + votes.set_known(0, first); + votes.set_known(0, older); + assert_eq!(votes.trackers[0].known, Some(first)); + votes.set_known(0, same_slot_higher_root); + assert_eq!(votes.trackers[0].known, Some(same_slot_higher_root)); + } + + #[test] + fn prune_keeps_finalized_subtree_and_remaps_votes() { + let mut rng = Lcg(99); + let mut scenario = random_scenario(&mut rng, 40); + let votes = random_votes(&mut rng, &mut scenario, 20, true); + let head_before = scenario.proto.update_head(&scenario.anchor).unwrap(); + let mut finalized = head_before; + for _ in 0..3 { + if let Some((_, parent)) = scenario.blocks.get(&finalized) { + if !parent.is_zero() && scenario.blocks.contains_key(parent) { + finalized = *parent; + } + } + } + scenario.proto.prune(&finalized); + assert!(scenario.proto.contains_block(&finalized)); + assert!(scenario.proto.contains_block(&head_before)); + let head_after = scenario.proto.update_head(&finalized).unwrap(); + let surviving: HashMap = scenario + .blocks + .iter() + .filter(|(root, _)| scenario.proto.contains_block(root)) + .map(|(r, v)| (*r, *v)) + .collect(); + let surviving_votes: HashMap = votes + .into_iter() + .filter(|(_, (head, _))| surviving.contains_key(head)) + .collect(); + let oracle = oracle_head(&surviving, finalized, &surviving_votes, 0); + assert_eq!(head_after, oracle); + } + + #[test] + fn empty_array_bootstraps_from_first_block() { + let mut proto = ProtoForkChoice::default(); + let genesis = h(5); + proto.on_block(0, genesis, H256::zero()); + assert!(proto.contains_block(&genesis)); + assert_eq!(proto.update_head(&genesis), Some(genesis)); + assert_eq!(proto.update_head(&H256::zero()), Some(genesis)); + } +} diff --git a/lean_client/fork_choice/src/reaggregate.rs b/lean_client/fork_choice/src/reaggregate.rs index 571ce1a9..8be98a83 100644 --- a/lean_client/fork_choice/src/reaggregate.rs +++ b/lean_client/fork_choice/src/reaggregate.rs @@ -14,6 +14,9 @@ pub const MAX_REAGGREGATIONS_PER_BLOCK: usize = 4; pub struct ReaggregateContext { pubkeys_per_component: Vec>, + /// The `(message, slot)` binding of each component, index-aligned with + /// `pubkeys_per_component`: one per attestation, then the proposer's. + components: Vec<(H256, u32)>, candidates: Vec<(H256, AggregationBits)>, } @@ -41,6 +44,7 @@ pub fn select_candidates( let mut pubkeys_per_component: Vec> = Vec::with_capacity(attestations.len_u64() as usize + 1); + let mut components: Vec<(H256, u32)> = Vec::with_capacity(attestations.len_u64() as usize + 1); for att in attestations.into_iter() { let validator_ids = att.aggregation_bits.to_validator_indices(); @@ -62,6 +66,7 @@ pub fn select_candidates( pks.push(v.attestation_pubkey.clone()); } pubkeys_per_component.push(pks); + components.push((att.data.hash_tree_root(), att.data.slot.0 as u32)); } let proposer = match validators.get(proposer_index) { @@ -76,6 +81,7 @@ pub fn select_candidates( } }; pubkeys_per_component.push(vec![proposer.proposal_pubkey.clone()]); + components.push((block.hash_tree_root(), block.slot.0 as u32)); let latest_justified_slot = store.latest_justified.slot; let mut candidates: Vec<(H256, AggregationBits)> = Vec::new(); @@ -147,6 +153,7 @@ pub fn select_candidates( Some(ReaggregateContext { pubkeys_per_component, + components, candidates, }) } @@ -166,6 +173,7 @@ pub fn compute_recoveries( for (data_root, participants) in context.candidates { match signed_block.proof.split_by_message( &pubkeys_per_component_view, + &context.components, data_root, log_inv_rate, ) { @@ -198,11 +206,7 @@ pub fn compute_recoveries( pub fn apply_recoveries(store: &mut Store, recoveries: Vec<(H256, AggregatedSignatureProof)>) { for (data_root, proof) in recoveries { - store - .latest_new_aggregated_payloads - .entry(data_root) - .or_default() - .push(proof); + store.add_new_aggregated_payload(data_root, proof); } } @@ -233,6 +237,19 @@ pub fn run_sync(store: &mut Store, signed_block: &SignedBlock, log_inv_rate: usi } pub fn run_in_executor(store: Arc>, signed_block: SignedBlock, log_inv_rate: usize) { + if xmss::prover_busy() { + debug!( + block_root = %signed_block.block.hash_tree_root(), + "reaggregate skipped: prover busy" + ); + METRICS.get().map(|m| { + m.grandine_reaggregate_split_outcomes_total + .with_label_values(&["skipped_prover_busy"]) + .inc() + }); + return; + } + let context = { let s = store.read(); let parent_validators = match s.states.get(&signed_block.block.parent_root) { diff --git a/lean_client/fork_choice/src/store.rs b/lean_client/fork_choice/src/store.rs index 18ffe095..4d95a359 100644 --- a/lean_client/fork_choice/src/store.rs +++ b/lean_client/fork_choice/src/store.rs @@ -12,6 +12,8 @@ use ssz::{H256, SszHash}; use tracing::{info, warn}; use xmss::Signature; +use crate::proto::ProtoForkChoice; + pub type Interval = u64; pub const INTERVALS_PER_SLOT: Interval = 5; pub const SECONDS_PER_SLOT: u64 = 4; @@ -109,6 +111,29 @@ pub struct Store { pub pending_fetch_roots: HashSet, pub log_inv_rate: usize, + + pub proto: ProtoForkChoice, +} + +impl Store { + pub fn add_new_aggregated_payload( + &mut self, + data_root: H256, + proof: AggregatedSignatureProof, + ) { + if let Some(data) = self.attestation_data_by_root.get(&data_root).cloned() { + self.proto.ingest_payload(&data, data_root, &proof, false); + } else { + warn!( + %data_root, + "aggregated payload stored without attestation data; vote not counted" + ); + } + self.latest_new_aggregated_payloads + .entry(data_root) + .or_default() + .push(proof); + } } const JUSTIFICATION_LOOKBACK_SLOTS: u64 = 3; @@ -295,6 +320,11 @@ pub fn get_forkchoice_store( pending_aggregated_attestations: HashMap::new(), pending_fetch_roots: HashSet::new(), log_inv_rate, + proto: { + let mut proto = ProtoForkChoice::default(); + proto.on_block(block_slot.0, block_root, H256::zero()); + proto + }, } } @@ -395,14 +425,15 @@ pub fn get_latest_justified(states: &HashMap>) -> Option<&Check pub fn update_head(store: &mut Store) { let old_head = store.head; - let latest_votes = extract_attestations_from_aggregated_payloads( - &store.latest_known_aggregated_payloads, - &store.attestation_data_by_root, - store.latest_finalized.slot, - ); - - // Compute new head using LMD-GHOST from latest justified root - let new_head = get_fork_choice_head(store, store.latest_justified.root, &latest_votes, 0); + let justified_root = store.latest_justified.root; + let new_head = store.proto.update_head(&justified_root).unwrap_or_else(|| { + warn!( + %justified_root, + justified_slot = store.latest_justified.slot.0, + "justified root not in proto fork choice, returning justified root as head" + ); + justified_root + }); store.head = new_head; if let Some(head_state) = store.states.get(&new_head) { @@ -420,11 +451,15 @@ pub fn update_head(store: &mut Store) { } if let Some(block) = store.blocks.get(&finalized_root) { if block.slot == finalized_slot { + let advanced = finalized_slot > store.latest_finalized.slot; store.latest_finalized = Checkpoint { root: finalized_root, slot: finalized_slot, }; store.finalized_ever_updated = true; + if advanced { + store.proto.prune(&finalized_root); + } METRICS.get().map(|m| { if let Ok(s) = i64::try_from(finalized_slot.0) { m.lean_latest_finalized_slot.set(s); @@ -552,17 +587,12 @@ pub fn update_safe_target(store: &mut Store) { let min_score = (n_validators * 2 + 2) / 3; let root = store.latest_justified.root; - // Extract per-validator attestations from the "new" pool only. - // The "known" pool is intentionally excluded — see the doc comment above + // Only the "new" pool is consulted here — see the doc comment above // for the availability rationale tied to the interval-3/interval-4 ordering. - let attestations = extract_attestations_from_aggregated_payloads( - &store.latest_new_aggregated_payloads, - &store.attestation_data_by_root, - store.latest_finalized.slot, - ); - - // Run LMD-GHOST with 2/3 threshold to find safe target - let new_safe_target = get_fork_choice_head(store, root, &attestations, min_score); + let new_safe_target = store + .proto + .update_safe_target(&root, min_score as u64) + .unwrap_or(root); store.safe_target = new_safe_target; set_gauge_u64( @@ -579,6 +609,7 @@ pub fn update_safe_target(store: &mut Store) { } pub fn accept_new_attestations(store: &mut Store) { + store.proto.promote_new_to_known(); store .latest_known_attestations .extend(store.latest_new_attestations.drain()); diff --git a/lean_client/networking/src/network/service.rs b/lean_client/networking/src/network/service.rs index a8766f67..ebeaf0c4 100644 --- a/lean_client/networking/src/network/service.rs +++ b/lean_client/networking/src/network/service.rs @@ -1789,36 +1789,21 @@ where tokio::spawn(async move { for block in blocks { let slot = block.block.slot.0; - match chain_sink.try_send(ChainMessage::ProcessBlock { - signed_block: block, - is_trusted: false, - should_gossip: false, - cached_post_state: None, - }) { - Ok(()) => {} - Err(tokio::sync::mpsc::error::TrySendError::Full( - _, - )) => { - warn!( - slot, - protocol = "blocks_by_range", - "Dropping RPC chunk: chain channel full" - ); - METRICS.get().map(|m| { - m.lean_chain_message_drop_total - .with_label_values(&["blocks_by_range"]) - .inc() - }); - } - Err( - tokio::sync::mpsc::error::TrySendError::Closed(_), - ) => { - warn!( - slot, - "Failed to forward range block to chain: channel closed" - ); - break; - } + if let Err(err) = chain_sink + .send(ChainMessage::ProcessBlock { + signed_block: block, + is_trusted: false, + should_gossip: false, + cached_post_state: None, + }) + .await + { + warn!( + slot, + ?err, + "Failed to forward range block to chain" + ); + break; } } }); diff --git a/lean_client/src/main.rs b/lean_client/src/main.rs index 4055511e..c47902ae 100644 --- a/lean_client/src/main.rs +++ b/lean_client/src/main.rs @@ -58,24 +58,86 @@ mod aggregation; const BACKFILL_ORPHAN_TRIGGER: usize = 5; const BACKFILL_RANGE_COUNT: u64 = 256; +const RANGE_REQUEST_RETRY: Duration = Duration::from_secs(10); +const MAX_FETCH_RETRIES: u32 = 10; +const INITIAL_FETCH_BACKOFF: Duration = Duration::from_secs(1); +const MAX_FETCH_BACKOFF: Duration = Duration::from_secs(30); + +struct FetchQueue { + entries: HashMap, +} + +impl FetchQueue { + fn new() -> Self { + Self { + entries: HashMap::new(), + } + } + + fn add(&mut self, root: H256) { + self.entries.entry(root).or_insert((0, Instant::now())); + } + + fn resolve(&mut self, root: &H256) { + self.entries.remove(root); + } + + fn due(&mut self) -> Vec { + let now = Instant::now(); + let mut ready = Vec::new(); + self.entries.retain(|root, (attempts, next_attempt)| { + if *next_attempt > now { + return true; + } + if *attempts >= MAX_FETCH_RETRIES { + return false; + } + *attempts += 1; + let backoff = INITIAL_FETCH_BACKOFF + .saturating_mul(1u32 << (*attempts - 1).min(5)) + .min(MAX_FETCH_BACKOFF); + *next_attempt = now + backoff; + ready.push(*root); + true + }); + ready + } +} + fn dispatch_backfill( missing: Vec, local_head_slot: u64, orphan_count: usize, + range_watermark: &mut Option<(u64, Instant)>, outbound_p2p_sender: &mpsc::UnboundedSender, ) { + if !missing.is_empty() { + if let Err(e) = outbound_p2p_sender.send(OutboundP2pRequest::RequestBlocksByRoot(missing)) { + warn!("Failed to dispatch BlocksByRoot backfill: {}", e); + } + } if orphan_count >= BACKFILL_ORPHAN_TRIGGER { + let head_next = local_head_slot.saturating_add(1); + let start_slot = match *range_watermark { + Some((covered, requested_at)) if requested_at.elapsed() < RANGE_REQUEST_RETRY => { + if covered >= head_next.saturating_add(BACKFILL_RANGE_COUNT - 1) { + return; + } + covered.saturating_add(1).max(head_next) + } + _ => head_next, + }; + *range_watermark = Some(( + start_slot.saturating_add(BACKFILL_RANGE_COUNT - 1), + Instant::now(), + )); let request = OutboundP2pRequest::RequestBlocksByRange { - start_slot: local_head_slot.saturating_add(1), + start_slot, count: BACKFILL_RANGE_COUNT, }; if let Err(e) = outbound_p2p_sender.send(request) { warn!("Failed to dispatch BlocksByRange backfill: {}", e); } - } else if !missing.is_empty() { - if let Err(e) = outbound_p2p_sender.send(OutboundP2pRequest::RequestBlocksByRoot(missing)) { - warn!("Failed to dispatch BlocksByRoot backfill: {}", e); - } } } @@ -1235,6 +1297,9 @@ async fn main() -> Result<()> { // rx error; the caller can retry on the next tick. let mut current_produce_block: Option<(Slot, tokio::task::JoinHandle<()>)> = None; let mut block_cache = BlockCache::new(); + let mut range_watermark: Option<(u64, Instant)> = None; + let mut last_pruned_finalized = Slot(0); + let mut fetch_queue = FetchQueue::new(); let mut sync_state = if has_aggregator { SyncState::Syncing } else { @@ -1452,22 +1517,51 @@ async fn main() -> Result<()> { let _ = block_slot_tx.send(block_slot.0); { - let exec = cpu_verify_executor.clone(); - let store_for_reagg = store.clone(); - let signed_block_for_reagg = signed_block.clone(); - let log_inv_rate = chain_log_inv_rate; - tokio::spawn(async move { - let job = exec.spawn(async move { - fork_choice::reaggregate::run_in_executor( - store_for_reagg, - signed_block_for_reagg, - log_inv_rate, - ); + let (head_slot_now, latest_justified_slot, wall_slot) = { + let s = store.read(); + ( + s.blocks.get(&s.head).map(|b| b.slot.0).unwrap_or(0), + s.latest_justified.slot.0, + s.time / INTERVALS_PER_SLOT, + ) + }; + let wall_head_lag = wall_slot.saturating_sub(head_slot_now); + let peer_head = *network_head_slot.lock(); + let has_fresher_peer_near_wall = peer_head + .map(|h| { + h.saturating_add(BLOCK_PROPOSAL_MAX_HEAD_LAG_SLOTS) + >= wall_slot + }) + .unwrap_or(false); + if should_suppress_proposal_for_head_lag( + wall_head_lag, + BLOCK_PROPOSAL_MAX_HEAD_LAG_SLOTS, + latest_justified_slot, + has_fresher_peer_near_wall, + ) { + debug!( + wall_head_lag, + head_slot = head_slot_now, + "Skipping reaggregation: local head is behind wall clock" + ); + } else { + let exec = cpu_verify_executor.clone(); + let store_for_reagg = store.clone(); + let signed_block_for_reagg = signed_block.clone(); + let log_inv_rate = chain_log_inv_rate; + tokio::spawn(async move { + let job = exec.spawn(async move { + fork_choice::reaggregate::run_in_executor( + store_for_reagg, + signed_block_for_reagg, + log_inv_rate, + ); + }); + if let Err(e) = job.await { + warn!("reaggregate executor failed: {e}"); + } }); - if let Err(e) = job.await { - warn!("reaggregate executor failed: {e}"); - } - }); + } } { @@ -1480,6 +1574,17 @@ async fn main() -> Result<()> { }; } + { + let finalized_slot = store.read().latest_finalized.slot; + if finalized_slot > last_pruned_finalized { + last_pruned_finalized = finalized_slot; + block_cache.prune_finalized(finalized_slot); + METRICS.get().map(|m| { + m.grandine_block_cache_size.set(block_cache.len() as i64) + }); + } + } + // Bug 1 invariant: rebroadcast only on Apply (apply_verified_block // is reached only when the dedup guard above missed, i.e. the // block is new to us). The original ChainMessage carried @@ -1562,7 +1667,12 @@ async fn main() -> Result<()> { .get() .map(|m| m.grandine_block_cache_size.set(block_cache.len() as i64)); - let missing: Vec = store.write().pending_fetch_roots.drain().collect(); + for root in store.write().pending_fetch_roots.drain() { + if !block_cache.contains(&root) { + fetch_queue.add(root); + } + } + let missing: Vec = fetch_queue.due(); METRICS.get().map(|m| m.grandine_pending_fetch_roots.set(0)); let local_head_slot = { let s = store.read(); @@ -1572,6 +1682,7 @@ async fn main() -> Result<()> { missing, local_head_slot, block_cache.orphan_count(), + &mut range_watermark, &outbound_p2p_sender, ); } @@ -1597,10 +1708,10 @@ async fn main() -> Result<()> { let mut s = store.write(); for aggregation in aggregations { let data_root = aggregation.data.hash_tree_root(); - s.latest_new_aggregated_payloads + s.attestation_data_by_root .entry(data_root) - .or_default() - .push(aggregation.proof.clone()); + .or_insert_with(|| aggregation.data.clone()); + s.add_new_aggregated_payload(data_root, aggregation.proof.clone()); to_publish.push(aggregation); } s.gossip_signatures.retain(|key, _| { @@ -1653,10 +1764,20 @@ async fn main() -> Result<()> { let block_root = signed_block.block.hash_tree_root(); let parent_root = signed_block.block.parent_root; + fetch_queue.resolve(&block_root); + if store.read().blocks.contains_key(&block_root) { continue; } + let parent_state = { + let s = store.read(); + s.states.get(&parent_root).cloned() + }; + if parent_state.is_none() && block_cache.contains(&block_root) { + continue; + } + if AggregatedAttestation::has_duplicate_data( &signed_block.block.body.attestations, ) { @@ -1703,16 +1824,17 @@ async fn main() -> Result<()> { } } - // Snapshot parent state for the executor. If absent, treat as - // orphan and re-queue. Genesis (parent_root.is_zero()) is loaded - // at startup, so a None here for non-zero parents means the - // parent block hasn't been processed yet. - let parent_state = { - let s = store.read(); - s.states.get(&parent_root).cloned() - }; - let Some(parent_state) = parent_state else { + let finalized_slot = { store.read().latest_finalized.slot }; + if block_slot <= finalized_slot { + debug!( + slot = block_slot.0, + finalized_slot = finalized_slot.0, + "Dropping orphan at or below finalized slot" + ); + continue; + } + block_cache.add( signed_block.clone(), block_root, @@ -1733,13 +1855,19 @@ async fn main() -> Result<()> { "Block cached (proactive) - parent not found, requesting via BlocksByRoot" ); - let missing: Vec = store.write().pending_fetch_roots.drain().collect(); + for root in store.write().pending_fetch_roots.drain() { + if !block_cache.contains(&root) { + fetch_queue.add(root); + } + } + let missing: Vec = fetch_queue.due(); METRICS.get().map(|m| m.grandine_pending_fetch_roots.set(0)); let head_slot = { let s = store.read(); s.blocks.get(&s.head).map(|b| b.slot.0).unwrap_or(0) }; dispatch_backfill( missing, head_slot, block_cache.orphan_count(), + &mut range_watermark, &outbound_p2p_sender, ); @@ -1750,6 +1878,8 @@ async fn main() -> Result<()> { continue; }; + block_cache.remove(&block_root); + // Phase 1: ship verify+state_transition to the cpu_normal executor. // The chain task returns to the select! loop immediately and processes // the verified outcome later via verify_result_rx (Phase 3). @@ -1840,7 +1970,12 @@ async fn main() -> Result<()> { Err(e) => warn!("Error processing attestation: {}", e), } - let missing: Vec = store.write().pending_fetch_roots.drain().collect(); + for root in store.write().pending_fetch_roots.drain() { + if !block_cache.contains(&root) { + fetch_queue.add(root); + } + } + let missing: Vec = fetch_queue.due(); METRICS.get().map(|m| m.grandine_pending_fetch_roots.set(0)); let local_head_slot = { let s = store.read(); @@ -1850,6 +1985,7 @@ async fn main() -> Result<()> { missing, local_head_slot, block_cache.orphan_count(), + &mut range_watermark, &outbound_p2p_sender, ); } @@ -1899,7 +2035,12 @@ async fn main() -> Result<()> { }); } - let missing: Vec = store.write().pending_fetch_roots.drain().collect(); + for root in store.write().pending_fetch_roots.drain() { + if !block_cache.contains(&root) { + fetch_queue.add(root); + } + } + let missing: Vec = fetch_queue.due(); METRICS.get().map(|m| m.grandine_pending_fetch_roots.set(0)); let local_head_slot = { let s = store.read(); @@ -1909,6 +2050,7 @@ async fn main() -> Result<()> { missing, local_head_slot, block_cache.orphan_count(), + &mut range_watermark, &outbound_p2p_sender, ); @@ -2058,6 +2200,36 @@ async fn main() -> Result<()> { let _ = sender.send(Err(anyhow::anyhow!( "not ready: justified checkpoint has not advanced from anchor value" ))); + continue; + } + + let head_slot = store_read + .blocks + .get(&store_read.head) + .map(|b| b.slot.0) + .unwrap_or(0); + let wall_head_lag = slot.0.saturating_sub(head_slot); + let peer_head = *network_head_slot.lock(); + let has_fresher_peer_near_wall = peer_head + .map(|h| { + h.saturating_add(BLOCK_PROPOSAL_MAX_HEAD_LAG_SLOTS) >= slot.0 + }) + .unwrap_or(false); + if should_suppress_proposal_for_head_lag( + wall_head_lag, + BLOCK_PROPOSAL_MAX_HEAD_LAG_SLOTS, + store_read.latest_justified.slot.0, + has_fresher_peer_near_wall, + ) { + warn!( + slot = slot.0, + wall_head_lag, + head_slot, + "Skipping attestation production: local head is {wall_head_lag} wall-clock slots behind" + ); + let _ = sender.send(Err(anyhow::anyhow!( + "attestation suppressed: head {wall_head_lag} slots behind wall" + ))); } else { let result = store_read.produce_attestation_data(slot); let _ = sender.send(result); diff --git a/lean_client/validator/src/lib.rs b/lean_client/validator/src/lib.rs index 332fee8f..26efcbe7 100644 --- a/lean_client/validator/src/lib.rs +++ b/lean_client/validator/src/lib.rs @@ -661,14 +661,38 @@ impl ValidatorService { } pubkeys_owned.push(vec![proposer_proposal_pubkey]); - let mut parts: Vec<(&AggregatedSignature, &[xmss::PublicKey])> = + // Each Type-1 part carries its own (message, slot) binding, needed + // to decode the proof bytes: attestation proofs are index-aligned + // with the block body's attestations, the proposer part binds the + // block root at the block's slot. + let attestation_bindings: Vec<(H256, u32)> = (&block.body.attestations) + .into_iter() + .map(|att| (att.data.hash_tree_root(), att.data.slot.0 as u32)) + .collect(); + if attestation_bindings.len() != attestation_signatures.len() { + bail!( + "attestation proof count ({}) does not match block body attestations ({})", + attestation_signatures.len(), + attestation_bindings.len() + ); + } + + let mut parts: Vec<(&AggregatedSignature, &[xmss::PublicKey], H256, u32)> = Vec::with_capacity(attestation_signatures.len() + 1); for (i, sig) in attestation_signatures.iter().enumerate() { - parts.push((&sig.proof_data, pubkeys_owned[i].as_slice())); + let (data_root, data_slot) = attestation_bindings[i]; + parts.push(( + &sig.proof_data, + pubkeys_owned[i].as_slice(), + data_root, + data_slot, + )); } parts.push(( &proposer_type1, pubkeys_owned[attestation_signatures.len()].as_slice(), + block_root, + block_slot, )); let type2_start = std::time::Instant::now(); diff --git a/lean_client/xmss/Cargo.toml b/lean_client/xmss/Cargo.toml index 65dfda08..33a604ec 100644 --- a/lean_client/xmss/Cargo.toml +++ b/lean_client/xmss/Cargo.toml @@ -12,7 +12,7 @@ derive_more = { workspace = true } eth_ssz = { workspace = true } ethereum-types = { workspace = true } hex = { workspace = true } -lean-multisig = { workspace = true } +leanvm = { workspace = true } metrics = { workspace = true } postcard = { workspace = true } rand = { workspace = true } diff --git a/lean_client/xmss/src/aggregated_signature.rs b/lean_client/xmss/src/aggregated_signature.rs index 3871aa9d..6a79edc5 100644 --- a/lean_client/xmss/src/aggregated_signature.rs +++ b/lean_client/xmss/src/aggregated_signature.rs @@ -10,10 +10,7 @@ use std::{ use crate::{PublicKey, Signature}; use anyhow::{Context, Error, Result, anyhow, bail}; use ethereum_types::H256; -use lean_multisig::{ - SingleMessageAggregateSignature, XmssPublicKey, aggregate_single_message_signatures, - verify_single_message_aggregate, -}; +use leanvm::{EthereumProof, SignatureClaims, XmssClaimGroup, aggregate, xmss::XmssPublicKey}; use metrics::{METRICS, stop_and_discard}; use serde::{Deserialize, Deserializer, Serialize, Serializer, de}; use ssz::{ByteList, ReadError, Size, SszHash, SszRead, SszSize, SszWrite, U1, WriteError}; @@ -24,8 +21,9 @@ type AggregatedSignatureSizeLimit = U524288; /// Cryptographic proof that a set of validators signed a message. /// -/// Wire form: the lean-multisig `to_bytes_without_pubkeys()` output. Pubkeys -/// are not baked into the bytes — verifiers must supply them externally, +/// Wire form: the leanVM `EthereumProof::to_bytes_without_pubkeys()` output. +/// Neither the pubkeys nor the `(message, slot)` binding are baked into the +/// bytes — verifiers must supply them externally, /// matching `SingleMessageAggregate.proof: ByteList512KiB` in leanSpec /// (`forks/lstar/containers/aggregation.py`) and zeam's `xmss_verify_type_1` /// FFI which takes `pks + msg + slot + wire` as separate args. @@ -71,17 +69,38 @@ pub fn setup_aggregation() { static SETUP: Once = Once::new(); SETUP.call_once(|| { if USE_ARENA.load(Ordering::SeqCst) { - lean_multisig::setup_prover(); + leanvm::setup_prover(); } else { - lean_multisig::setup_prover_without_arena(); + leanvm::setup_prover_without_arena(); } }); } +/// The signer set leanVM binds a Type-1 aggregate to: one XMSS group carrying +/// the `(message, slot)` claim and its strictly sorted, deduplicated keys. +/// The binding is not on the wire; a set other than the one aggregated decodes +/// fine and fails inside the SNARK verifier. +pub(crate) fn one_group(message: H256, slot: u32, mut keys: Vec) -> SignatureClaims { + keys.sort(); + keys.dedup(); + SignatureClaims { + xmss: vec![XmssClaimGroup { + epoch: slot, + message: *message.as_fixed_bytes(), + keys, + }], + sphincs: Vec::new(), + } +} + /// leanVM allows one proof at a time per process; a second concurrent one /// panics. Hold this lock across every prove call; verification needs no permit. pub(crate) static PROVER_PERMIT: Mutex<()> = Mutex::new(()); +pub fn prover_busy() -> bool { + PROVER_PERMIT.try_lock().is_err() +} + impl AggregatedSignature { pub fn new(bytes: &[u8]) -> Result { let bytes = ByteList::try_from(bytes.to_vec()) @@ -163,23 +182,19 @@ impl AggregatedSignature { let raw_xmss = public_keys .into_iter() .zip(signatures) - .map(|(pk, sig)| (pk.as_lean(), sig.as_lean())) + .map(|(pk, sig)| (pk.as_lean(), slot, *message.as_fixed_bytes(), sig.as_lean())) .collect::>(); - let children_arg: Vec = children + let children_arg: Vec = children .iter() - .map(|(pks, agg)| agg.as_lean(sorted_dedup_lean_pubkeys(pks))) + .map(|(pks, agg)| { + agg.as_lean(message, slot, pks.iter().map(|pk| pk.as_lean()).collect()) + }) .collect::>>()?; let _permit = PROVER_PERMIT.lock().unwrap(); - let agg = aggregate_single_message_signatures( - &children_arg, - raw_xmss, - *message.as_fixed_bytes(), - slot, - log_inv_rate, - )?; + let agg = aggregate(&children_arg, raw_xmss, vec![], &[], None, log_inv_rate)?; METRICS.get().map(|metrics| { metrics @@ -216,25 +231,17 @@ impl AggregatedSignature { .start_timer() }); - let mut expected_pubkeys = public_keys + let expected_pubkeys = public_keys .into_iter() .map(|k| k.as_lean()) .collect::>(); - expected_pubkeys.sort(); - expected_pubkeys.dedup(); - let agg = self.as_lean(expected_pubkeys)?; + // The `(message, slot)` binding travels out of band: it goes into the + // signer set the proof's digest commits to, so a proof bound to another + // message, slot, or key set fails inside the SNARK verifier. + let agg = self.as_lean(message, slot, expected_pubkeys)?; - if agg.info.core.message != *message.as_fixed_bytes() { - bail!("aggregated signature bound to a different message than expected"); - } - if agg.info.core.slot != slot { - bail!("aggregated signature bound to a different slot than expected"); - } - - let result = verify_single_message_aggregate(&agg) - .map(|_| ()) - .map_err(|err| anyhow!("{err:?}")); + let result = agg.verify().map_err(|err| anyhow!("{err:?}")); match &result { Ok(()) => { @@ -255,10 +262,15 @@ impl AggregatedSignature { pub(crate) fn as_lean( &self, + message: H256, + slot: u32, pubkeys: Vec, - ) -> Result { - SingleMessageAggregateSignature::from_bytes_without_pubkeys(self.0.as_bytes(), pubkeys) - .ok_or_else(|| anyhow!("invalid aggregated XMSS signature")) + ) -> Result { + EthereumProof::from_bytes_without_pubkeys( + self.0.as_bytes(), + one_group(message, slot, pubkeys), + ) + .map_err(|err| anyhow!("invalid aggregated XMSS signature: {err}")) } // todo(xmss): this is a function used only for testing. ideally, it should not exist @@ -272,13 +284,6 @@ impl AggregatedSignature { } } -fn sorted_dedup_lean_pubkeys(pks: &[PublicKey]) -> Vec { - let mut lean: Vec = pks.iter().map(|pk| pk.as_lean()).collect(); - lean.sort(); - lean.dedup(); - lean -} - impl Display for AggregatedSignature { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "0x{}", hex::encode(self.0.as_bytes())) diff --git a/lean_client/xmss/src/lib.rs b/lean_client/xmss/src/lib.rs index 06dda6e6..44dc5f51 100644 --- a/lean_client/xmss/src/lib.rs +++ b/lean_client/xmss/src/lib.rs @@ -11,7 +11,9 @@ mod signature; #[cfg(shadow_mode)] pub mod shadow_cost; -pub use aggregated_signature::{AggregatedSignature, set_prover_arena, setup_aggregation}; +pub use aggregated_signature::{ + AggregatedSignature, prover_busy, set_prover_arena, setup_aggregation, +}; pub use multi_message::MultiMessageAggregate; pub use public_key::PublicKey; pub use secret_key::SecretKey; diff --git a/lean_client/xmss/src/multi_message.rs b/lean_client/xmss/src/multi_message.rs index c71323fd..701af9a3 100644 --- a/lean_client/xmss/src/multi_message.rs +++ b/lean_client/xmss/src/multi_message.rs @@ -3,10 +3,7 @@ use std::str::FromStr; use anyhow::{Context, Error, Result, anyhow, bail}; use ethereum_types::H256; -use lean_multisig::{ - MultiMessageAggregateSignature, XmssPublicKey, merge_single_message_aggregates, - split_multi_message_aggregate, verify_multi_message_aggregate, -}; +use leanvm::{ClaimSelection, EthereumProof, SignatureClaims, XmssClaimGroup, aggregate}; use serde::{Deserialize, Deserializer, Serialize, Serializer, de}; use ssz::{ByteList, Ssz}; use typenum::U524288; @@ -30,8 +27,12 @@ impl MultiMessageAggregate { Ok(Self { proof }) } + /// Merge Type-1 components, each bound to its own `(message, slot)`, into + /// one Type-2 proof. leanVM groups claims by `(epoch, message)`, so a slot + /// may carry several messages; components sharing a `(slot, message)` have + /// their keys unioned into one group. pub fn aggregate( - parts: &[(&AggregatedSignature, &[PublicKey])], + parts: &[(&AggregatedSignature, &[PublicKey], H256, u32)], log_inv_rate: usize, ) -> Result { setup_aggregation(); @@ -45,7 +46,7 @@ impl MultiMessageAggregate { let merge_n = parts.len(); let count_bytes = merge_n.to_le_bytes(); let mut seed: Vec<&[u8]> = Vec::with_capacity(parts.len() + 1); - for (sig, _) in parts { + for (sig, _, _, _) in parts { seed.push(sig.as_bytes()); } seed.push(&count_bytes); @@ -58,17 +59,14 @@ impl MultiMessageAggregate { let parts_lean = parts .iter() - .map(|(sig, pks)| { - let mut lean: Vec = pks.iter().map(|pk| pk.as_lean()).collect(); - lean.sort(); - lean.dedup(); - sig.as_lean(lean) + .map(|(sig, pks, message, slot)| { + sig.as_lean(*message, *slot, pks.iter().map(|pk| pk.as_lean()).collect()) }) .collect::>>()?; let _permit = PROVER_PERMIT.lock().unwrap(); - let merged = merge_single_message_aggregates(parts_lean, log_inv_rate)?; + let merged = aggregate(&parts_lean, vec![], vec![], &[], None, log_inv_rate)?; let bytes = merged.to_bytes_without_pubkeys(); Self::new(&bytes) } @@ -80,51 +78,27 @@ impl MultiMessageAggregate { ) -> Result<()> { setup_aggregation(); - if pubkeys_per_message.len() != messages.len() { - bail!( - "binding length mismatch: {} pubkey sets vs {} messages", - pubkeys_per_message.len(), - messages.len() - ); - } - #[cfg(shadow_mode)] if crate::shadow_cost::fake_xmss() { return Ok(()); } - let pubkeys_per_info = sorted_dedup_lean_pubkeys(pubkeys_per_message); + let sig = self.as_lean(pubkeys_per_message, messages)?; - let sig = MultiMessageAggregateSignature::from_bytes_without_pubkeys( - self.proof.as_bytes(), - pubkeys_per_info, - ) - .ok_or_else(|| anyhow!("invalid multi-message aggregate bytes"))?; - - if sig.info.len() != messages.len() { - bail!( - "component count mismatch: proof has {}, expected {}", - sig.info.len(), - messages.len() - ); - } - for (i, (expected_message, expected_slot)) in messages.iter().enumerate() { - if sig.info[i].core.message != *expected_message.as_fixed_bytes() { - bail!("component {i} bound to a different message than expected"); - } - if sig.info[i].core.slot != *expected_slot { - bail!("component {i} bound to a different slot than expected"); - } - } - - verify_multi_message_aggregate(&sig) - .map(|_| ()) - .map_err(|err| anyhow!("{err:?}")) + sig.verify().map_err(|err| anyhow!("{err:?}")) } + /// Narrow the Type-2 down to the single claim bound to `message`, yielding + /// a Type-1 for it. leanVM does this by re-aggregating the parent with a + /// declaration of the one group to keep, so it generates a fresh SNARK. + /// + /// `messages` gives every `(message, slot)` claim the parent carries, in + /// the same order as `pubkeys_per_message`; the binding is not on the wire, + /// so decoding needs them all even though only one survives. pub fn split_by_message( &self, pubkeys_per_message: &[&[PublicKey]], + messages: &[(H256, u32)], message: H256, log_inv_rate: usize, ) -> Result { @@ -141,43 +115,89 @@ impl MultiMessageAggregate { return AggregatedSignature::new(&bytes); } - let pubkeys_per_info = sorted_dedup_lean_pubkeys(pubkeys_per_message); - let sig = MultiMessageAggregateSignature::from_bytes_without_pubkeys( - self.proof.as_bytes(), - pubkeys_per_info, - ) - .ok_or_else(|| anyhow!("invalid multi-message aggregate bytes"))?; + let sig = self.as_lean(pubkeys_per_message, messages)?; - let matches: Vec = sig - .info + // Groups are keyed by (epoch, message), so a message appears in one + // group unless two slots signed the very same bytes. + let matches: Vec<&XmssClaimGroup> = sig + .xmss_signers() .iter() - .enumerate() - .filter_map(|(i, info)| (info.core.message == *message.as_fixed_bytes()).then_some(i)) + .filter(|group| group.message == *message.as_fixed_bytes()) .collect(); - let index = match matches.as_slice() { - [i] => *i, + let group = match matches.as_slice() { + [group] => (*group).clone(), [] => bail!("split-by-message target not found in multi-message components"), _ => bail!("split-by-message target matched multiple components"), }; + let kept = SignatureClaims { + xmss: vec![group], + sphincs: Vec::new(), + }; + let declare = ClaimSelection { + signatures: &kept, + da_commitments: &[], + }; + let _permit = PROVER_PERMIT.lock().unwrap(); - let recovered = split_multi_message_aggregate(sig, index, log_inv_rate)?; + let recovered = aggregate(&[sig], vec![], vec![], &[], Some(declare), log_inv_rate)?; let bytes = recovered.to_bytes_without_pubkeys(); AggregatedSignature::new(&bytes) } + + fn as_lean( + &self, + pubkeys_per_message: &[&[PublicKey]], + messages: &[(H256, u32)], + ) -> Result { + let claims = wire_claims(pubkeys_per_message, messages)?; + EthereumProof::from_bytes_without_pubkeys(self.proof.as_bytes(), claims) + .map_err(|err| anyhow!("invalid multi-message aggregate bytes: {err}")) + } } -fn sorted_dedup_lean_pubkeys(per_component: &[&[PublicKey]]) -> Vec> { - per_component - .iter() - .map(|pks| { - let mut lean: Vec = pks.iter().map(|pk| pk.as_lean()).collect(); - lean.sort(); - lean.dedup(); - lean - }) - .collect() +/// The signer set leanVM binds a Type-2 aggregate to, built from the caller's +/// view of the claims: one group per `(slot, message)` pair, holding its +/// strictly sorted, deduplicated keys, with the groups themselves strictly +/// increasing on `(epoch, message)`. Claims sharing a `(slot, message)` merge +/// into one group; a slot may carry several messages, each its own group. +fn wire_claims( + pubkeys_per_message: &[&[PublicKey]], + messages: &[(H256, u32)], +) -> Result { + if pubkeys_per_message.len() != messages.len() { + bail!( + "binding length mismatch: {} pubkey sets vs {} messages", + pubkeys_per_message.len(), + messages.len() + ); + } + + let mut groups: Vec = Vec::with_capacity(messages.len()); + for ((message, slot), pks) in messages.iter().zip(pubkeys_per_message) { + let keys = pks.iter().map(|pk| pk.as_lean()); + match groups + .iter_mut() + .find(|group| group.epoch == *slot && group.message == *message.as_fixed_bytes()) + { + Some(group) => group.keys.extend(keys), + None => groups.push(XmssClaimGroup { + epoch: *slot, + message: *message.as_fixed_bytes(), + keys: keys.collect(), + }), + } + } + for group in &mut groups { + group.keys.sort_unstable(); + group.keys.dedup(); + } + groups.sort_unstable_by_key(|group| (group.epoch, group.message)); + Ok(SignatureClaims { + xmss: groups, + sphincs: Vec::new(), + }) } impl Display for MultiMessageAggregate { diff --git a/lean_client/xmss/src/public_key.rs b/lean_client/xmss/src/public_key.rs index ce444bcc..417de599 100644 --- a/lean_client/xmss/src/public_key.rs +++ b/lean_client/xmss/src/public_key.rs @@ -6,7 +6,7 @@ use core::{ use anyhow::{Error, anyhow}; use eth_ssz::{Decode as _, DecodeError, Encode as _}; -use lean_multisig::XmssPublicKey; +use leanvm::xmss::XmssPublicKey; use serde::{ Deserialize, Serialize, de::{self, Visitor}, diff --git a/lean_client/xmss/src/secret_key.rs b/lean_client/xmss/src/secret_key.rs index a26b9dc9..89360ccd 100644 --- a/lean_client/xmss/src/secret_key.rs +++ b/lean_client/xmss/src/secret_key.rs @@ -1,6 +1,6 @@ use anyhow::{Error, Result, anyhow}; use derive_more::Debug; -use lean_multisig::{XmssSecretKey, xmss_key_gen, xmss_sign}; +use leanvm::xmss::{self, XmssSecretKey}; use rand::CryptoRng; use ssz::H256; @@ -16,11 +16,11 @@ pub struct SecretKey(XmssSecretKey); impl SecretKey { pub fn sign(&self, message: H256, epoch: u32) -> Result { - if !self.0.activation_slots().contains(&epoch) { + if !self.0.epoch_range().contains(&epoch) { return Err(anyhow!("epoch {epoch} outside key activation window")); } - let sig = xmss_sign(&self.0, epoch, message.as_fixed_bytes()) + let sig = xmss::sign(&self.0, message.as_fixed_bytes(), epoch) .map_err(|err| anyhow!("failed to sign message: {err:?}"))?; Ok(Signature::from_lean(sig)) } @@ -30,7 +30,18 @@ impl SecretKey { activation_epoch: u32, num_active_epochs: u32, ) -> (PublicKey, SecretKey) { - let (pk, sk) = xmss_key_gen(rng, activation_epoch as u64, num_active_epochs as u64) + let epoch_end = num_active_epochs + .checked_sub(1) + .and_then(|len| activation_epoch.checked_add(len)) + .expect("activation range must be non-empty and fit within the 2^32 key lifetime"); + + // leanVM's `key_gen` draws its seed from leanVM's own (older) `rand`, + // whose traits the caller's rng does not implement; drawing the seed + // here and using the deterministic entry point sidesteps the mismatch. + let mut seed = [0u8; 32]; + rng.fill_bytes(&mut seed); + + let (sk, pk) = xmss::key_gen_from_seed(seed, activation_epoch, epoch_end) .expect("activation range must fit within the 2^32 key lifetime"); (PublicKey::from_lean(pk), SecretKey(sk)) } diff --git a/lean_client/xmss/src/signature.rs b/lean_client/xmss/src/signature.rs index efc7176a..8b685b5e 100644 --- a/lean_client/xmss/src/signature.rs +++ b/lean_client/xmss/src/signature.rs @@ -7,7 +7,7 @@ use core::{ use crate::public_key::PublicKey; use anyhow::{Error, Result, anyhow}; use eth_ssz::{Decode as _, DecodeError, Encode as _}; -use lean_multisig::{XmssSignature, xmss_verify}; +use leanvm::xmss::{self, XmssSignature}; use metrics::METRICS; use serde::de::{self, Visitor}; use serde::{Deserialize, Serialize}; @@ -35,11 +35,11 @@ impl Signature { } pub fn verify(&self, public_key: &PublicKey, epoch: u32, message: H256) -> Result<()> { - match xmss_verify( + match xmss::verify( &public_key.as_lean(), - epoch, message.as_fixed_bytes(), &self.as_lean(), + epoch, ) { Ok(()) => { METRICS.get().map(|metrics| {