track leanVM main, add gossip arrival metrics

lean_client/Cargo.toml

@@ -253,10 +253,8 @@ indexmap = "2"
 http-body-util = "0.1"
 http_api_utils = { git = "https://github.com/grandinetech/grandine", rev = "c4b676e3daa0ddcb86d0bcb321b7166d59f920f9" }
 k256 = "0.13"
-rec_aggregation = { git = "https://github.com/leanEthereum/leanVM.git", rev = "e2592df4e30fdddbbf8ae26a333116c68cec7026" }
-backend = { git = "https://github.com/leanEthereum/leanVM.git", rev = "e2592df4e30fdddbbf8ae26a333116c68cec7026" }
-leansig = { git = "https://github.com/leanEthereum/leanSig", branch = "devnet4" }
-leansig_wrapper = { git = "https://github.com/leanEthereum/leanVM.git", rev = "e2592df4e30fdddbbf8ae26a333116c68cec7026" }
+lean-multisig = { git = "https://github.com/leanEthereum/leanVM.git", rev = "a5909d18647de6aed38640c098d9177fab2bf36a" }
+postcard = { version = "1.1.3", features = ["alloc"] }
 libp2p = { git = "https://github.com/libp2p/rust-libp2p.git", rev = "91e8931e275bcd1c72791d18b09fea8b77209baf", default-features = false, features = [
     'dns',
     'gossipsub',

lean_client/metrics/src/lib.rs

@@ -3,5 +3,9 @@ mod metrics;
 mod server;
 
 pub use helpers::{set_gauge_u64, stop_and_discard, stop_and_record};
-pub use metrics::{DisconnectReason, METRICS, Metrics};
+pub use metrics::{
+    DisconnectReason, METRICS, Metrics, observe_gossip_aggregation_arrival,
+    observe_gossip_attestation_arrival, observe_gossip_block_arrival, set_gossip_arrival_clock,
+    unix_now_ms,
+};
 pub use server::{MetricsServerConfig, run_server};

lean_client/metrics/src/metrics.rs

@@ -266,6 +266,13 @@ pub struct Metrics {
     pub lean_gossip_block_size_bytes: Histogram,
     pub lean_gossip_attestation_size_bytes: Histogram,
     pub lean_gossip_aggregation_size_bytes: Histogram,
+
+    pub lean_gossip_block_arrival_delay_seconds: Histogram,
+    pub lean_gossip_attestation_arrival_delay_seconds: Histogram,
+    pub lean_gossip_aggregation_arrival_delay_seconds: Histogram,
+    pub lean_gossip_block_arrival_total: IntCounterVec,
+    pub lean_gossip_attestation_arrival_total: IntCounterVec,
+    pub lean_gossip_aggregation_arrival_total: IntCounterVec,
 }
 
 impl Metrics {
@@ -781,6 +788,54 @@ impl Metrics {
                     1_048_576.0,
                 ]
             ))?,
+
+            // Gossip Arrival Metrics
+            lean_gossip_block_arrival_delay_seconds: Histogram::with_opts(histogram_opts!(
+                "lean_gossip_block_arrival_delay_seconds",
+                "Absolute delay between a gossip block's arrival and the start of the interval \
+                 it was due in",
+                gossip_arrival_delay_buckets()
+            ))?,
+            lean_gossip_attestation_arrival_delay_seconds: Histogram::with_opts(histogram_opts!(
+                "lean_gossip_attestation_arrival_delay_seconds",
+                "Absolute delay between a gossip attestation's arrival and the start of the \
+                 interval it was due in",
+                gossip_arrival_delay_buckets()
+            ))?,
+            lean_gossip_aggregation_arrival_delay_seconds: Histogram::with_opts(histogram_opts!(
+                "lean_gossip_aggregation_arrival_delay_seconds",
+                "Absolute delay between an aggregate becoming available, whether received on \
+                 gossip or produced locally, and the most recent aggregation-interval boundary \
+                 at or before it. Local aggregation starts at that boundary, so a locally \
+                 produced aggregate reports its proving latency past it",
+                gossip_arrival_delay_buckets()
+            ))?,
+            lean_gossip_block_arrival_total: IntCounterVec::new(
+                opts!(
+                    "lean_gossip_block_arrival_total",
+                    "Gossip blocks by arrival position relative to the interval they were due in"
+                ),
+                &["position"],
+            )?,
+            lean_gossip_attestation_arrival_total: IntCounterVec::new(
+                opts!(
+                    "lean_gossip_attestation_arrival_total",
+                    "Gossip attestations by arrival position relative to the interval they were \
+                     due in"
+                ),
+                &["position"],
+            )?,
+            lean_gossip_aggregation_arrival_total: IntCounterVec::new(
+                opts!(
+                    "lean_gossip_aggregation_arrival_total",
+                    "Aggregates, received on gossip or produced locally, by arrival position \
+                     relative to the most recent aggregation-interval boundary. Anchored to the \
+                     latest such boundary rather than the aggregate's own data slot, so an \
+                     arrival can never precede it: only `inside` and `after` occur, never \
+                     `before`."
+                ),
+                &["position"],
+            )?,
         })
     }
 
