Optimise payload attributes calculation and add SSE (sigp#4027)

Closes sigp#3896
Closes sigp#3998
Closes sigp#3700

- Optimise the calculation of withdrawals for payload attributes by avoiding state clones, avoiding unnecessary state advances and reading from the snapshot cache if possible.
- Use the execution layer's payload attributes cache to avoid re-calculating payload attributes. I actually implemented a new LRU cache just for withdrawals but it had the exact same key and most of the same data as the existing payload attributes cache, so I deleted it.
- Add a new SSE event that fires when payloadAttributes are calculated. This is useful for block builders, a la ethereum/beacon-APIs#244.
- Add a new CLI flag `--always-prepare-payload` which forces payload attributes to be sent with every fcU regardless of connected proposers. This is intended for use by builders/relays.

For maximum effect, the flags I've been using to run Lighthouse in "payload builder mode" are:

```
--always-prepare-payload \
--prepare-payload-lookahead 12000 \
--suggested-fee-recipient 0x0000000000000000000000000000000000000000
```

The fee recipient is required so Lighthouse has something to pack in the payload attributes (it can be ignored by the builder). The lookahead causes fcU to be sent at the start of every slot rather than at 8s. As usual, fcU will also be sent after each change of head block. I think this combination is sufficient for builders to build on all viable heads. Often there will be two fcU (and two payload attributes) sent for the same slot: one sent at the start of the slot with the head from `n - 1` as the parent, and one sent after the block arrives with `n` as the parent.

Example usage of the new event stream:

```bash
curl -N "http://localhost:5052/eth/v1/events?topics=payload_attributes"
```

- [x] Tests added by updating the proposer re-org tests. This has the benefit of testing the proposer re-org code paths with withdrawals too, confirming that the new changes don't interact poorly.
- [ ] Benchmarking with `blockdreamer` on devnet-7 showed promising results but I'm yet to do a comparison to `unstable`.

Co-authored-by: Michael Sproul <[email protected]>

Author: michaelsproul

beacon_node/beacon_chain/src/chain_config.rs

@@ -67,6 +67,10 @@ pub struct ChainConfig {
     pub prepare_payload_lookahead: Duration,
     /// Use EL-free optimistic sync for the finalized part of the chain.
     pub optimistic_finalized_sync: bool,
+    /// Whether to send payload attributes every slot, regardless of connected proposers.
+    ///
+    /// This is useful for block builders and testing.
+    pub always_prepare_payload: bool,
 }
 
 impl Default for ChainConfig {
@@ -92,6 +96,7 @@ impl Default for ChainConfig {
             checkpoint_sync_url_timeout: 60,
             prepare_payload_lookahead: Duration::from_secs(4),
             optimistic_finalized_sync: true,
+            always_prepare_payload: false,
         }
     }
 }

beacon_node/beacon_chain/src/events.rs

