Merge remote-tracking branch 'origin/unstable' into deneb-free-blobs

book/src/api-bn.md

@@ -126,6 +126,22 @@ curl -X GET "http://localhost:5052/eth/v1/beacon/states/head/validators/1" -H  "
 ```
 You can replace `1` in the above command with the validator index that you would like to query. Other API query can be done similarly by changing the link according to the Beacon API.
 
+### Events API
+The [events API](https://ethereum.github.io/beacon-APIs/#/Events/eventstream) provides information such as the payload attributes that are of interest to block builders and relays. To query the payload attributes, it is necessary to run Lighthouse beacon node with the flag `--always-prepare-payload`. It is also recommended to add the flag `--prepare-payload-lookahead 8000` which configures the payload attributes to be sent at 4s into each slot (or 8s from the start of the next slot). An example of the command is:
+
+```bash
+curl -X 'GET' \
+'http://localhost:5052/eth/v1/events?topics=payload_attributes' \
+-H 'accept: text/event-stream' 
+```
+
+An example of response is:
+
+```json
+data:{"version":"capella","data":{"proposal_slot":"11047","proposer_index":"336057","parent_block_root":"0x26f8999d270dd4677c2a1c815361707157a531f6c599f78fa942c98b545e1799","parent_block_number":"9259","parent_block_hash":"0x7fb788cd7afa814e578afa00a3edd250cdd4c8e35c22badd327d981b5bda33d2","payload_attributes":{"timestamp":"1696034964","prev_randao":"0xeee34d7a3f6b99ade6c6a881046c9c0e96baab2ed9469102d46eb8d6e4fde14c","suggested_fee_recipient":"0x0000000000000000000000000000000000000001","withdrawals":[{"index":"40705","validator_index":"360712","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1202941"},{"index":"40706","validator_index":"360713","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1201138"},{"index":"40707","validator_index":"360714","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1215255"},{"index":"40708","validator_index":"360715","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1161977"},{"index":"40709","validator_index":"360716","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1257278"},{"index":"40710","validator_index":"360717","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1247740"},{"index":"40711","validator_index":"360718","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1204337"},{"index":"40712","validator_index":"360719","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1183575"},{"index":"40713","validator_index":"360720","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1157785"},{"index":"40714","validator_index":"360721","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1143371"},{"index":"40715","validator_index":"360722","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1234787"},{"index":"40716","validator_index":"360723","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1286673"},{"index":"40717","validator_index":"360724","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1419241"},{"index":"40718","validator_index":"360725","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1231015"},{"index":"40719","validator_index":"360726","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1304321"},{"index":"40720","validator_index":"360727","address":"0x73b2e0e54510239e22cc936f0b4a6de1acf0abde","amount":"1236543"}]}}}
+```
+
+
 ## Serving the HTTP API over TLS
 > **Warning**: This feature is currently experimental.
 

book/src/api-lighthouse.md

@@ -547,21 +547,6 @@ reconstruction has yet to be completed. For more information
 on the specific meanings of these fields see the docs on [Checkpoint
 Sync](./checkpoint-sync.md#reconstructing-states).
 
-### `/lighthouse/database/reconstruct`
-
-Instruct Lighthouse to begin reconstructing historic states, see
-[Reconstructing States](./checkpoint-sync.md#reconstructing-states). This is an alternative
-to the `--reconstruct-historic-states` flag.
-
-```
-curl -X POST "http://localhost:5052/lighthouse/database/reconstruct" | jq
-```
-
-```json
-"success"
-```
-
-The endpoint will return immediately. See the beacon node logs for an indication of progress.
 
 ### `/lighthouse/merge_readiness`
 Returns the current difficulty and terminal total difficulty of the network. Before [The Merge](https://ethereum.org/en/roadmap/merge/) on 15<sup>th</sup> September 2022, you will see that the current difficulty is less than the terminal total difficulty, An example is shown below:

book/src/builders.md

@@ -258,6 +258,9 @@ used in place of one from the builder:
 INFO Reconstructing a full block using a local payload
 ```
 
