feat(validator): publish proposer preferences (gloas)

CLAUDE.md

@@ -853,8 +853,11 @@ transitions are in `ethlambda-types`, per the section above. Nothing above
   `SlotClock` picks the
   offsets by each slot's own fork, and `/eth/v1/config/spec` supplies them. With
   `--builder-proposals`, pre-gloas slots register validators with the node's builder
-  and may sign a blinded block, published to the producing node only. It
-  still keeps no slashing-protection record; PTC votes have an in-memory
+  and may sign a blinded block, published to the producing node only. From the
+  epoch before gloas it signs proposer preferences every slot for its proposals in
+  the current and next epoch (`PreferencesService`, v2 duties, the gossip rule's
+  signing domain), unconditionally, and `produceBlockV4` carries the boost factor.
+  It still keeps no slashing-protection record; PTC votes have an in-memory
   `(validator, slot)` dedup only, and envelopes are unguarded.
   `get_head_node` is one bottom-up walk for every fork, with
   the current slot's fork selecting the payload rules and the boost gate, so a

bin/ethlambda/src/validator.rs

@@ -78,9 +78,10 @@ pub(crate) struct ValidatorOptions {
 
     /// Propose builder blocks: register each validator with the beacon node's
     /// builder and accept a blinded block when the builder's bid wins. Requires
-    /// a beacon node started with a builder (ethlambda: `--builder`). Fulu and
-    /// earlier only; from gloas the builder market works through proposer
-    /// preferences instead.
+    /// a beacon node started with a builder (ethlambda: `--builder`). The
+    /// registrations and blinded blocks are fulu and earlier only; from gloas
+    /// the builder market works through proposer preferences, and this flag's
+    /// boost factor weighs p2p bids in `produceBlockV4`.
     #[arg(long, default_value = "false")]
     pub(crate) builder_proposals: bool,
 

crates/net/rpc/src/beacon/builder_market_tests.rs

@@ -1061,6 +1061,92 @@ async fn valid_preferences_are_cached_and_published_once() {
     assert_eq!(world.network.proposer_preferences.lock().unwrap().len(), 1);
 }
 
+/// The validator client's own path end to end against this node: it reads the
+/// proposer duty and its v2 `dependent_root` from the served duties route,
+/// signs with `SigningContext::sign_proposer_preferences`, and posts through
+/// `HttpBeaconNode`, and the node's gossip rules accept it. That pins the
+/// duties route's dependent root as the one the preference must carry, and the
+/// JSON the client posts as the one the route decodes.
+///
+/// It does not pin the fork-version choice: this world has gloas at epoch 0
+/// and the node accepts either version, so `signing.rs`'s unit test is the pin
+/// for the domain at a fork boundary.
+#[tokio::test]
+async fn preferences_signed_by_the_validator_client_pass_this_nodes_gossip_rules() {
+    use ethlambda_blockchain::{SyncStatusController, metrics::SyncStatus};
+    use ethlambda_validator::{
+        beacon_node::{BeaconNodeApi as _, dto::parse_pubkey, http::HttpBeaconNode},
+        keys::ValidatorStore,
+        signing::SigningContext,
+    };
+
+    let world = World::new();
+    let duties_routes = crate::beacon::validator::routes()
+        .with_state(world.store.clone())
+        .layer(Extension(SyncStatusController::new(SyncStatus::Synced)));
+    let router = world.app(None).merge(duties_routes);
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let bound = listener.local_addr().unwrap();
+    tokio::spawn(async move { axum::serve(listener, router).await });
+    let client = HttpBeaconNode::new(format!("http://{bound}")).unwrap();
+
+    // The duty and the dependent root the way the validator client reads them.
+    let duties = client
+        .proposer_duties_v2(compute_epoch_at_slot(SLOT))
+        .await
+        .expect("the node serves v2 proposer duties");
+    let duty = duties
+        .duties
+        .iter()
+        .find(|duty| duty.slot == SLOT)
+        .expect("the slot has a proposer");
+    let proposer = duty.validator_index;
+    let message = ProposerPreferences {
+        dependent_root: duties.dependent_root,
+        proposal_slot: SLOT,
+        validator_index: proposer,
+        fee_recipient: address(0xaa),
+        target_gas_limit: 30_000_000,
+    };
+
+    let mut keys = ValidatorStore::new();
+    let pubkey = keys
+        .insert_secret("test", &secret_key_for(proposer as usize).to_bytes())
+        .expect("the test secret is a valid key");
+    assert_eq!(
+        pubkey,
+        parse_pubkey(&duty.pubkey).expect("the duty names a key"),
+        "the test registry's key for the proposer"
+    );
+    let context = SigningContext {
+        config: chain_config(),
+        genesis_validators_root: world.state.genesis_validators_root(),
+    };
+    let signature = context
+        .sign_proposer_preferences(&keys, &pubkey, &message)
+        .expect("signs");
+    let signed = SignedProposerPreferences { message, signature };
+
+    client
+        .submit_proposer_preferences(
+            ethlambda_types::beacon::fork::ForkName::Gloas,
+            std::slice::from_ref(&signed),
+        )
+        .await
+        .expect("the node accepts what the validator client signed");
+
+    assert_eq!(
+        *world.network.proposer_preferences.lock().unwrap(),
+        vec![signed.clone()]
+    );
+    assert_eq!(
+        world
+            .market
+            .preferences(SLOT, signed.message.dependent_root),
+        Some(signed)
+    );
+}
+
 #[tokio::test]
 async fn cached_preferences_resubmitted_are_a_200_without_a_publish() {
     // Needs only the market: an identical entry short-circuits the rules.

crates/net/rpc/src/beacon/validator.rs

@@ -35,7 +35,9 @@ use serde::{Deserialize, Serialize};
 
 use ethlambda_network_api::RpcToP2PRef;
 use ethlambda_state_transition::beacon::{
-    fork_choice::{checkpoint_state, get_current_store_epoch, get_head_node},
+    fork_choice::{
+        checkpoint_state, compute_shuffling_dependent_slot, get_current_store_epoch, get_head_node,
+    },
     gossip::attestation::compute_subnet_for_attestation,
     helpers::accessors::{CommitteeCacheExt as _, get_block_root_at_slot},
     helpers::altair::compute_sync_committee_period,
@@ -535,11 +537,11 @@ impl DependentRoot {
     /// definition names the genesis block where its subtraction would
     /// underflow, which is slot 0.
     fn slot(self, epoch: Epoch) -> Slot {
-        let epoch = match self {
-            Self::V1 => epoch,
-            Self::V2 => epoch.saturating_sub(1),
-        };
-        compute_start_slot_at_epoch(epoch).saturating_sub(1)
+        match self {
+            Self::V1 => compute_start_slot_at_epoch(epoch).saturating_sub(1),
+            // The state transition's own definition, so the two cannot drift.
+            Self::V2 => compute_shuffling_dependent_slot(epoch),
+        }
     }
 }
 

crates/validator/src/beacon_node/fallback.rs

@@ -86,6 +86,32 @@ impl<B: BeaconNodeApi> FallbackBeaconNode<B> {
     /// answered, for the calls whose follow-up must go to the same node. The
     /// logging and the final `warn!` live here so the two cannot drift.
     async fn try_each_at<'p, T, F>(&'p self, what: &'static str, call: F) -> Result<(usize, T)>
+    where
+        F: Fn(&'p B) -> BoxFuture<'p, Result<T>>,
+    {
+        self.try_each_at_inner(what, true, call).await
+    }
+
+    /// [`Self::try_each`] without the final `warn!`, for a caller whose own
+    /// duty did not depend on the answer and which reports the failure itself
+    /// (the counterpart of [`Self::try_all`]'s `all_failed: None`). A node that
+    /// is syncing, or lacks the endpoint, would otherwise make this warn on
+    /// every call for as long as it lasts.
+    async fn try_each_quiet<'p, T, F>(&'p self, what: &'static str, call: F) -> Result<T>
+    where
+        F: Fn(&'p B) -> BoxFuture<'p, Result<T>>,
+    {
+        self.try_each_at_inner(what, false, call)
+            .await
+            .map(|(_, value)| value)
+    }
+
+    async fn try_each_at_inner<'p, T, F>(
+        &'p self,
+        what: &'static str,
+        warn_when_all_fail: bool,
+        call: F,
+    ) -> Result<(usize, T)>
     where
         F: Fn(&'p B) -> BoxFuture<'p, Result<T>>,
     {
@@ -106,7 +132,9 @@ impl<B: BeaconNodeApi> FallbackBeaconNode<B> {
         let detail = last
             .map(|err| err.to_string())
             .unwrap_or_else(|| "no nodes configured".to_string());
-        warn!(%what, %detail, "All configured beacon nodes failed; duty likely missed");
+        if warn_when_all_fail {
+            warn!(%what, %detail, "All configured beacon nodes failed; duty likely missed");
+        }
         Err(Error::AllBeaconNodesFailed(detail))
     }
 
@@ -142,6 +170,39 @@ impl<B: BeaconNodeApi> FallbackBeaconNode<B> {
         all_failed: Option<&'static str>,
         call: F,
     ) -> Result<()>
+    where
+        F: Fn(&'p B) -> BoxFuture<'p, Result<()>>,
+    {
+        self.try_all_inner(what, node_effect, all_failed, true, call)
+            .await
+    }
+
+    /// [`Self::try_all`] logging each node's refusal at `debug!` instead of
+    /// `warn!`, for a call the caller retries every slot and reports itself
+    /// (once per entry): a per-node warning here would repeat on every retry
+    /// for as long as the refusal lasts. Always without an `all_failed`
+    /// message, for the same reason.
+    async fn try_all_quiet<'p, F>(
+        &'p self,
+        what: &'static str,
+        node_effect: &'static str,
+        call: F,
+    ) -> Result<()>
+    where
+        F: Fn(&'p B) -> BoxFuture<'p, Result<()>>,
+    {
+        self.try_all_inner(what, node_effect, None, false, call)
+            .await
+    }
+
+    async fn try_all_inner<'p, F>(
+        &'p self,
+        what: &'static str,
+        node_effect: &'static str,
+        all_failed: Option<&'static str>,
+        warn_per_node: bool,
+        call: F,
+    ) -> Result<()>
     where
         F: Fn(&'p B) -> BoxFuture<'p, Result<()>>,
     {
@@ -151,7 +212,11 @@ impl<B: BeaconNodeApi> FallbackBeaconNode<B> {
             match call(node).await {
                 Ok(()) => accepted += 1,
                 Err(err) => {
-                    warn!(%what, position, %err, node_effect, "Beacon node did not accept");
+                    if warn_per_node {
+                        warn!(%what, position, %err, node_effect, "Beacon node did not accept");
+                    } else {
+                        debug!(%what, position, %err, node_effect, "Beacon node did not accept");
+                    }
                     last = Some(err);
                 }
             }
@@ -256,6 +321,13 @@ impl<B: BeaconNodeApi> BeaconNodeApi for FallbackBeaconNode<B> {
             .await
     }
 
+    async fn proposer_duties_v2(&self, epoch: Epoch) -> Result<ProposerDuties> {
+        // Quiet: the preferences service reports its own failure once per
+        // epoch, and no duty depends on this answer.
+        self.try_each_quiet("proposer_duties_v2", |node| node.proposer_duties_v2(epoch))
+            .await
+    }
+
     async fn attestation_data(&self, slot: Slot, fork: ForkName) -> Result<AttestationData> {
         self.try_each("attestation_data", |node| node.attestation_data(slot, fork))
             .await
@@ -342,6 +414,24 @@ impl<B: BeaconNodeApi> BeaconNodeApi for FallbackBeaconNode<B> {
         .await
     }
 
+    /// To every node, since each one gossips what it accepts: a preference
+    /// only the first node heard reaches the mesh through that node alone, and
+    /// builders subscribed elsewhere would ignore the slot.
+    async fn submit_proposer_preferences(
+        &self,
+        fork: ForkName,
+        preferences: &[gloas::SignedProposerPreferences],
+    ) -> Result<()> {
+        // Quiet: the service retries a refused preference every slot and warns
+        // once per entry, with what it costs.
+        self.try_all_quiet(
+            "submit_proposer_preferences",
+            "it will not gossip these proposer preferences",
+            |node| node.submit_proposer_preferences(fork, preferences),
+        )
+        .await
+    }
+
     /// Fails over like any query: the envelope the node cached is only on the
     /// node that produced the block, and one that did not answers 404.
     async fn execution_payload_envelope(
@@ -1061,6 +1151,49 @@ mod tests {
             .expect_err("the record is gone");
     }
 
+    #[tokio::test]
+    async fn proposer_preferences_reach_every_node_past_a_failing_one() {
+        use ethlambda_types::beacon::containers::gloas::{
+            ProposerPreferences, SignedProposerPreferences,
+        };
+
+        let preference = SignedProposerPreferences {
+            message: ProposerPreferences {
+                proposal_slot: 70,
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        let fallback = FallbackBeaconNode::new(vec![
+            MockBeaconNode::new(),
+            MockBeaconNode::failing("middle down"),
+            MockBeaconNode::new(),
+        ]);
+
+        fallback
+            .submit_proposer_preferences(ForkName::Fulu, std::slice::from_ref(&preference))
+            .await
+            .expect("two of three accepted");
+
+        for position in [0, 2] {
+            assert_eq!(
+                *fallback.nodes[position]
+                    .submitted_preferences
+                    .lock()
+                    .expect("lock"),
+                vec![(ForkName::Fulu, vec![preference.clone()])],
+                "node {position}"
+            );
+        }
+        assert!(
+            fallback.nodes[1]
+                .submitted_preferences
+                .lock()
+                .expect("lock")
+                .is_empty()
+        );
+    }
+
     #[tokio::test]
     async fn registrations_reach_every_node() {
         use ethlambda_types::beacon::builder::{