@@ -1012,6 +1067,31 @@ impl Metrics {
         default_registry.register(Box::new(self.lean_gossip_attestation_size_bytes.clone()))?;
         default_registry.register(Box::new(self.lean_gossip_aggregation_size_bytes.clone()))?;
 
+        // Gossip Arrival Metrics
+        default_registry.register(Box::new(
+            self.lean_gossip_block_arrival_delay_seconds.clone(),
+        ))?;
+        default_registry.register(Box::new(
+            self.lean_gossip_attestation_arrival_delay_seconds.clone(),
+        ))?;
+        default_registry.register(Box::new(
+            self.lean_gossip_aggregation_arrival_delay_seconds.clone(),
+        ))?;
+        default_registry.register(Box::new(self.lean_gossip_block_arrival_total.clone()))?;
+        default_registry.register(Box::new(self.lean_gossip_attestation_arrival_total.clone()))?;
+        default_registry.register(Box::new(self.lean_gossip_aggregation_arrival_total.clone()))?;
+
+        for position in ["before", "inside", "after"] {
+            self.lean_gossip_block_arrival_total
+                .with_label_values(&[position]);
+            self.lean_gossip_attestation_arrival_total
+                .with_label_values(&[position]);
+        }
+        for position in ["inside", "after"] {
+            self.lean_gossip_aggregation_arrival_total
+                .with_label_values(&[position]);
+        }
+
         Ok(())
     }
 
@@ -1101,3 +1181,113 @@ pub enum DisconnectReason {
     LocalClose,
     Error,
 }
