[WIP] Committee aggregation

docs/client/networking.md

@@ -33,6 +33,7 @@ Each node entry contains an ENR. This is an Ethereum Node Record. It includes:
 - The node's public key
 - Network address
 - Port numbers
+- Committee assignments (for aggregators)
 - Other metadata
 
 In production, dynamic discovery would replace static configuration.
@@ -62,15 +63,35 @@ Messages are organized by topic. Topic names follow a pattern that includes:
 
 This structure lets clients subscribe to relevant messages and ignore others.
 
+The payload carried in the gossipsub message is the SSZ-encoded, 
+Snappy-compressed message, which type is identified by the topic:
+
+| Topic Name                                                 | Message Type                | Encoding     |
+|------------------------------------------------------------|-----------------------------|--------------|
+| /lean/consensus/devnet3/blocks/ssz_snappy                  | SignedBlockWithAttestation  | SSZ + Snappy |
+| /lean/consensus/devnet3/attestations/ssz_snappy            | SignedAttestation           | SSZ + Snappy |
+| /lean/consensus/devnet3/attestation_{subnet_id}/ssz_snappy | SignedAttestation           | SSZ + Snappy |
+| /lean/consensus/devnet3/aggregation/ssz_snappy             | SignedAggregatedAttestation | SSZ + Snappy |
+
 ### Message Types
 
-Two main message types exist:
+Three main message types exist:
+
+* _Blocks_, defined by the `SignedBlockWithAttestation` type, are proposed by 
+validators and propagated on the block topic. Every node needs to see blocks 
+quickly.
 
-Blocks are proposed by validators. They propagate on the block topic. Every
-node needs to see blocks quickly.
+* _Attestations_, defined by the `SignedAttestation` type, come from all 
+validators. They propagate on the global attestation topic. Additionally,
+each committee has its own attestation topic. Validators publish to their
+committee's attestation topic and global attestation topic. Non-aggregating 
+validators subscribe only to the global attestation topic, while aggregators 
+subscribe to both the global and their committee's attestation topic.
 
-Attestations come from all validators. They propagate on the attestation topic. High volume
-but small messages.
+* _Committee aggregations_, defined by the `SignedAggregatedAttestation` type, 
+created by committee aggregators. These combine attestations from committee 
+members. Aggregations propagate on the aggregation topic to which every 
+validator subscribes.
 
 ### Encoding
 

docs/client/validator.md

@@ -2,8 +2,9 @@
 
 ## Overview
 
-Validators participate in consensus by proposing blocks and producing attestations. This
-document describes what honest validators do.
+Validators participate in consensus by proposing blocks and producing attestations. 
+Optionally validators can opt-in to behave as aggregators in their committee . 
+This document describes what honest validators do.
 
 ## Validator Assignment
 
@@ -16,6 +17,32 @@ diversity helps test interoperability.
 In production, validator assignment will work differently. The current approach
 is temporary for devnet testing.
 
+## Attestation Committees and Subnets
+
+Attestation committee is a group of validators contributing to the common 
+aggregated attestations. Subnets are network channels dedicated to specific committees.
+
+In the devnet-3 design, however, there is one global subnet for signed 
+attestations propagation, in addition to publishing into per committee subnets.
+This is due to 3SF-mini consensus design, that requires 2/3+ of all 
+attestations to be observed by any validator to compute safe target correctly.
+
+Note that non-aggregating validators do not need to subscribe to committee
+attestation subnets. They only need to subscribe to the global attestation 
+subnet.
+
+Every validator is assigned to a single committee. Number of committees is 
+defined in config.yaml. Each committee maps to a subnet ID. Validator's
+subnet ID is derived using their validator index modulo number of committees.
+This is to simplify debugging and testing. In the future, validator's subnet ID
+will be assigned randomly per epoch.
+
+## Aggregator assignment
+
+Some validators are self-assigned as aggregators. Aggregators collect and combine
+attestations from other validators in their committee. To become an aggregator,
+a validator sets `is_aggregator` flag to true as ENR record field.
+
 ## Proposing Blocks
 
 Each slot has exactly one designated proposer. The proposer is determined by
@@ -52,7 +79,7 @@ receive and validate it.
 
 ## Attesting
 
-Every validator attestations in every slot. Attesting happens in the second interval,
+Every validator attests in every slot. Attesting happens in the second interval,
 after proposals are made.
 
 ### What to Attest For
@@ -78,8 +105,8 @@ compute the head.
 
 ### Broadcasting Attestations
 
-Validators sign their attestations and broadcast them. The network uses a single topic
-for all attestations. No subnets or committees in the current design.
+Validators sign their attestations and broadcast them into the global 
+attestation topic and its corresponding subnet topic.
 
 ## Timing
 
@@ -98,11 +125,7 @@ blocks and attestations.
 Attestation aggregation combines multiple attestations into one. This saves bandwidth and
 block space.
 