+## Information for block builders and relays
+Block builders and relays can query beacon node events from the [Events API](https://ethereum.github.io/beacon-APIs/#/Events/eventstream). An example of querying the payload attributes in the Events API is outlined in [Beacon node API - Events API](./api-bn.md#events-api)
+
 [mev-rs]: https://github.com/ralexstokes/mev-rs
 [mev-boost]: https://github.com/flashbots/mev-boost
 [gas-limit-api]: https://ethereum.github.io/keymanager-APIs/#/Gas%20Limit

book/src/validator-inclusion.md

@@ -8,7 +8,7 @@ These endpoints are not stable or included in the Ethereum consensus standard AP
 they are subject to change or removal without a change in major release
 version.
 
-In order to apply these APIs, you need to have historical states information in the database of your node. This means adding the flag `--reconstruct-historic-states` in the beacon node or using the [/lighthouse/database/reconstruct API](./api-lighthouse.md#lighthousedatabasereconstruct). Once the state reconstruction process is completed, you can apply these APIs to any epoch.
+In order to apply these APIs, you need to have historical states information in the database of your node. This means adding the flag `--reconstruct-historic-states` in the beacon node. Once the state reconstruction process is completed, you can apply these APIs to any epoch.
 
 ## Endpoints
 

validator_client/src/duties_service.rs

@@ -21,11 +21,12 @@ use eth2::types::{
 };
 use futures::{stream, StreamExt};
 use parking_lot::RwLock;
-use safe_arith::ArithError;
+use safe_arith::{ArithError, SafeArith};
 use slog::{debug, error, info, warn, Logger};
 use slot_clock::SlotClock;
 use std::cmp::min;
 use std::collections::{hash_map, BTreeMap, HashMap, HashSet};
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 use sync::poll_sync_committee_duties;
@@ -33,14 +34,6 @@ use sync::SyncDutiesMap;
 use tokio::{sync::mpsc::Sender, time::sleep};
 use types::{ChainSpec, Epoch, EthSpec, Hash256, PublicKeyBytes, SelectionProof, Slot};
 
-/// Since the BN does not like it when we subscribe to slots that are close to the current time, we
-/// will only subscribe to slots which are further than `SUBSCRIPTION_BUFFER_SLOTS` away.
-///
-/// This number is based upon `MIN_PEER_DISCOVERY_SLOT_LOOK_AHEAD` value in the
-/// `beacon_node::network::attestation_service` crate. It is not imported directly to avoid
-/// bringing in the entire crate.
-const SUBSCRIPTION_BUFFER_SLOTS: u64 = 2;
-
 /// Only retain `HISTORICAL_DUTIES_EPOCHS` duties prior to the current epoch.
 const HISTORICAL_DUTIES_EPOCHS: u64 = 2;
 
@@ -62,6 +55,36 @@ const VALIDATOR_METRICS_MIN_COUNT: usize = 64;
 /// reduces the amount of data that needs to be transferred.
 const INITIAL_DUTIES_QUERY_SIZE: usize = 1;
 
+/// Offsets from the attestation duty slot at which a subscription should be sent.
+const ATTESTATION_SUBSCRIPTION_OFFSETS: [u64; 8] = [3, 4, 5, 6, 7, 8, 16, 32];
+
+/// Check that `ATTESTATION_SUBSCRIPTION_OFFSETS` is sorted ascendingly.
+const _: () = assert!({
+    let mut i = 0;
+    loop {
+        let prev = if i > 0 {
+            ATTESTATION_SUBSCRIPTION_OFFSETS[i - 1]
+        } else {
+            0
+        };
+        let curr = ATTESTATION_SUBSCRIPTION_OFFSETS[i];
+        if curr < prev {
+            break false;
+        }
+        i += 1;
+        if i == ATTESTATION_SUBSCRIPTION_OFFSETS.len() {
+            break true;
+        }
+    }
+});
+/// Since the BN does not like it when we subscribe to slots that are close to the current time, we
+/// will only subscribe to slots which are further than 2 slots away.
+///
+/// This number is based upon `MIN_PEER_DISCOVERY_SLOT_LOOK_AHEAD` value in the
+/// `beacon_node::network::attestation_service` crate. It is not imported directly to avoid
+/// bringing in the entire crate.
+const _: () = assert!(ATTESTATION_SUBSCRIPTION_OFFSETS[0] > 2);
+
 #[derive(Debug)]
 pub enum Error {
     UnableToReadSlotClock,
@@ -84,6 +107,16 @@ pub struct DutyAndProof {
     pub duty: AttesterData,
     /// This value is only set to `Some` if the proof indicates that the validator is an aggregator.
     pub selection_proof: Option<SelectionProof>,
+    /// Track which slots we should send subscriptions at for this duty.
+    ///
+    /// This value is updated after each subscription is successfully sent.
+    pub subscription_slots: Arc<SubscriptionSlots>,
+}
+
+/// Tracker containing the slots at which an attestation subscription should be sent.
+pub struct SubscriptionSlots {
+    /// Pairs of `(slot, already_sent)` in slot-descending order.
+    slots: Vec<(Slot, AtomicBool)>,
 }
 
 impl DutyAndProof {
@@ -111,17 +144,55 @@ impl DutyAndProof {
                 }
             })?;
 
+        let subscription_slots = SubscriptionSlots::new(duty.slot);
+
         Ok(Self {
             duty,
             selection_proof,
+            subscription_slots,
         })
     }
 
     /// Create a new `DutyAndProof` with the selection proof waiting to be filled in.
     pub fn new_without_selection_proof(duty: AttesterData) -> Self {
+        let subscription_slots = SubscriptionSlots::new(duty.slot);
         Self {
             duty,
             selection_proof: None,
+            subscription_slots,
+        }
+    }
+}
+
+impl SubscriptionSlots {
+    fn new(duty_slot: Slot) -> Arc<Self> {
+        let slots = ATTESTATION_SUBSCRIPTION_OFFSETS
+            .into_iter()
+            .filter_map(|offset| duty_slot.safe_sub(offset).ok())
+            .map(|scheduled_slot| (scheduled_slot, AtomicBool::new(false)))
+            .collect();
+        Arc::new(Self { slots })
+    }
+
+    /// Return `true` if we should send a subscription at `slot`.
+    fn should_send_subscription_at(&self, slot: Slot) -> bool {
+        // Iterate slots from smallest to largest looking for one that hasn't been completed yet.
+        self.slots
+            .iter()
+            .rev()
+            .any(|(scheduled_slot, already_sent)| {
+                slot >= *scheduled_slot && !already_sent.load(Ordering::Relaxed)
+            })
+    }
+
+    /// Update our record of subscribed slots to account for successful subscription at `slot`.
+    fn record_successful_subscription_at(&self, slot: Slot) {
+        for (scheduled_slot, already_sent) in self.slots.iter().rev() {
+            if slot >= *scheduled_slot {
+                already_sent.store(true, Ordering::Relaxed);
+            } else {
+                break;
+            }
         }
     }
 }
