[Merged by Bors] - Allow per validator fee recipient via flag or file in validator client (similar to graffiti / graffiti-file)

account_manager/src/validator/import.rs

@@ -273,9 +273,15 @@ pub fn cli_run(matches: &ArgMatches, validator_dir: PathBuf) -> Result<(), Strin
         eprintln!("Successfully imported keystore.");
         num_imported_keystores += 1;
 
-        let validator_def =
-            ValidatorDefinition::new_keystore_with_password(&dest_keystore, password_opt, None)
-                .map_err(|e| format!("Unable to create new validator definition: {:?}", e))?;
+        let graffiti = None;
+        let suggested_fee_recipient = None;
+        let validator_def = ValidatorDefinition::new_keystore_with_password(
+            &dest_keystore,
+            password_opt,
+            graffiti,
+            suggested_fee_recipient,
+        )
+        .map_err(|e| format!("Unable to create new validator definition: {:?}", e))?;
 
         defs.push(validator_def);
 

beacon_node/beacon_chain/src/execution_payload.rs

@@ -273,6 +273,7 @@ pub async fn prepare_execution_payload<T: BeaconChainTypes>(
     let timestamp = compute_timestamp_at_slot(state, spec).map_err(BeaconStateError::from)?;
     let random = *state.get_randao_mix(state.current_epoch())?;
     let finalized_root = state.finalized_checkpoint().root;
+    let proposer_index = state.get_beacon_proposer_index(state.slot(), spec)? as u64;
 
     // The finalized block hash is not included in the specification, however we provide this
     // parameter so that the execution layer can produce a payload id if one is not already known
@@ -300,6 +301,7 @@ pub async fn prepare_execution_payload<T: BeaconChainTypes>(
             timestamp,
             random,
             finalized_block_hash.unwrap_or_else(Hash256::zero),
+            proposer_index,
         )
         .await
         .map_err(BlockProductionError::GetPayloadFailed)?;

beacon_node/client/src/builder.rs

@@ -700,6 +700,11 @@ where
 
                 // Spawn a routine that tracks the status of the execution engines.
                 execution_layer.spawn_watchdog_routine(beacon_chain.slot_clock.clone());
+
+                // Spawn a routine that removes expired proposer preparations.
+                execution_layer.spawn_clean_proposer_preparation_routine::<TSlotClock, TEthSpec>(
+                    beacon_chain.slot_clock.clone(),
+                );
             }
         }
 

beacon_node/execution_layer/src/lib.rs

@@ -10,6 +10,7 @@ use lru::LruCache;
 use sensitive_url::SensitiveUrl;
 use slog::{crit, debug, error, info, Logger};
 use slot_clock::SlotClock;
+use std::collections::HashMap;
 use std::future::Future;
 use std::sync::Arc;
 use std::time::Duration;
@@ -18,7 +19,7 @@ use tokio::{
     sync::{Mutex, MutexGuard},
     time::{sleep, sleep_until, Instant},
 };
-use types::ChainSpec;
+use types::{ChainSpec, Epoch, ProposerPreparationData};
 
 pub use engine_api::{http::HttpJsonRpc, ExecutePayloadResponseStatus};
 
@@ -30,6 +31,16 @@ pub mod test_utils;
 /// in an LRU cache to avoid redundant lookups. This is the size of that cache.
 const EXECUTION_BLOCKS_LRU_CACHE_SIZE: usize = 128;
 
+/// A fee recipient address for use during block production. Only used as a very last resort if
+/// there is no address provided by the user.
+///
+/// ## Note
+///
+/// This is *not* the zero-address, since Geth has been known to return errors for a coinbase of
+/// 0x00..00.
+const DEFAULT_SUGGESTED_FEE_RECIPIENT: [u8; 20] =
+    [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1];
+
 #[derive(Debug)]
 pub enum Error {
     NoEngines,
@@ -46,9 +57,16 @@ impl From<ApiError> for Error {
     }
 }
 
