feat(aggregation): score recursive aggregation through a per-aggregator subnet window

bin/ethlambda/src/cli.rs

@@ -83,8 +83,35 @@ pub(crate) struct NodeOptions {
     pub(crate) attestation_committee_count: Option<u64>,
     /// Subnet IDs this aggregator should subscribe to (comma-separated).
     /// Requires --is-aggregator. Defaults to the subnets of the node's validators.
+    ///
+    /// The first ID is also this node's aggregation duty subnet: where its
+    /// aggregation window starts, and what --skip-redundant-aggregation
+    /// rotates ownership over. Order matters, so give co-located aggregators
+    /// different first IDs. Unset, the duty subnet falls back to the lowest
+    /// subscribed subnet, which is the same value on every node whose
+    /// validators span all subnets.
     #[arg(long, value_delimiter = ',', requires = "is_aggregator")]
     pub(crate) aggregate_subnet_ids: Option<Vec<u64>>,
+    /// Sit out aggregation candidates whose level another duty subnet owns
+    /// this slot. Requires --is-aggregator.
+    ///
+    /// By default every aggregator merges proofs for a window of subnets
+    /// starting at its duty subnet, and windows belonging to neighbouring duty
+    /// subnets overlap, so some prover work is duplicated. With this flag an
+    /// aggregator skips a candidate whose width it does not own in the current
+    /// slot and spends that job on the next-best attestation data instead. The
+    /// owner rotates with the slot, so no node is permanently the one sitting
+    /// out, and the narrowest width is owned by everyone, so a candidate whose
+    /// pool holds nothing on this node's subnet, which is the raw-signature
+    /// case, is never skipped.
+    ///
+    /// Worth enabling when leanVM prover CPU is the bottleneck on co-located
+    /// aggregators. The cost is that a level in a given slot has few
+    /// producers, so a node that is down or late forfeits that slot's merge at
+    /// its level; the narrower levels still run and the next slot rotates to a
+    /// different owner.
+    #[arg(long, default_value = "false", requires = "is_aggregator")]
+    pub(crate) skip_redundant_aggregation: bool,
     /// Directory for RocksDB storage
     #[arg(long, default_value = "./data")]
     pub(crate) data_dir: PathBuf,

bin/ethlambda/src/main.rs

@@ -25,7 +25,7 @@ static ALLOC: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
 static malloc_conf: &[u8] = b"prof:true,prof_active:true,lg_prof_sample:19\0";
 
 use std::{
-    collections::{BTreeMap, HashMap},
+    collections::{BTreeMap, HashMap, HashSet},
     net::{IpAddr, SocketAddr},
     path::{Path, PathBuf},
     sync::Arc,
@@ -279,12 +279,34 @@ async fn run_node(options: NodeOptions) -> eyre::Result<()> {
     // receiver-count guard in `emit` makes every emission a no-op.
     let events = EventBus::default();
 
+    let aggregation_duty_subnet = resolve_aggregation_duty_subnet(
+        options.aggregate_subnet_ids.as_deref(),
+        &subscribed_subnets,
+    );
+    info!(
+        aggregation_duty_subnet,
+        assigned = options.aggregate_subnet_ids.is_some(),
+        "Resolved aggregation duty subnet"
+    );
+    if options.skip_redundant_aggregation && options.aggregate_subnet_ids.is_none() {
+        warn!(
+            aggregation_duty_subnet,
+            "--skip-redundant-aggregation is set but the duty subnet was derived, not assigned: \
+             every co-located aggregator whose validators span all subnets derives the same duty \
+             subnet, so they will sit out in lockstep in the same slot instead of taking turns, \
+             and the widest level gets no producer at all in most slots. Give each aggregator a \
+             distinct first --aggregate-subnet-ids value to fix this."
+        );
+    }
+
     let blockchain_config = BlockChainConfig {
         aggregator: aggregator.clone(),
         sync_status_controller: sync_status.clone(),
         attestation_committee_count,
         gate_duties: !options.disable_duty_sync_gate,
         subscribed_subnets: subscribed_subnets.clone(),
+        aggregation_duty_subnet,
+        skip_redundant_aggregation: options.skip_redundant_aggregation,
         proposer_config: ProposerConfig {
             enable_proposer_aggregation: options.enable_proposer_aggregation,
             max_attestations_per_block: options.max_attestations_per_block,
@@ -818,13 +840,67 @@ async fn fetch_initial_state(
     Ok(store)
 }
 
+/// The subnet this node is responsible for when scoring recursive aggregation.
+///
+/// Operators assign it explicitly via --aggregate-subnet-ids so co-located
+/// aggregators land on different subnets and merge different proofs. Without
+/// an assignment, fall back to the lowest subnet this node already listens
+/// on: `min` rather than an arbitrary pick because `HashSet` iteration order
+/// is not stable and the duty subnet must be.
+fn resolve_aggregation_duty_subnet(
+    assigned_subnet_ids: Option<&[u64]>,
+    subscribed_subnets: &HashSet<u64>,
+) -> u64 {
+    assigned_subnet_ids
+        .and_then(|ids| ids.first().copied())
+        .or_else(|| subscribed_subnets.iter().copied().min())
+        .unwrap_or(0)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use ethlambda_storage::backend::InMemoryBackend;
     use ethlambda_types::constants::DEFAULT_MILLISECONDS_PER_SLOT;
     use ethlambda_types::genesis::GenesisValidatorEntry;
 
+    /// The duty subnet is the first explicitly assigned subnet, so an operator
+    /// can place co-located aggregators on different subnets deliberately.
+    #[test]
+    fn duty_subnet_prefers_the_first_assigned_id() {
+        let subscribed = HashSet::from([0u64, 1, 2, 3]);
+        assert_eq!(
+            resolve_aggregation_duty_subnet(Some(&[3, 1]), &subscribed),
+            3,
+            "the first assigned id wins, not the lowest"
+        );
+    }
+
+    /// With no assignment, the lowest subscribed subnet is used, which is
+    /// stable across restarts unlike an arbitrary pick from the set.
+    #[test]
+    fn duty_subnet_falls_back_to_the_lowest_subscribed() {
+        let subscribed = HashSet::from([5u64, 2]);
+        assert_eq!(resolve_aggregation_duty_subnet(None, &subscribed), 2);
+    }
+
+    /// A node with nothing assigned and nothing subscribed still needs an
+    /// answer; subnet 0 always exists.
+    #[test]
+    fn duty_subnet_defaults_to_zero_with_nothing_to_go_on() {
+        assert_eq!(resolve_aggregation_duty_subnet(None, &HashSet::new()), 0);
+    }
+
+    /// An empty list is no assignment at all, so the subscription fallback
+    /// still applies rather than the last-resort zero.
+    #[test]
+    fn duty_subnet_treats_an_empty_assignment_as_no_assignment() {
+        assert_eq!(
+            resolve_aggregation_duty_subnet(Some(&[]), &HashSet::from([4u64])),
+            4
+        );
+    }
+
     /// Validator-config snippet matching `lean-quickstart`'s ansible-devnet
     /// where networks share a non-default committee count.
     const VC_WITH_COMMITTEE_COUNT: &str = r#"