-Devnet 0 has no aggregation. Each attestation is separate. Future devnets will add
-aggregation.
-
-When aggregation is added, aggregators will collect attestations and combine them.
-Aggregated attestations will be broadcast separately.
+Devnet-3 introduces signatures aggregation. Aggregators will collect attestations and combine them. Aggregated attestations will be broadcast separately.
 
 ## Signature Handling
 

src/lean_spec/subspecs/chain/config.py

@@ -37,6 +37,9 @@
 VALIDATOR_REGISTRY_LIMIT: Final = Uint64(2**12)
 """The maximum number of validators that can be in the registry."""
 
+ATTESTATION_COMMITTEE_COUNT: Final = Uint64(1)
+"""The number of attestation committees per slot."""
+
 
 class _ChainConfig(StrictBaseModel):
     """

src/lean_spec/subspecs/containers/attestation/attestation.py

@@ -20,6 +20,7 @@
 from lean_spec.subspecs.ssz import hash_tree_root
 from lean_spec.types import Bytes32, Container, Uint64
 
+from ...xmss.aggregation import AggregatedSignatureProof
 from ...xmss.containers import Signature
 from ..checkpoint import Checkpoint
 from .aggregation_bits import AggregationBits
@@ -107,3 +108,10 @@ def aggregate_by_data(
             )
             for data, validator_ids in data_to_validator_ids.items()
         ]
+
+class SignedAggregatedAttestation(Container):
+    data: AttestationData
+    """Combined attestation data similar to the beacon chain format."""
+
+    proof: AggregatedSignatureProof
+    """Aggregated signature proof covering all participating validators."""

src/lean_spec/subspecs/containers/config.py

@@ -14,3 +14,6 @@ class Config(Container):
 
     genesis_time: Uint64
     """The timestamp of the genesis block."""
+
+    attestation_subnet_count: Uint64
+    """The number of attestation subnets in the network."""

src/lean_spec/subspecs/containers/state/state.py

@@ -30,6 +30,7 @@
     JustifiedSlots,
     Validators,
 )
+from ...chain.config import ATTESTATION_COMMITTEE_COUNT
 
 
 class State(Container):
@@ -90,6 +91,7 @@ def generate_genesis(cls, genesis_time: Uint64, validators: Validators) -> "Stat
         # Configure the genesis state.
         genesis_config = Config(
             genesis_time=genesis_time,
+            attestation_subnet_count=ATTESTATION_COMMITTEE_COUNT,
         )
 
         # Build the genesis block header for the state.
@@ -715,15 +717,13 @@ def build_block(
             # Add new attestations and continue iteration
             attestations.extend(new_attestations)
 
-        # Compute the aggregated signatures for the attestations.
-        # If the attestations cannot be aggregated, split it in a greedy way.
-        aggregated_attestations, aggregated_signatures = self.compute_aggregated_signatures(
+        # Select aggregated attestations and proofs for the final block
+        aggregated_attestations, aggregated_signatures = self.select_aggregated_proofs(
             attestations,
-            gossip_signatures,
             aggregated_payloads,
         )
 
-        # Update the block with the aggregated attestations
+        # Update the block with the aggregated attestations and proofs
         final_block = candidate_block.model_copy(
             update={
                 "body": BlockBody(
@@ -738,42 +738,30 @@ def build_block(
 
         return final_block, post_state, aggregated_attestations, aggregated_signatures
 
-    def compute_aggregated_signatures(
+    def aggregate_gossip_signatures(
         self,
         attestations: list[Attestation],
         gossip_signatures: dict[SignatureKey, "Signature"] | None = None,
-        aggregated_payloads: dict[SignatureKey, list[AggregatedSignatureProof]] | None = None,
-    ) -> tuple[list[AggregatedAttestation], list[AggregatedSignatureProof]]:
+    ) -> list[tuple[AggregatedAttestation, AggregatedSignatureProof]]:
         """
-        Compute aggregated signatures for a set of attestations.
-
-        This method implements a two-phase signature collection strategy:
+        Collect aggregated signatures from gossip network and aggregate them.
 
-        1. **Gossip Phase**: For each attestation group, first attempt to collect
-           individual XMSS signatures from the gossip network. These are fresh
-           signatures that validators broadcast when they attest.
-
-        2. **Fallback Phase**: For any validators not covered by gossip, fall back
-           to previously-seen aggregated proofs from blocks. This uses a greedy
-           set-cover approach to minimize the number of proofs needed.
-
-        The result is a list of (attestation, proof) pairs ready for block inclusion.
+        For each attestation group, attempt to collect individual XMSS signatures
+        from the gossip network. These are fresh signatures that validators
+        broadcast when they attest.
 
         Parameters
         ----------
         attestations : list[Attestation]
             Individual attestations to aggregate and sign.
         gossip_signatures : dict[SignatureKey, Signature] | None
             Per-validator XMSS signatures learned from the gossip network.
-        aggregated_payloads : dict[SignatureKey, list[AggregatedSignatureProof]] | None
-            Aggregated proofs learned from previously-seen blocks.
 
         Returns:
         -------