+#[derive(Clone)]
+pub struct ProposerPreparationDataEntry {
+    update_epoch: Epoch,
+    preparation_data: ProposerPreparationData,
+}
+
 struct Inner {
     engines: Engines<HttpJsonRpc>,
     suggested_fee_recipient: Option<Address>,
+    proposer_preparation_data: Mutex<HashMap<u64, ProposerPreparationDataEntry>>,
     execution_blocks: Mutex<LruCache<Hash256, ExecutionBlock>>,
     executor: TaskExecutor,
     log: Logger,
@@ -96,6 +114,7 @@ impl ExecutionLayer {
                 log: log.clone(),
             },
             suggested_fee_recipient,
+            proposer_preparation_data: Mutex::new(HashMap::new()),
             execution_blocks: Mutex::new(LruCache::new(EXECUTION_BLOCKS_LRU_CACHE_SIZE)),
             executor,
             log,
@@ -116,17 +135,18 @@ impl ExecutionLayer {
         &self.inner.executor
     }
 
-    fn suggested_fee_recipient(&self) -> Result<Address, Error> {
-        self.inner
-            .suggested_fee_recipient
-            .ok_or(Error::FeeRecipientUnspecified)
-    }
-
     /// Note: this function returns a mutex guard, be careful to avoid deadlocks.
     async fn execution_blocks(&self) -> MutexGuard<'_, LruCache<Hash256, ExecutionBlock>> {
         self.inner.execution_blocks.lock().await
     }
 
+    /// Note: this function returns a mutex guard, be careful to avoid deadlocks.
+    async fn proposer_preparation_data(
+        &self,
+    ) -> MutexGuard<'_, HashMap<u64, ProposerPreparationDataEntry>> {
+        self.inner.proposer_preparation_data.lock().await
+    }
+
     fn log(&self) -> &Logger {
         &self.inner.log
     }
@@ -234,11 +254,124 @@ impl ExecutionLayer {
         self.engines().upcheck_not_synced(Logging::Disabled).await;
     }
 
+    /// Spawns a routine which cleans the cached proposer preparations periodically.
+    pub fn spawn_clean_proposer_preparation_routine<S: SlotClock + 'static, T: EthSpec>(
+        &self,
+        slot_clock: S,
+    ) {
+        let preparation_cleaner = |el: ExecutionLayer| async move {
+            // Start the loop to periodically clean proposer preparation cache.
+            loop {
+                if let Some(duration_to_next_epoch) =
+                    slot_clock.duration_to_next_epoch(T::slots_per_epoch())
+                {
+                    // Wait for next epoch
+                    sleep(duration_to_next_epoch).await;
+
+                    match slot_clock
+                        .now()
+                        .map(|slot| slot.epoch(T::slots_per_epoch()))
+                    {
+                        Some(current_epoch) => el
+                            .clean_proposer_preparation(current_epoch)
+                            .await
+                            .map_err(|e| {
+                                error!(
+                                    el.log(),
+                                    "Failed to clean proposer preparation cache";
+                                    "error" => format!("{:?}", e)
+                                )
+                            })
+                            .unwrap_or(()),
+                        None => error!(el.log(), "Failed to get current epoch from slot clock"),
+                    }
+                } else {
+                    error!(el.log(), "Failed to read slot clock");
+                    // If we can't read the slot clock, just wait another slot and retry.
+                    sleep(slot_clock.slot_duration()).await;
+                }
+            }
+        };
+
+        self.spawn(preparation_cleaner, "exec_preparation_cleanup");
+    }
+
     /// Returns `true` if there is at least one synced and reachable engine.
     pub async fn is_synced(&self) -> bool {
         self.engines().any_synced().await
     }
 
