Misc fixes running in kurtosis

Dockerfile

@@ -16,10 +16,10 @@ ENV FEATURES ${FEATURES}
 
 RUN apt-get update && apt-get -y upgrade && apt-get install -y libclang-dev pkg-config
 
-RUN cargo chef cook --profile ${BUILD_PROFILE} --features ${FEATURES} --recipe-path recipe.json
+RUN cargo chef cook --profile ${BUILD_PROFILE} --features "$FEATURES" --recipe-path recipe.json
 
 COPY . .
-RUN cargo build --profile ${BUILD_PROFILE} --features ${FEATURES}  --locked --bin mev
+RUN cargo build --profile ${BUILD_PROFILE} --features "$FEATURES"  --locked --bin mev
 
 RUN cp /app/target/${BUILD_PROFILE}/mev /app/mev
 

justfile

@@ -1,11 +1,11 @@
 validate-example-config:
     cargo run config example.config.toml
 
-docker-build:
+build-docker:
     docker build -t ralexstokes/mev-rs .
-docker-push:
+push-docker:
     docker push ralexstokes/mev-rs
-docker-update: docker-build docker-push
+update-docker-hub: build-docker push-docker
 
 test:
     cargo test --all

mev-build-rs/src/auctioneer/auction_schedule.rs

@@ -9,7 +9,7 @@ use std::{
 pub type RelaySet = HashSet<Arc<Relay>>;
 pub type Proposals = HashMap<Proposer, RelaySet>;
 
-#[derive(Debug, Default, Hash, PartialEq, Eq)]
+#[derive(Debug, Clone, Default, Hash, PartialEq, Eq)]
 pub struct Proposer {
     pub public_key: BlsPublicKey,
     pub fee_recipient: Address,
@@ -26,8 +26,8 @@ impl AuctionSchedule {
         self.schedule.retain(|&slot, _| slot >= retain_slot);
     }
 
-    pub fn take_matching_proposals(&mut self, slot: Slot) -> Option<Proposals> {
-        self.schedule.remove(&slot)
+    pub fn get_matching_proposals(&self, slot: Slot) -> Option<&Proposals> {
+        self.schedule.get(&slot)
     }
 
     pub fn process(&mut self, relay: Arc<Relay>, schedule: &[ProposerSchedule]) -> Vec<Slot> {

mev-build-rs/src/auctioneer/service.rs

@@ -24,13 +24,16 @@ use reth::{
     payload::{EthBuiltPayload, Events, PayloadBuilderHandle, PayloadId, PayloadStore},
 };
 use serde::Deserialize;
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
 use tokio::sync::{
     broadcast,
     mpsc::{Receiver, Sender},
 };
 use tokio_stream::StreamExt;
-use tracing::{error, info, warn};
+use tracing::{error, info, trace, warn};
 
 fn make_attributes_for_proposer(
     attributes: &BuilderPayloadBuilderAttributes,
@@ -129,6 +132,7 @@ pub struct Service<
 
     auction_schedule: AuctionSchedule,
     open_auctions: HashMap<PayloadId, Arc<AuctionContext>>,
+    processed_payload_attributes: HashMap<Slot, HashSet<PayloadId>>,
 }
 
 impl<
@@ -168,6 +172,7 @@ impl<
             bidder_rx,
             auction_schedule: Default::default(),
             open_auctions: Default::default(),
+            processed_payload_attributes: Default::default(),
         }
     }
 
@@ -188,29 +193,36 @@ impl<
         }
 
         // NOTE: clear stale state
-        let slot = epoch * self.context.slots_per_epoch;
-        self.auction_schedule.clear(slot);
-        self.open_auctions.retain(|_, auction| auction.slot >= slot);
+        let retain_slot = epoch * self.context.slots_per_epoch;
+        self.auction_schedule.clear(retain_slot);
+        self.open_auctions.retain(|_, auction| auction.slot >= retain_slot);
+        self.processed_payload_attributes.retain(|&slot, _| slot >= retain_slot);
     }
 
-    fn take_proposals(&mut self, slot: Slot) -> Option<Proposals> {
-        self.auction_schedule.take_matching_proposals(slot)
+    fn get_proposals(&self, slot: Slot) -> Option<&Proposals> {
+        self.auction_schedule.get_matching_proposals(slot)
     }
 
     async fn process_proposals(
         &self,
         slot: Slot,
         attributes: BuilderPayloadBuilderAttributes,
-        proposals: Proposals,
+        proposals: &Proposals,
     ) -> Vec<AuctionContext> {
         let mut new_auctions = vec![];
         for (proposer, relays) in proposals {
-            let attributes = make_attributes_for_proposer(&attributes, &proposer);
+            let attributes = make_attributes_for_proposer(&attributes, proposer);
 
             if self.start_build(&attributes).await.is_some() {
                 // TODO: can likely skip full attributes in `AuctionContext`
                 // TODO: consider data layout here...
-                let auction = AuctionContext { slot, attributes, proposer, relays };
+                // TODO: can likely refactor around auction schedule to skip some clones...
+                let auction = AuctionContext {
+                    slot,
+                    attributes,
+                    proposer: proposer.clone(),
+                    relays: relays.clone(),
+                };
                 new_auctions.push(auction);
             }
         }
@@ -251,8 +263,17 @@ impl<
             self.context.seconds_per_slot,
         )
         .expect("is past genesis");
+
+        let processed_set = self.processed_payload_attributes.entry(slot).or_default();
+        let is_new = processed_set.insert(attributes.payload_id());
+
+        if !is_new {
+            trace!(payload_id = %attributes.payload_id(), "ignoring duplicate payload attributes");
+            return
+        }
+
         // TODO: consolidate once stable
-        if let Some(proposals) = self.take_proposals(slot) {
+        if let Some(proposals) = self.get_proposals(slot) {
             let auctions = self.process_proposals(slot, attributes, proposals).await;
             for auction in auctions {
                 self.process_new_auction(auction).await;
@@ -285,13 +306,12 @@ impl<
             BidderMessage::Dispatch { payload_id, value: _value, keep_alive: _keep_alive } => {
                 // TODO: forward keep alive signal to builder
                 // TODO: sort out streaming comms to builder
+                // TOOD: backpressure on bidder...?
                 if let Some(payload) = self.payload_store.resolve(payload_id).await {
                     match payload {
                         Ok(payload) => self.submit_payload(payload).await,
                         Err(err) => warn!(%err, "payload resolution failed"),
                     }
-                } else {
-                    warn!(%payload_id, "no payload could be retrieved from payload store for bid")
                 }
             }
             _ => {}

mev-rs/src/genesis.rs

@@ -33,8 +33,9 @@ pub async fn get_genesis_time(
             }
 
             // fallback
-            warn!("could not get genesis time from context or connection to consensus node; using best guess");
-            typical_genesis_time(context)
+            let genesis_time = typical_genesis_time(context);
+            warn!(genesis_time, "could not get genesis time from context or connection to consensus node; using best guess");
+            genesis_time
         }
     }
 }