+
+fn gossip_arrival_delay_buckets() -> Vec<f64> {
+    vec![0.05, 0.1, 0.2, 0.4, 0.8, 1.2, 1.6, 2.4, 4.0, 8.0, 16.0]
+}
+
+const BLOCK_INTERVAL_INDEX: u64 = 0;
+const ATTESTATION_INTERVAL_INDEX: u64 = 1;
+const AGGREGATION_INTERVAL_INDEX: u64 = 2;
+
+#[derive(Clone, Copy, Debug)]
+struct GossipArrivalClock {
+    genesis_ms: u64,
+    millis_per_interval: u64,
+    millis_per_slot: u64,
+}
+
+static GOSSIP_ARRIVAL_CLOCK: OnceCell<GossipArrivalClock> = OnceCell::new();
+
+pub fn set_gossip_arrival_clock(
+    genesis_ms: u64,
+    millis_per_interval: u64,
+    intervals_per_slot: u64,
+) {
+    let _ = GOSSIP_ARRIVAL_CLOCK.set(GossipArrivalClock {
+        genesis_ms,
+        millis_per_interval,
+        millis_per_slot: millis_per_interval * intervals_per_slot,
+    });
+}
+
+pub fn unix_now_ms() -> u64 {
+    SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0)
+}
+
+fn interval_delta_ms(
+    clock: GossipArrivalClock,
+    arrival_ms: u64,
+    anchor_slot: u64,
+    interval_index: u64,
+) -> i64 {
+    let expected_ms = clock.genesis_ms
+        + anchor_slot * clock.millis_per_slot
+        + interval_index * clock.millis_per_interval;
+    arrival_ms as i64 - expected_ms as i64
+}
+
+fn latest_interval_delta_ms(
+    clock: GossipArrivalClock,
+    arrival_ms: u64,
+    interval_index: u64,
+) -> i64 {
+    let since_genesis = arrival_ms.saturating_sub(clock.genesis_ms) as i64;
+    let anchor_offset = (interval_index * clock.millis_per_interval) as i64;
+    (since_genesis - anchor_offset).rem_euclid(clock.millis_per_slot as i64)
+}
+
+fn position_from_delta(clock: GossipArrivalClock, delta_ms: i64) -> &'static str {
+    if delta_ms < 0 {
+        "before"
+    } else if delta_ms < clock.millis_per_interval as i64 {
+        "inside"
+    } else {
+        "after"
+    }
+}
+
+pub fn observe_gossip_block_arrival(arrival_ms: u64, block_slot: u64) {
+    let (Some(clock), Some(metrics)) = (GOSSIP_ARRIVAL_CLOCK.get(), METRICS.get()) else {
+        return;
+    };
+    let delta_ms = interval_delta_ms(*clock, arrival_ms, block_slot, BLOCK_INTERVAL_INDEX);
+    metrics
+        .lean_gossip_block_arrival_delay_seconds
+        .observe(delta_ms.unsigned_abs() as f64 / 1000.0);
+    metrics
+        .lean_gossip_block_arrival_total
+        .with_label_values(&[position_from_delta(*clock, delta_ms)])
+        .inc();
+}
+
+pub fn observe_gossip_attestation_arrival(arrival_ms: u64, data_slot: u64) {
+    let (Some(clock), Some(metrics)) = (GOSSIP_ARRIVAL_CLOCK.get(), METRICS.get()) else {
+        return;
+    };
+    let delta_ms = interval_delta_ms(*clock, arrival_ms, data_slot, ATTESTATION_INTERVAL_INDEX);
+    metrics
+        .lean_gossip_attestation_arrival_delay_seconds
+        .observe(delta_ms.unsigned_abs() as f64 / 1000.0);
+    metrics
+        .lean_gossip_attestation_arrival_total
+        .with_label_values(&[position_from_delta(*clock, delta_ms)])
+        .inc();
+}
+
+pub fn observe_gossip_aggregation_arrival(arrival_ms: u64) {
+    let (Some(clock), Some(metrics)) = (GOSSIP_ARRIVAL_CLOCK.get(), METRICS.get()) else {
+        return;
+    };
+    let delta_ms = latest_interval_delta_ms(*clock, arrival_ms, AGGREGATION_INTERVAL_INDEX);
+    metrics
+        .lean_gossip_aggregation_arrival_delay_seconds
+        .observe(delta_ms.unsigned_abs() as f64 / 1000.0);
+    metrics
+        .lean_gossip_aggregation_arrival_total
+        .with_label_values(&[position_from_delta(*clock, delta_ms)])
+        .inc();
+}

lean_client/networking/src/network/service.rs

