feat(beacon): sync committee support in the node and the validator client

CLAUDE.md

@@ -334,7 +334,13 @@ actual_slot = finalized_slot + 1 + relative_index
   - Beacon wire: `validate_messages()` is on, so every beacon message waits for a verdict (~4.2s before gossipsub's cache evicts it). Rules in `state_transition::beacon::gossip` (cheap half inline, stateful half on a bounded `spawn_blocking` task); plumbing in `p2p/src/beacon/verdict.rs`. Lean gossip still auto-forwards
   - Data columns: every check runs in p2p. A column gossip did not accept (`Queue`/`Overloaded`), every fetched column, and parked columns replayed after their parent imports go through `column::chain_checks` in `p2p/src/beacon/column_checks.rs`. The chain actor stores what it gets unchecked; only debug builds re-run `chain_checks` there
   - Beacon subscribes seven global topics plus two node-id-derived subnet families: custody
-    columns and backbone attestation subnets. Gloas digests add two more topics,
+    columns and backbone attestation subnets. One of the seven is
+    `sync_committee_contribution_and_proof`; the four `sync_committee_{0..3}` subnets are
+    joined on demand only (a validator client's `sync_committee_subscriptions`, held
+    until its `until_epoch`, under every held digest) and advertised in MetaData
+    `syncnets`, never the ENR. Messages and contributions validate in p2p on their own
+    permit pool (`gossip::sync_committee`, `p2p/src/beacon/sync_committee.rs`), and
+    accepted ones go into the shared `SyncCommitteePool`. Gloas digests add two more topics,
     `execution_payload` and `payload_attestation_message` (`BeaconTopics::for_fork`; earlier
     digests never carry them). Gloas has its own rules for `beacon_block`,
     `data_column_sidecar` (fork enum `DataColumnSidecar`, fork from the topic's digest),
@@ -803,7 +809,15 @@ transitions are in `ethlambda-types`, per the section above. Nothing above
   `Eth-Blob-Data-Included: true`, both inside the proposal's attester-offset
   budget; an envelope failure is logged and counted, not a failed proposal),
   attest at `ATTESTATION_DUE_BPS_GLOAS`, aggregate at `AGGREGATE_DUE_BPS_GLOAS`,
-  then the PTC vote at `PAYLOAD_ATTESTATION_DUE_BPS`. `SlotClock` picks the
+  then the PTC vote at `PAYLOAD_ATTESTATION_DUE_BPS`. Sync committee duties
+  (`SyncCommitteeService`, `crates/validator/src/sync_committee.rs`) run at every fork:
+  one message per validator over the head root at `SYNC_MESSAGE_DUE_BPS(_GLOAS)`
+  (refused while the head is optimistic), then contributions for each selected
+  (validator, subnet) pair at `CONTRIBUTION_DUE_BPS(_GLOAS)`, with subscriptions
+  re-sent every epoch for the current and next period. The node pools the messages
+  and a block at slot N packs only `(N-1, parent_root)`, replaced by the empty
+  aggregate when `verified_sync_aggregate` fails (see `docs/spec_deviations.md`).
+  `SlotClock` picks the
   offsets by each slot's own fork, and `/eth/v1/config/spec` supplies them. It
   still keeps no slashing-protection record; PTC votes have an in-memory
   `(validator, slot)` dedup only, and envelopes are unguarded.

bin/ethlambda/src/beacon.rs

@@ -258,14 +258,12 @@ pub fn wire_params(
         "Backboning attestation subnets"
     );
 
-    // Say plainly what is still advertised without being backed by behavior,
-    // so a running node never implies more than it does. Storing and serving
-    // the custodied columns logged above is no longer in that gap, and neither
-    // is the attestation subnet backbone; sync committee subnet subscription,
-    // and publishing, still are.
-    warn!(
-        "Advertising cgc={custody_group_count} while subscribing to no sync committee \
-         subnet, and publishing nothing"
+    // Say plainly what the node does about sync committees, so a running node
+    // never implies more than it does: its subnets are joined only when a
+    // validator client asks, and `syncnets` advertises exactly those.
+    info!(
+        "Advertising cgc={custody_group_count}; sync committee subnets are joined only on a \
+         validator client's request"
     );
 
     Ok(BeaconWireParams {

bin/ethlambda/src/main.rs

@@ -721,13 +721,18 @@ async fn run_node(options: Options) -> eyre::Result<()> {
     // by block production and `GET .../pool/payload_attestations`.
     let payload_attestation_pool =
         ethlambda_state_transition::beacon::payload_attestation_pool::SharedPayloadAttestationPool::default();
+    // Filled by gossip and the Beacon API's sync committee endpoints, read by
+    // block production and the contribution endpoint.
+    let sync_committee_pool =
+        ethlambda_state_transition::beacon::sync_committee_pool::SharedSyncCommitteePool::default();
     let p2p = P2P::spawn(
         built,
         setup.store.clone(),
         setup.node_names,
         discovery,
         attestation_pool.clone(),
         payload_attestation_pool.clone(),
+        sync_committee_pool.clone(),
     )
     .await
     .wrap_err("failed to start discv5 discovery")?;
@@ -769,6 +774,7 @@ async fn run_node(options: Options) -> eyre::Result<()> {
                     p2p: rpc_p2p,
                     attestation_pool: attestation_pool.clone(),
                     payload_attestation_pool: payload_attestation_pool.clone(),
+                    sync_committee_pool: sync_committee_pool.clone(),
                     custody_columns: rpc_custody_columns,
                     engine: rpc_engine,
                 },

crates/blockchain/state_transition/src/beacon/block_production.rs

@@ -38,12 +38,13 @@ use super::bls;
 use super::config::Config;
 use super::error::{Error, Result, verify};
 use super::helpers::accessors::{
-    CommitteeCache, get_beacon_proposer_index, get_block_root, get_current_epoch,
-    get_previous_epoch, get_randao_mix,
+    CommitteeCache, get_beacon_proposer_index, get_block_root, get_block_root_at_slot,
+    get_current_epoch, get_domain, get_previous_epoch, get_randao_mix,
 };
 use super::helpers::electra::{
     get_attesting_indices, get_indexed_attestation, is_valid_indexed_attestation,
 };
+use super::helpers::misc::compute_signing_root;
 use super::lean_boundary::lean_state_unreachable;
 use super::stf::{self, ExecutionEngine};
 
@@ -108,6 +109,52 @@ pub fn empty_sync_aggregate() -> SyncAggregate {
     }
 }
 
+/// `candidate` when it passes exactly the check `process_sync_aggregate` will
+/// hold it to, else [`empty_sync_aggregate`].
+///
+/// `state` is the block's pre-state advanced to the block's slot. The
+/// participants are `current_sync_committee`'s members by bits, signing the
+/// block root at `state.slot - 1` under `DOMAIN_SYNC_COMMITTEE` at that slot's
+/// epoch. A pooled aggregate can fail this when the proposer's parent is not
+/// the root the committee signed, or when the head moved across a period: it
+/// then costs the rewards, never the block.
+pub fn verified_sync_aggregate(state: &BeaconState, candidate: SyncAggregate) -> SyncAggregate {
+    let Ok((committee, _)) = state.sync_committees() else {
+        return empty_sync_aggregate();
+    };
+    let participants: Vec<_> = committee
+        .pubkeys
+        .iter()
+        .enumerate()
+        .filter(|(position, _)| {
+            candidate
+                .sync_committee_bits
+                .get(*position)
+                .unwrap_or(false)
+        })
+        .map(|(_, pubkey)| *pubkey)
+        .collect();
+    let previous_slot = state.slot().saturating_sub(1);
+    let Ok(block_root) = get_block_root_at_slot(state, previous_slot) else {
+        return empty_sync_aggregate();
+    };
+    let domain = get_domain(
+        state,
+        constants::DOMAIN_SYNC_COMMITTEE,
+        Some(compute_epoch_at_slot(previous_slot)),
+    );
+    let signing_root = compute_signing_root(block_root, domain);
+    if bls::eth_fast_aggregate_verify(
+        &participants,
+        signing_root,
+        &candidate.sync_committee_signature,
+    ) {
+        candidate
+    } else {
+        empty_sync_aggregate()
+    }
+}
+
 /// The inverse of `get_execution_requests_list`: the execution client's
 /// EIP-7685 request list back into the block body's `ExecutionRequests`.
 ///
@@ -296,15 +343,18 @@ pub struct BlockInputs {
     pub execution_payload: ExecutionPayload,
     pub blob_kzg_commitments: Vec<KzgCommitment>,
     pub execution_requests: ExecutionRequests,
+    /// The block's sync aggregate: [`empty_sync_aggregate`], or what
+    /// [`verified_sync_aggregate`] vouched for.
+    pub sync_aggregate: SyncAggregate,
 }
 
 /// The unsigned block for the slot `state` has been advanced to, with its
 /// state root computed.
 ///
 /// The body votes the state's own `eth1_data` and carries no deposits (the
 /// deposit contract's log has been replaced by EIP-6110's requests), no
-/// slashings, exits or credential changes (this node pools none), and an empty
-/// sync aggregate. The block is run through `process_block` on a copy of
+/// slashings, exits or credential changes (this node pools none), and the
+/// sync aggregate it is given. The block is run through `process_block` on a copy of
 /// `state` with an execution engine that accepts the payload, which is the
 /// node's own execution client's payload; that run is also what rejects a body
 /// the network would, before anything is signed.
@@ -321,7 +371,7 @@ pub fn assemble_block(
             .attestations
             .try_into()
             .map_err(|_| Error::SpecAssert("len(attestations) <= MAX_ATTESTATIONS_ELECTRA"))?,
-        sync_aggregate: empty_sync_aggregate(),
+        sync_aggregate: inputs.sync_aggregate,
         execution_payload: inputs.execution_payload,
         blob_kzg_commitments: inputs
             .blob_kzg_commitments
@@ -431,6 +481,7 @@ mod tests {
     fn an_assembled_block_passes_process_block_and_names_its_post_state() {
         let state = state_to_build_on();
         let inputs = BlockInputs {
+            sync_aggregate: empty_sync_aggregate(),
             randao_reveal: randao_reveal(&state),
             graffiti: Bytes32::repeat_byte(7),
             attestations: Vec::new(),
@@ -469,6 +520,7 @@ mod tests {
         let mut payload = payload_for(&state);
         payload.parent_hash = ExecutionBlockHash::repeat_byte(9);
         let inputs = BlockInputs {
+            sync_aggregate: empty_sync_aggregate(),
             randao_reveal: randao_reveal(&state),
             graffiti: Bytes32::ZERO,
             attestations: Vec::new(),
@@ -654,6 +706,84 @@ mod tests {
         assert!(parse_execution_requests(&[vec![0x7f, 0]]).is_err());
     }
 
+    /// A pool holding messages from `positions` of the committee, signed the way
+    /// `process_sync_aggregate` will check them (over the block root at
+    /// `state.slot - 1`, under the state's own domain), plus that root.
+    fn pooled_aggregate(
+        state: &BeaconState,
+        signed_root: Option<Root>,
+        positions: &[usize],
+    ) -> (SyncAggregate, Root) {
+        use crate::beacon::sync_committee_pool::SyncCommitteePool;
+        use ethlambda_types::beacon::containers::altair::{
+            SYNC_SUBCOMMITTEE_SIZE, SyncCommitteeMessage,
+        };
+
+        let previous_slot = state.slot() - 1;
+        let parent_root = get_block_root_at_slot(state, previous_slot).unwrap();
+        let root = signed_root.unwrap_or(parent_root);
+        let domain = get_domain(
+            state,
+            constants::DOMAIN_SYNC_COMMITTEE,
+            Some(compute_epoch_at_slot(previous_slot)),
+        );
+        let signing_root = compute_signing_root(root, domain);
+        let (committee, _) = state.sync_committees().unwrap();
+        let mut pool = SyncCommitteePool::default();
+        for &position in positions {
+            let pubkey = committee.pubkeys[position];
+            let index = (0..64)
+                .find(|&index| state.validator(index).unwrap().pubkey == pubkey)
+                .expect("the committee is drawn from the registry");
+            let message = SyncCommitteeMessage {
+                slot: previous_slot,
+                beacon_block_root: parent_root,
+                validator_index: index,
+                signature: sign_for(index as usize, signing_root),
+            };
+            let seats = [(
+                (position / SYNC_SUBCOMMITTEE_SIZE) as u64,
+                position % SYNC_SUBCOMMITTEE_SIZE,
+            )];
+            pool.insert_message(&message, &seats);
+        }
+        (
+            pool.sync_aggregate(previous_slot, parent_root)
+                .expect("something was pooled"),
+            parent_root,
+        )
+    }
+
+    #[test]
+    fn a_pooled_sync_aggregate_passes_assemble_block() {
+        let state = state_to_build_on();
+        let (candidate, _) = pooled_aggregate(&state, None, &[0, 1, 5]);
+        let verified = verified_sync_aggregate(&state, candidate.clone());
+        assert_eq!(verified, candidate);
+        assert!(verified.sync_committee_bits.count_ones() >= 3);
+        let inputs = BlockInputs {
+            randao_reveal: randao_reveal(&state),
+            graffiti: Bytes32::repeat_byte(7),
+            attestations: Vec::new(),
+            execution_payload: payload_for(&state),
+            blob_kzg_commitments: Vec::new(),
+            execution_requests: ExecutionRequests::default(),
+            sync_aggregate: verified.clone(),
+        };
+        let block = assemble_block(&state, inputs, &Config::mainnet()).unwrap();
+        assert_eq!(block.body.sync_aggregate, verified);
+    }
+
+    #[test]
+    fn a_sync_aggregate_over_the_wrong_root_is_replaced_by_the_empty_one() {
+        let state = state_to_build_on();
+        let (wrong, _) = pooled_aggregate(&state, Some(Root::repeat_byte(1)), &[0, 1]);
+        assert_eq!(
+            verified_sync_aggregate(&state, wrong),
+            empty_sync_aggregate()
+        );
+    }
+
     #[test]
     fn the_empty_sync_aggregate_signs_with_the_point_at_infinity() {
         let aggregate = empty_sync_aggregate();