+    /// Updates the proposer preparation data provided by validators
+    pub fn update_proposer_preparation_blocking(
+        &self,
+        update_epoch: Epoch,
+        preparation_data: &[ProposerPreparationData],
+    ) -> Result<(), Error> {
+        self.block_on_generic(|_| async move {
+            self.update_proposer_preparation(update_epoch, preparation_data)
+                .await
+        })?
+    }
+
+    /// Updates the proposer preparation data provided by validators
+    async fn update_proposer_preparation(
+        &self,
+        update_epoch: Epoch,
+        preparation_data: &[ProposerPreparationData],
+    ) -> Result<(), Error> {
+        let mut proposer_preparation_data = self.proposer_preparation_data().await;
+        for preparation_entry in preparation_data {
+            proposer_preparation_data.insert(
+                preparation_entry.validator_index,
+                ProposerPreparationDataEntry {
+                    update_epoch,
+                    preparation_data: preparation_entry.clone(),
+                },
+            );
+        }
+
+        Ok(())
+    }
+
+    /// Removes expired entries from cached proposer preparations
+    async fn clean_proposer_preparation(&self, current_epoch: Epoch) -> Result<(), Error> {
+        let mut proposer_preparation_data = self.proposer_preparation_data().await;
+
+        // Keep all entries that have been updated in the last 2 epochs
+        let retain_epoch = current_epoch.saturating_sub(Epoch::new(2));
+        proposer_preparation_data.retain(|_validator_index, preparation_entry| {
+            preparation_entry.update_epoch >= retain_epoch
+        });
+
+        Ok(())
+    }
+
+    /// Returns the fee-recipient address that should be used to build a block
+    async fn get_suggested_fee_recipient(&self, proposer_index: u64) -> Address {
+        if let Some(preparation_data_entry) =
+            self.proposer_preparation_data().await.get(&proposer_index)
+        {
+            // The values provided via the API have first priority.
+            preparation_data_entry.preparation_data.fee_recipient
+        } else if let Some(address) = self.inner.suggested_fee_recipient {
+            // If there has been no fee recipient provided via the API, but the BN has been provided
+            // with a global default address, use that.
+            address
+        } else {
+            // If there is no user-provided fee recipient, use a junk value and complain loudly.
+            crit!(
+                self.log(),
+                "Fee recipient unknown";
+                "msg" => "the suggested_fee_recipient was unknown during block production. \
+                a junk address was used, rewards were lost! \
+                check the --suggested-fee-recipient flag and VC configuration.",
+                "proposer_index" => ?proposer_index
+            );
+
+            Address::from_slice(&DEFAULT_SUGGESTED_FEE_RECIPIENT)
+        }
+    }
+
     /// Maps to the `engine_getPayload` JSON-RPC call.
     ///
     /// However, it will attempt to call `self.prepare_payload` if it cannot find an existing