@@ -574,8 +645,24 @@ async fn poll_beacon_attesters<T: SlotClock + 'static, E: EthSpec>(
     let subscriptions_timer =
         metrics::start_timer_vec(&metrics::DUTIES_SERVICE_TIMES, &[metrics::SUBSCRIPTIONS]);
 
-    // This vector is likely to be a little oversized, but it won't reallocate.
-    let mut subscriptions = Vec::with_capacity(local_pubkeys.len() * 2);
+    // This vector is intentionally oversized by 10% so that it won't reallocate.
+    // Each validator has 2 attestation duties occuring in the current and next epoch, for which
+    // they must send `ATTESTATION_SUBSCRIPTION_OFFSETS.len()` subscriptions. These subscription
+    // slots are approximately evenly distributed over the two epochs, usually with a slight lag
+    // that balances out (some subscriptions for the current epoch were sent in the previous, and
+    // some subscriptions for the next next epoch will be sent in the next epoch but aren't included
+    // in our calculation). We cancel the factor of 2 from the formula for simplicity.
+    let overallocation_numerator = 110;
+    let overallocation_denominator = 100;
+    let num_expected_subscriptions = overallocation_numerator
+        * std::cmp::max(
+            1,
+            local_pubkeys.len() * ATTESTATION_SUBSCRIPTION_OFFSETS.len()
+                / E::slots_per_epoch() as usize,
+        )
+        / overallocation_denominator;
+    let mut subscriptions = Vec::with_capacity(num_expected_subscriptions);
+    let mut subscription_slots_to_confirm = Vec::with_capacity(num_expected_subscriptions);
 
     // For this epoch and the next epoch, produce any beacon committee subscriptions.
     //
