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

bin/ethlambda/src/cli.rs

@@ -83,8 +83,45 @@ 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.
+    /// Every ID must be below --attestation-committee-count; the node refuses to
+    /// start otherwise, since a higher ID names a topic no validator publishes on.
+    ///
+    /// 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.
+    ///
+    /// Deployment precondition: give every subnet below
+    /// --attestation-committee-count an aggregator holding it as its duty
+    /// subnet. Ownership is `duty_subnet % width == slot % width`, so on a
+    /// sparser placement a width can have no owner at all while every
+    /// configured node is healthy. With duty subnets {0, 2} at committee count
+    /// 4, nothing owns width 4 in an odd slot, and since this flag also
+    /// disables the full-width fallback, that merge level is simply dropped
+    /// for the slot. The narrower levels still run and the next slot rotates to
+    /// a different owner, but the loss is structural, not just the cost of a
+    /// node that is down or late.
+    #[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,
@@ -217,6 +217,12 @@ async fn run_node(options: NodeOptions) -> eyre::Result<()> {
         attestation_committee_count,
         "Loaded attestation committee count"
     );
+    // Checked here rather than in clap: the committee count is only known once
+    // the CLI flag and the validator config have both been consulted.
+    validate_aggregate_subnet_ids(
+        options.aggregate_subnet_ids.as_deref(),
+        attestation_committee_count,
+    )?;
     ethlambda_blockchain::metrics::set_attestation_committee_count(attestation_committee_count);
 
     let bootnodes = read_bootnodes(&bootnodes_path)?;
@@ -279,12 +285,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 +846,132 @@ async fn fetch_initial_state(
     Ok(store)
 }
 
+/// Reject an `--aggregate-subnet-ids` value that names no subnet.
+///
+/// The flag feeds two consumers that read an out-of-range value differently:
+/// the P2P swarm subscribes to the raw id (`attestation_subscription_subnets`
+/// passes it through), while the aggregation window reduces it modulo the
+/// committee count. `--attestation-committee-count 4 --aggregate-subnet-ids 5`
+/// therefore subscribes to a topic no validator publishes on and aggregates as
+/// duty subnet 1, which the node does not listen to, with the startup log
+/// showing 5 either way. Refusing to start is the only reading of that
+/// configuration that cannot silently mean something else.
+fn validate_aggregate_subnet_ids(
+    assigned_subnet_ids: Option<&[u64]>,
+    attestation_committee_count: u64,
+) -> eyre::Result<()> {
+    let out_of_range = assigned_subnet_ids
+        .unwrap_or_default()
+        .iter()
+        .find(|&&id| id >= attestation_committee_count);
+    match out_of_range {
+        None => Ok(()),
+        Some(id) => Err(eyre::eyre!(
+            "--aggregate-subnet-ids value {id} is not a subnet: ids must be below \
+             attestation_committee_count ({attestation_committee_count})"
+        )),
+    }
+}
+
+/// 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
+        );
+    }
+
+    /// An id at or above the committee count names a topic no validator
+    /// publishes on, and would be reduced to a different subnet by the
+    /// aggregation window, so the node refuses it rather than running with
+    /// its subscriptions and its duty subnet disagreeing.
+    #[test]
+    fn an_out_of_range_aggregate_subnet_id_is_rejected() {
+        let err = validate_aggregate_subnet_ids(Some(&[5]), 4)
+            .expect_err("subnet 5 does not exist at committee count 4");
+        let message = err.to_string();
+        assert!(message.contains('5'), "names the offending id: {message}");
+        assert!(message.contains('4'), "names the bound: {message}");
+    }
+
+    /// The check covers every id, not just the first: only the first becomes
+    /// the duty subnet, but all of them become gossip subscriptions.
+    #[test]
+    fn an_out_of_range_aggregate_subnet_id_is_rejected_past_the_first() {
+        assert!(validate_aggregate_subnet_ids(Some(&[0, 9]), 4).is_err());
+    }
+
+    /// The bound is exclusive: subnets run 0..committee_count.
+    #[test]
+    fn in_range_aggregate_subnet_ids_are_accepted() {
+        assert!(validate_aggregate_subnet_ids(Some(&[0, 3]), 4).is_ok());
+        assert!(
+            validate_aggregate_subnet_ids(Some(&[0]), 1).is_ok(),
+            "subnet 0 is the only subnet at a committee count of 1"
+        );
+    }
+
+    /// Nothing assigned is nothing to validate; the duty subnet is then
+    /// derived from subscriptions, which are already in range by construction.
+    #[test]
+    fn an_unset_or_empty_assignment_passes_validation() {
+        assert!(validate_aggregate_subnet_ids(None, 4).is_ok());
+        assert!(validate_aggregate_subnet_ids(Some(&[]), 4).is_ok());
+    }
+
     /// Validator-config snippet matching `lean-quickstart`'s ansible-devnet
     /// where networks share a non-default committee count.
     const VC_WITH_COMMITTEE_COUNT: &str = r#"