@@ -639,12 +639,14 @@ where
                 ..
             } => {
                 let data_len = message.data.len();
+                let arrival_ms = metrics::unix_now_ms();
                 match GossipsubMessage::decode(&message.topic, &message.data) {
                     Ok(GossipsubMessage::Block(signed_block)) => {
                         METRICS
                             .get()
                             .map(|m| m.lean_gossip_block_size_bytes.observe(data_len as f64));
                         let slot = signed_block.block.slot.0;
+                        metrics::observe_gossip_block_arrival(arrival_ms, slot);
                         info!(slot, block_root = %signed_block.block.hash_tree_root(), "received block via gossip");
 
                         if let Err(err) = self
@@ -677,6 +679,7 @@ where
                             "received attestation via subnet gossip"
                         );
                         let slot = attestation.message.slot.0;
+                        metrics::observe_gossip_attestation_arrival(arrival_ms, slot);
 
                         if let Err(err) = self
                             .chain_message_sink
@@ -702,6 +705,7 @@ where
                             "received aggregated attestation via gossip"
                         );
                         let slot = signed_aggregated_attestation.data.slot.0;
+                        metrics::observe_gossip_aggregation_arrival(arrival_ms);
 
                         if let Err(err) = self
                             .chain_message_sink

lean_client/src/main.rs

@@ -430,6 +430,13 @@ struct Args {
     #[arg(long, default_value_t = false)]
     enable_proposer_aggregation: bool,
 
+    /// Run the leanVM prover on its bump arena: faster proving, but the arena
+    /// never returns pages to the OS, so RSS ratchets to the allocation
+    /// high-water mark for the process lifetime. Off by default (system
+    /// allocator: slower proving, bounded memory).
+    #[arg(long, default_value_t = false)]
+    prover_arena: bool,
+
     #[cfg(shadow_mode)]
     #[command(flatten)]
     shadow: ShadowOptions,
@@ -461,10 +468,6 @@ struct ShadowOptions {
 #[cfg_attr(shadow_mode, tokio::main(flavor = "current_thread"))]
 #[cfg_attr(not(shadow_mode), tokio::main)]
 async fn main() -> Result<()> {
-    let rayon_threads = num_cpus::get().saturating_sub(3).max(1);
-    xmss::configure_rayon_pool(rayon_threads);
-    xmss::setup_aggregation();
-
     tracing_subscriber::fmt()
         .with_ansi(std::io::stdout().is_terminal())
         .with_env_filter(
@@ -482,6 +485,9 @@ async fn main() -> Result<()> {
 
     let args = Args::parse();
 
+    xmss::set_prover_arena(args.prover_arena);
+    xmss::setup_aggregation();
+
     #[cfg(shadow_mode)]
     {
         let s = &args.shadow;
@@ -642,6 +648,8 @@ async fn main() -> Result<()> {
 
     let config = Config { genesis_time };
 
+    metrics::set_gossip_arrival_clock(genesis_time * 1000, MILLIS_PER_INTERVAL, INTERVALS_PER_SLOT);
+
     // ── Anchor state: download checkpoint or use genesis ─────────────────────────────────────
     // For checkpoint sync: state is downloaded now; the anchor block is fetched from the
     // network after the network service starts.  `checkpoint_block_root` holds the expected
@@ -1583,6 +1591,7 @@ async fn main() -> Result<()> {
                     aggregator.as_mut().unwrap().recv().await
                 }, if aggregator.is_some() => {
                     if let Some((aggregations, consumed_data_roots)) = maybe_agg {
+                        let arrival_ms = metrics::unix_now_ms();
                         let mut to_publish = Vec::with_capacity(aggregations.len());
                         {
                             let mut s = store.write();
@@ -1603,6 +1612,7 @@ async fn main() -> Result<()> {
                             });
                         }
                         for aggregation in to_publish {
+                            metrics::observe_gossip_aggregation_arrival(arrival_ms);
                             if let Err(e) = chain_outbound_sender.send(
                                 OutboundP2pRequest::GossipAggregation(aggregation)
                             ) {

lean_client/xmss/Cargo.toml

@@ -12,17 +12,13 @@ derive_more = { workspace = true }
 eth_ssz = { workspace = true }
 ethereum-types = { workspace = true }
 hex = { workspace = true }
-rec_aggregation = { workspace = true }
-backend = { workspace = true }
-leansig = { workspace = true }
-leansig_wrapper = { workspace = true }
+lean-multisig = { workspace = true }
 metrics = { workspace = true }
+postcard = { workspace = true }
 rand = { workspace = true }
-rayon = { workspace = true }
 ssz = { workspace = true }
 typenum = { workspace = true }
 serde = { workspace = true }
-zeroize = { workspace = true, features = ["derive"] }
 
 [dev-dependencies]
 rand_chacha = { workspace = true }