-        tuple[list[AggregatedAttestation], list[AggregatedSignatureProof]]
-            Paired attestations and their corresponding proofs.
+        list[tuple[AggregatedAttestation, AggregatedSignatureProof]]
+            - List of (attestation, proof) pairs from gossip collection.
         """
-        # Accumulator for (attestation, proof) pairs.
         results: list[tuple[AggregatedAttestation, AggregatedSignatureProof]] = []
 
         # Group individual attestations by data
@@ -790,8 +778,6 @@ def compute_aggregated_signatures(
             # Get the list of validators who attested to this data.
             validator_ids = aggregated.aggregation_bits.to_validator_indices()
 
-            # Phase 1: Gossip Collection
-            #
             # When a validator creates an attestation, it broadcasts the
             # individual XMSS signature over the gossip network. If we have
             # received these signatures, we can aggregate them ourselves.
@@ -803,16 +789,10 @@ def compute_aggregated_signatures(
             gossip_keys: list[PublicKey] = []
             gossip_ids: list[Uint64] = []
 
-            # Track validators we couldn't find signatures for.
-            #
-            # These will need to be covered by Phase 2 (existing proofs).
-            remaining: set[Uint64] = set()
-
             # Attempt to collect each validator's signature from gossip.
             #
             # Signatures are keyed by (validator ID, data root).
             # - If a signature exists, we add it to our collection.
-            # - Otherwise, we mark that validator as "remaining" for the fallback phase.
             if gossip_signatures:
                 for vid in validator_ids:
                     key = SignatureKey(vid, data_root)
@@ -821,12 +801,6 @@ def compute_aggregated_signatures(
                         gossip_sigs.append(sig)
                         gossip_keys.append(self.validators[vid].get_pubkey())
                         gossip_ids.append(vid)
-                    else:
-                        # No signature available: mark for fallback coverage.
-                        remaining.add(vid)
-            else:
-                # No gossip data at all: all validators need fallback coverage.
-                remaining = set(validator_ids)
 
             # If we collected any gossip signatures, aggregate them into a proof.
             #
@@ -841,14 +815,53 @@ def compute_aggregated_signatures(
                     message=data_root,
                     epoch=data.slot,
                 )
-                results.append(
-                    (
-                        AggregatedAttestation(aggregation_bits=participants, data=data),
-                        proof,
-                    )
-                )
+                attestation = AggregatedAttestation(aggregation_bits=participants, data=data)
+                results.append((attestation, proof))
+
+        return results
+
+    def select_aggregated_proofs(
+        self,
+        attestations: list[Attestation],
+        aggregated_payloads: dict[SignatureKey, list[AggregatedSignatureProof]] | None = None,
+    ) -> tuple[list[AggregatedAttestation], list[AggregatedSignatureProof]]:
+        """
+        Select aggregated proofs for a set of attestations.
+
+        This method selects aggregated proofs from aggregated_payloads,
+        prioritizing proofs from the most recent blocks.
 
-            # Phase 2: Fallback to existing proofs
+        Strategy:
+        1. For each attestation group, aggregate as many signatures as possible
+           from the most recent block's proofs.
+        2. If remaining validators exist after step 1, include proofs from
+           previous blocks that cover them.
+
+        Parameters:
+        ----------
+        attestations : list[Attestation]
+            Individual attestations to aggregate and sign.
+        aggregated_payloads : dict[SignatureKey, list[AggregatedSignatureProof]] | None
+            Aggregated proofs learned from previously-seen blocks.
+            The list for each key should be ordered with most recent proofs first.
+
+        Returns:
+        -------
+        tuple[list[AggregatedAttestation], list[AggregatedSignatureProof]]
+            Paired attestations and their corresponding proofs.
+        """
+        results: list[tuple[AggregatedAttestation, AggregatedSignatureProof]] = []
+
+        # Group individual attestations by data
+        for aggregated in AggregatedAttestation.aggregate_by_data(attestations):
+            data = aggregated.data
+            data_root = data.data_root_bytes()
+            validator_ids = aggregated.aggregation_bits.to_validator_indices() # validators contributed to this attestation
+
+            # Validators that are missing in the current aggregation are put into remaining.
+            remaining: set[Uint64] = set(validator_ids)
+
+            # Fallback to existing proofs
             #
             # Some validators may not have broadcast their signatures over gossip,
             # but we might have seen proofs for them in previously-received blocks.
@@ -924,14 +937,10 @@ def compute_aggregated_signatures(
                 remaining -= covered
 
         # Final Assembly
-        #
-        # - We built a list of (attestation, proof) tuples.
-        # - Now we unzip them into two parallel lists for the return value.
-
-        # Handle the empty case explicitly.
         if not results:
             return [], []
 
         # Unzip the results into parallel lists.
         aggregated_attestations, aggregated_proofs = zip(*results, strict=True)
         return list(aggregated_attestations), list(aggregated_proofs)
+