@@ -254,8 +387,10 @@ impl ExecutionLayer {
         timestamp: u64,
         random: Hash256,
         finalized_block_hash: Hash256,
+        proposer_index: u64,
     ) -> Result<ExecutionPayload<T>, Error> {
-        let suggested_fee_recipient = self.suggested_fee_recipient()?;
+        let suggested_fee_recipient = self.get_suggested_fee_recipient(proposer_index).await;
+
         debug!(
             self.log(),
             "Issuing engine_getPayload";

beacon_node/execution_layer/src/test_utils/mock_execution_layer.rs

@@ -127,9 +127,16 @@ impl<T: EthSpec> MockExecutionLayer<T> {
             .await
             .unwrap();
 
+        let validator_index = 0;
         let payload = self
             .el
-            .get_payload::<T>(parent_hash, timestamp, random, finalized_block_hash)
+            .get_payload::<T>(
+                parent_hash,
+                timestamp,
+                random,
+                finalized_block_hash,
+                validator_index,
+            )
             .await
             .unwrap();
         let block_hash = payload.block_hash;

beacon_node/http_api/src/lib.rs

@@ -45,9 +45,9 @@ use tokio::sync::mpsc::UnboundedSender;
 use tokio_stream::{wrappers::BroadcastStream, StreamExt};
 use types::{
     Attestation, AttesterSlashing, BeaconStateError, CommitteeCache, ConfigAndPreset, Epoch,
-    EthSpec, ForkName, ProposerSlashing, RelativeEpoch, SignedAggregateAndProof, SignedBeaconBlock,
-    SignedContributionAndProof, SignedVoluntaryExit, Slot, SyncCommitteeMessage,
-    SyncContributionData,
+    EthSpec, ForkName, ProposerPreparationData, ProposerSlashing, RelativeEpoch,
+    SignedAggregateAndProof, SignedBeaconBlock, SignedContributionAndProof, SignedVoluntaryExit,
+    Slot, SyncCommitteeMessage, SyncContributionData,
 };
 use version::{
     add_consensus_version_header, fork_versioned_response, inconsistent_fork_rejection,
@@ -2186,6 +2186,53 @@ pub fn serve<T: BeaconChainTypes>(
             },
         );
 
+    // POST validator/prepare_beacon_proposer
+    let post_validator_prepare_beacon_proposer = eth1_v1
+        .and(warp::path("validator"))
+        .and(warp::path("prepare_beacon_proposer"))
+        .and(warp::path::end())
+        .and(not_while_syncing_filter.clone())
+        .and(chain_filter.clone())
+        .and(warp::addr::remote())
+        .and(log_filter.clone())
+        .and(warp::body::json())
+        .and_then(
+            |chain: Arc<BeaconChain<T>>,
+             client_addr: Option<SocketAddr>,
+             log: Logger,
+             preparation_data: Vec<ProposerPreparationData>| {
+                blocking_json_task(move || {
+                    let execution_layer = chain
+                        .execution_layer
+                        .as_ref()
+                        .ok_or(BeaconChainError::ExecutionLayerMissing)
+                        .map_err(warp_utils::reject::beacon_chain_error)?;
+                    let current_epoch = chain
+                        .epoch()
+                        .map_err(warp_utils::reject::beacon_chain_error)?;
+
+                    info!(
+                        log,
+                        "Received proposer preparation data";
+                        "count" => preparation_data.len(),
+                        "client" => client_addr
+                            .map(|a| a.to_string())
+                            .unwrap_or_else(|| "unknown".to_string()),
+                    );
+
+                    execution_layer
+                        .update_proposer_preparation_blocking(current_epoch, &preparation_data)
+                        .map_err(|_e| {
+                            warp_utils::reject::custom_bad_request(
+                                "error processing proposer preparations".to_string(),
+                            )
+                        })?;
+
+                    Ok(())
+                })
+            },
+        );
+
     // POST validator/sync_committee_subscriptions
     let post_validator_sync_committee_subscriptions = eth1_v1
         .and(warp::path("validator"))
@@ -2710,6 +2757,7 @@ pub fn serve<T: BeaconChainTypes>(
                 .or(post_validator_contribution_and_proofs.boxed())
                 .or(post_validator_beacon_committee_subscriptions.boxed())
                 .or(post_validator_sync_committee_subscriptions.boxed())
+                .or(post_validator_prepare_beacon_proposer.boxed())
                 .or(post_lighthouse_liveness.boxed())
                 .or(post_lighthouse_database_reconstruct.boxed())
                 .or(post_lighthouse_database_historical_blocks.boxed()),

beacon_node/src/cli.rs

@@ -108,7 +108,7 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
             Arg::with_name("network-load")
                 .long("network-load")
                 .value_name("INTEGER")
-                .help("Lighthouse's network can be tuned for bandwidth/performance. Setting this to a high value, will increase the bandwidth lighthouse uses, increasing the likelihood of redundant information in exchange for faster communication. This can increase profit of validators marginally by receiving messages faster on the network. Lower values decrease bandwidth usage, but makes communication slower which can lead to validator performance reduction. Values are in the range [1,5].") 
+                .help("Lighthouse's network can be tuned for bandwidth/performance. Setting this to a high value, will increase the bandwidth lighthouse uses, increasing the likelihood of redundant information in exchange for faster communication. This can increase profit of validators marginally by receiving messages faster on the network. Lower values decrease bandwidth usage, but makes communication slower which can lead to validator performance reduction. Values are in the range [1,5].")
                 .default_value("3")
                 .set(clap::ArgSettings::Hidden)
                 .takes_value(true),
@@ -409,9 +409,9 @@ pub fn cli_app<'a, 'b>() -> App<'a, 'b> {
                 .takes_value(true)
         )
         .arg(
-            Arg::with_name("fee-recipient")
-                .long("fee-recipient")
-                .value_name("FEE-RECIPIENT")
+            Arg::with_name("suggested-fee-recipient")
+                .long("suggested-fee-recipient")
+                .value_name("SUGGESTED-FEE-RECIPIENT")
                 .help("Once the merge has happened, this address will receive transaction fees \
                        collected from any blocks produced by this node. Defaults to a junk \
                        address whilst the merge is in development stages. THE DEFAULT VALUE \