@@ -588,10 +675,10 @@ async fn poll_beacon_attesters<T: SlotClock + 'static, E: EthSpec>(
             .read()
             .iter()
             .filter_map(|(_, map)| map.get(epoch))
-            // The BN logs a warning if we try and subscribe to current or near-by slots. Give it a
-            // buffer.
             .filter(|(_, duty_and_proof)| {
-                current_slot + SUBSCRIPTION_BUFFER_SLOTS < duty_and_proof.duty.slot
+                duty_and_proof
+                    .subscription_slots
+                    .should_send_subscription_at(current_slot)
             })
             .for_each(|(_, duty_and_proof)| {
                 let duty = &duty_and_proof.duty;
@@ -603,7 +690,8 @@ async fn poll_beacon_attesters<T: SlotClock + 'static, E: EthSpec>(
                     committees_at_slot: duty.committees_at_slot,
                     slot: duty.slot,
                     is_aggregator,
-                })
+                });
+                subscription_slots_to_confirm.push(duty_and_proof.subscription_slots.clone());
             });
     }
 
@@ -632,6 +720,16 @@ async fn poll_beacon_attesters<T: SlotClock + 'static, E: EthSpec>(
                 "Failed to subscribe validators";
                 "error" => %e
             )
+        } else {
+            // Record that subscriptions were successfully sent.
+            debug!(
+                log,
+                "Broadcast attestation subscriptions";
+                "count" => subscriptions.len(),
+            );
+            for subscription_slots in subscription_slots_to_confirm {
+                subscription_slots.record_successful_subscription_at(current_slot);
+            }
         }
     }
 
@@ -1200,3 +1298,67 @@ async fn notify_block_production_service<T: SlotClock + 'static, E: EthSpec>(
         };
     }
 }
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn subscription_slots_exact() {
+        for duty_slot in [
+            Slot::new(32),
+            Slot::new(47),
+            Slot::new(99),
+            Slot::new(1002003),
+        ] {
+            let subscription_slots = SubscriptionSlots::new(duty_slot);
+
+            // Run twice to check idempotence (subscription slots shouldn't be marked as done until
+            // we mark them manually).
+            for _ in 0..2 {
+                for offset in ATTESTATION_SUBSCRIPTION_OFFSETS {
+                    assert!(subscription_slots.should_send_subscription_at(duty_slot - offset));
+                }
+            }
+
+            // Mark each slot as complete and check that all prior slots are still marked
+            // incomplete.
+            for (i, offset) in ATTESTATION_SUBSCRIPTION_OFFSETS
+                .into_iter()
+                .rev()
+                .enumerate()
+            {
+                subscription_slots.record_successful_subscription_at(duty_slot - offset);
+                for lower_offset in ATTESTATION_SUBSCRIPTION_OFFSETS
+                    .into_iter()
+                    .rev()
+                    .skip(i + 1)
+                {
+                    assert!(lower_offset < offset);
+                    assert!(
+                        subscription_slots.should_send_subscription_at(duty_slot - lower_offset)
+                    );
+                }
+            }
+        }
+    }
+    #[test]
+    fn subscription_slots_mark_multiple() {
+        for (i, offset) in ATTESTATION_SUBSCRIPTION_OFFSETS.into_iter().enumerate() {
+            let duty_slot = Slot::new(64);
+            let subscription_slots = SubscriptionSlots::new(duty_slot);
+
+            subscription_slots.record_successful_subscription_at(duty_slot - offset);
+
+            // All past offsets (earlier slots) should be marked as complete.
+            for (j, other_offset) in ATTESTATION_SUBSCRIPTION_OFFSETS.into_iter().enumerate() {
+                let past = j >= i;
+                assert_eq!(other_offset >= offset, past);
+                assert_eq!(
+                    subscription_slots.should_send_subscription_at(duty_slot - other_offset),
+                    !past
+                );
+            }
+        }
+    }
+}