@@ -14,6 +14,7 @@ pub struct ServerSentEventHandler<T: EthSpec> {
     exit_tx: Sender<EventKind<T>>,
     chain_reorg_tx: Sender<EventKind<T>>,
     contribution_tx: Sender<EventKind<T>>,
+    payload_attributes_tx: Sender<EventKind<T>>,
     late_head: Sender<EventKind<T>>,
     block_reward_tx: Sender<EventKind<T>>,
     log: Logger,
@@ -32,6 +33,7 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
         let (exit_tx, _) = broadcast::channel(capacity);
         let (chain_reorg_tx, _) = broadcast::channel(capacity);
         let (contribution_tx, _) = broadcast::channel(capacity);
+        let (payload_attributes_tx, _) = broadcast::channel(capacity);
         let (late_head, _) = broadcast::channel(capacity);
         let (block_reward_tx, _) = broadcast::channel(capacity);
 
@@ -43,35 +45,63 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
             exit_tx,
             chain_reorg_tx,
             contribution_tx,
+            payload_attributes_tx,
             late_head,
             block_reward_tx,
             log,
         }
     }
 
     pub fn register(&self, kind: EventKind<T>) {
-        let result = match kind {
-            EventKind::Attestation(attestation) => self
+        let log_count = |name, count| {
+            trace!(
+                self.log,
+                "Registering server-sent event";
+                "kind" => name,
+                "receiver_count" => count
+            );
+        };
+        let result = match &kind {
+            EventKind::Attestation(_) => self
                 .attestation_tx
-                .send(EventKind::Attestation(attestation))
-                .map(|count| trace!(self.log, "Registering server-sent attestation event"; "receiver_count" => count)),
-            EventKind::Block(block) => self.block_tx.send(EventKind::Block(block))
-                .map(|count| trace!(self.log, "Registering server-sent block event"; "receiver_count" => count)),
-            EventKind::FinalizedCheckpoint(checkpoint) => self.finalized_tx
-                .send(EventKind::FinalizedCheckpoint(checkpoint))
-                .map(|count| trace!(self.log, "Registering server-sent finalized checkpoint event"; "receiver_count" => count)),
-            EventKind::Head(head) => self.head_tx.send(EventKind::Head(head))
-                .map(|count| trace!(self.log, "Registering server-sent head event"; "receiver_count" => count)),
-            EventKind::VoluntaryExit(exit) => self.exit_tx.send(EventKind::VoluntaryExit(exit))
-                .map(|count| trace!(self.log, "Registering server-sent voluntary exit event"; "receiver_count" => count)),
-            EventKind::ChainReorg(reorg) => self.chain_reorg_tx.send(EventKind::ChainReorg(reorg))
-                .map(|count| trace!(self.log, "Registering server-sent chain reorg event"; "receiver_count" => count)),
-            EventKind::ContributionAndProof(contribution_and_proof) => self.contribution_tx.send(EventKind::ContributionAndProof(contribution_and_proof))
-                .map(|count| trace!(self.log, "Registering server-sent contribution and proof event"; "receiver_count" => count)),
-            EventKind::LateHead(late_head) => self.late_head.send(EventKind::LateHead(late_head))
-                .map(|count| trace!(self.log, "Registering server-sent late head event"; "receiver_count" => count)),
-            EventKind::BlockReward(block_reward) => self.block_reward_tx.send(EventKind::BlockReward(block_reward))
-                .map(|count| trace!(self.log, "Registering server-sent contribution and proof event"; "receiver_count" => count)),
+                .send(kind)
+                .map(|count| log_count(count, "attestation")),
+            EventKind::Block(_) => self
+                .block_tx
+                .send(kind)
+                .map(|count| log_count(count, "block")),
+            EventKind::FinalizedCheckpoint(_) => self
+                .finalized_tx
+                .send(kind)
+                .map(|count| log_count(count, "finalized checkpoint")),
+            EventKind::Head(_) => self
+                .head_tx
+                .send(kind)
+                .map(|count| log_count(count, "head")),
+            EventKind::VoluntaryExit(_) => self
+                .exit_tx
+                .send(kind)
+                .map(|count| log_count(count, "exit")),
+            EventKind::ChainReorg(_) => self
+                .chain_reorg_tx
+                .send(kind)
+                .map(|count| log_count(count, "chain reorg")),
+            EventKind::ContributionAndProof(_) => self
+                .contribution_tx
+                .send(kind)
+                .map(|count| log_count(count, "contribution and proof")),
+            EventKind::PayloadAttributes(_) => self
+                .payload_attributes_tx
+                .send(kind)
+                .map(|count| log_count(count, "payload attributes")),
+            EventKind::LateHead(_) => self
+                .late_head
+                .send(kind)
+                .map(|count| log_count(count, "late head")),
+            EventKind::BlockReward(_) => self
+                .block_reward_tx
+                .send(kind)
+                .map(|count| log_count(count, "block reward")),
         };
         if let Err(SendError(event)) = result {
             trace!(self.log, "No receivers registered to listen for event"; "event" => ?event);
@@ -106,6 +136,10 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
         self.contribution_tx.subscribe()
     }
 
+    pub fn subscribe_payload_attributes(&self) -> Receiver<EventKind<T>> {
+        self.payload_attributes_tx.subscribe()
+    }
+
     pub fn subscribe_late_head(&self) -> Receiver<EventKind<T>> {
         self.late_head.subscribe()
     }
@@ -142,6 +176,10 @@ impl<T: EthSpec> ServerSentEventHandler<T> {
         self.contribution_tx.receiver_count() > 0
     }
 
+    pub fn has_payload_attributes_subscribers(&self) -> bool {
+        self.payload_attributes_tx.receiver_count() > 0
+    }
+
     pub fn has_late_head_subscribers(&self) -> bool {
         self.late_head.receiver_count() > 0
     }

beacon_node/beacon_chain/src/test_utils.rs

@@ -107,6 +107,14 @@ pub enum AttestationStrategy {
     SomeValidators(Vec<usize>),
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SyncCommitteeStrategy {
+    /// All sync committee validators sign.
+    AllValidators,
+    /// No validators sign.
+    NoValidators,
+}
+
 /// Indicates whether the `BeaconChainHarness` should use the `state.current_sync_committee` or
 /// `state.next_sync_committee` when creating sync messages or contributions.
 #[derive(Clone, Debug)]
@@ -1763,15 +1771,64 @@ where
         self.process_attestations(attestations);
     }
 
+    pub fn sync_committee_sign_block(
+        &self,
+        state: &BeaconState<E>,
+        block_hash: Hash256,
+        slot: Slot,
+        relative_sync_committee: RelativeSyncCommittee,
+    ) {
+        let sync_contributions =
+            self.make_sync_contributions(state, block_hash, slot, relative_sync_committee);
+        self.process_sync_contributions(sync_contributions).unwrap()
+    }
+
     pub async fn add_attested_block_at_slot(
         &self,
         slot: Slot,
         state: BeaconState<E>,
         state_root: Hash256,
         validators: &[usize],
+    ) -> Result<(SignedBeaconBlockHash, BeaconState<E>), BlockError<E>> {
+        self.add_attested_block_at_slot_with_sync(
+            slot,
+            state,
+            state_root,
+            validators,
+            SyncCommitteeStrategy::NoValidators,
+        )
+        .await
+    }
+
+    pub async fn add_attested_block_at_slot_with_sync(
+        &self,
+        slot: Slot,
+        state: BeaconState<E>,
+        state_root: Hash256,
+        validators: &[usize],
+        sync_committee_strategy: SyncCommitteeStrategy,
     ) -> Result<(SignedBeaconBlockHash, BeaconState<E>), BlockError<E>> {
         let (block_hash, block, state) = self.add_block_at_slot(slot, state).await?;
         self.attest_block(&state, state_root, block_hash, &block, validators);
+
+        if sync_committee_strategy == SyncCommitteeStrategy::AllValidators
+            && state.current_sync_committee().is_ok()
+        {
+            self.sync_committee_sign_block(
+                &state,
+                block_hash.into(),
+                slot,
+                if (slot + 1).epoch(E::slots_per_epoch())
+                    % self.spec.epochs_per_sync_committee_period
+                    == 0
+                {
+                    RelativeSyncCommittee::Next
+                } else {
+                    RelativeSyncCommittee::Current
+                },
+            );
+        }
+
         Ok((block_hash, state))
     }
 
@@ -1781,10 +1838,35 @@ where
         state_root: Hash256,
         slots: &[Slot],
         validators: &[usize],
+    ) -> AddBlocksResult<E> {
+        self.add_attested_blocks_at_slots_with_sync(
+            state,
+            state_root,
+            slots,
+            validators,
+            SyncCommitteeStrategy::NoValidators,
+        )
+        .await
+    }
+
+    pub async fn add_attested_blocks_at_slots_with_sync(
+        &self,
+        state: BeaconState<E>,
+        state_root: Hash256,
+        slots: &[Slot],
+        validators: &[usize],
+        sync_committee_strategy: SyncCommitteeStrategy,
     ) -> AddBlocksResult<E> {
         assert!(!slots.is_empty());
-        self.add_attested_blocks_at_slots_given_lbh(state, state_root, slots, validators, None)
-            .await
+        self.add_attested_blocks_at_slots_given_lbh(
+            state,
+            state_root,
+            slots,
+            validators,
+            None,
+            sync_committee_strategy,
+        )
+        .await
     }
 
     async fn add_attested_blocks_at_slots_given_lbh(
@@ -1794,6 +1876,7 @@ where
         slots: &[Slot],
         validators: &[usize],
         mut latest_block_hash: Option<SignedBeaconBlockHash>,
+        sync_committee_strategy: SyncCommitteeStrategy,
     ) -> AddBlocksResult<E> {
         assert!(
             slots.windows(2).all(|w| w[0] <= w[1]),
@@ -1803,7 +1886,13 @@ where
         let mut state_hash_from_slot: HashMap<Slot, BeaconStateHash> = HashMap::new();
         for slot in slots {
             let (block_hash, new_state) = self
-                .add_attested_block_at_slot(*slot, state, state_root, validators)
+                .add_attested_block_at_slot_with_sync(
+                    *slot,
+                    state,
+                    state_root,
+                    validators,
+                    sync_committee_strategy,
+                )
                 .await
                 .unwrap();
             state = new_state;
@@ -1885,6 +1974,7 @@ where
                         &epoch_slots,
                         &validators,
                         Some(head_block),
+                        SyncCommitteeStrategy::NoValidators, // for backwards compat
                     )
                     .await;
 
@@ -2002,6 +2092,22 @@ where
         num_blocks: usize,
         block_strategy: BlockStrategy,
         attestation_strategy: AttestationStrategy,
+    ) -> Hash256 {
+        self.extend_chain_with_sync(
+            num_blocks,
+            block_strategy,
+            attestation_strategy,
+            SyncCommitteeStrategy::NoValidators,
+        )
+        .await
+    }
+
+    pub async fn extend_chain_with_sync(
+        &self,
+        num_blocks: usize,
+        block_strategy: BlockStrategy,
+        attestation_strategy: AttestationStrategy,
+        sync_committee_strategy: SyncCommitteeStrategy,
     ) -> Hash256 {
         let (mut state, slots) = match block_strategy {
             BlockStrategy::OnCanonicalHead => {
@@ -2033,7 +2139,13 @@ where
         };
         let state_root = state.update_tree_hash_cache().unwrap();
         let (_, _, last_produced_block_hash, _) = self
-            .add_attested_blocks_at_slots(state, state_root, &slots, &validators)
+            .add_attested_blocks_at_slots_with_sync(
+                state,
+                state_root,
+                &slots,
+                &validators,
+                sync_committee_strategy,
+            )
             .await;
         last_produced_block_hash.into()
     }

beacon_node/http_api/src/lib.rs

@@ -3501,6 +3501,9 @@ pub fn serve<T: BeaconChainTypes>(
                                 api_types::EventTopic::ContributionAndProof => {
                                     event_handler.subscribe_contributions()
                                 }
+                                api_types::EventTopic::PayloadAttributes => {
+                                    event_handler.subscribe_payload_attributes()
+                                }
                                 api_types::EventTopic::LateHead => {
                                     event_handler.subscribe_late_head()
                                 }