kabu_broadcast_broadcaster/
flashbots.rs

1use std::sync::Arc;
2
3use alloy_primitives::Bytes;
4use alloy_provider::{ProviderBuilder, ext::MevApi};
5use alloy_rpc_types_mev::EthSendBundle;
6use alloy_signer_local::PrivateKeySigner;
7use eyre::{Result, eyre};
8use reth_tasks::TaskExecutor;
9use tokio::sync::broadcast;
10use tokio::sync::broadcast::error::RecvError;
11use tracing::{debug, error};
12use url::Url;
13
14use kabu_core_components::Component;
15use kabu_types_events::{MessageTxCompose, RlpState, TxComposeData, TxComposeMessageType};
16
17#[derive(Clone)]
18pub struct RelayConfig {
19    pub id: u64,
20    pub url: String,
21    pub name: String,
22    pub no_sign: Option<bool>,
23}
24
25#[derive(Clone)]
26struct FlashbotsRelay {
27    url: String,
28    signer: Option<PrivateKeySigner>,
29    name: String,
30}
31
32impl FlashbotsRelay {
33    fn new(url: &str, signer: Option<PrivateKeySigner>) -> Result<Self> {
34        Ok(FlashbotsRelay { url: url.to_string(), signer, name: url.to_string() })
35    }
36
37    async fn send_bundle(&self, bundle: EthSendBundle) -> Result<()> {
38        let provider = ProviderBuilder::new().connect_http(Url::parse(&self.url)?);
39
40        let result = match self.signer.clone() {
41            Some(signer) => provider.send_bundle(bundle).with_auth(signer).await,
42            None => provider.send_bundle(bundle).await,
43        };
44
45        match result {
46            Ok(_resp) => {
47                debug!("Bundle sent to: {}", self.name);
48                Ok(())
49            }
50            Err(error) => {
51                error!("{} {}", self.name, error.to_string());
52                Err(eyre!("FLASHBOTS_RELAY_ERROR"))
53            }
54        }
55    }
56}
57
58async fn broadcast_task(broadcast_request: TxComposeData, relays: Arc<Vec<FlashbotsRelay>>) -> Result<()> {
59    let block_number = broadcast_request.next_block_number;
60
61    if let Some(rlp_bundle) = broadcast_request.rlp_bundle.clone() {
62        let stuffing_rlp_bundle: Vec<Bytes> = rlp_bundle.iter().map(|item| item.unwrap()).collect();
63        let backrun_rlp_bundle: Vec<Bytes> =
64            rlp_bundle.iter().filter(|item| matches!(item, RlpState::Backrun(_))).map(|item| item.unwrap()).collect();
65
66        if stuffing_rlp_bundle.iter().any(|i| i.is_empty()) || backrun_rlp_bundle.iter().any(|i| i.is_empty()) {
67            return Err(eyre!("RLP_BUNDLE_IS_INCORRECT"));
68        }
69
70        // Send backrun bundle
71        let backrun_bundle = EthSendBundle { txs: backrun_rlp_bundle, block_number, ..Default::default() };
72
73        // Send stuffing bundle
74        let stuffing_bundle = EthSendBundle { txs: stuffing_rlp_bundle, block_number, ..Default::default() };
75
76        // Broadcast to all relays concurrently
77        for relay in relays.iter() {
78            let relay_clone = relay.clone();
79            let backrun_bundle_clone = backrun_bundle.clone();
80            let stuffing_bundle_clone = stuffing_bundle.clone();
81
82            tokio::spawn(async move {
83                debug!("Sending bundles to {}", relay_clone.name);
84
85                if let Err(e) = relay_clone.send_bundle(backrun_bundle_clone).await {
86                    error!("Failed to send backrun bundle to {}: {}", relay_clone.name, e);
87                }
88
89                if let Err(e) = relay_clone.send_bundle(stuffing_bundle_clone).await {
90                    error!("Failed to send stuffing bundle to {}: {}", relay_clone.name, e);
91                }
92            });
93        }
94
95        Ok(())
96    } else {
97        error!("rlp_bundle is None");
98        Err(eyre!("RLP_BUNDLE_IS_NONE"))
99    }
100}
101
102async fn flashbots_broadcaster_worker(
103    relays: Arc<Vec<FlashbotsRelay>>,
104    mut bundle_rx: broadcast::Receiver<MessageTxCompose>,
105    allow_broadcast: bool,
106) {
107    loop {
108        tokio::select! {
109            msg = bundle_rx.recv() => {
110                let broadcast_msg : Result<MessageTxCompose, RecvError> = msg;
111                match broadcast_msg {
112                    Ok(compose_request) => {
113                        if let TxComposeMessageType::Broadcast(broadcast_request) = compose_request.inner
114                            && allow_broadcast {
115                                tokio::task::spawn(
116                                    broadcast_task(
117                                        broadcast_request,
118                                        relays.clone(),
119                                    )
120                                );
121                        }
122                    }
123                    Err(RecvError::Closed) => {
124                        debug!("flashbots_broadcaster_worker channel closed, shutting down");
125                        break;
126                    }
127                    Err(e) => {
128                        error!("flashbots_broadcaster_worker {}", e)
129                    }
130                }
131            }
132        }
133    }
134}
135
136pub struct FlashbotsBroadcastComponent {
137    relays: Arc<Vec<FlashbotsRelay>>,
138    signer: Arc<PrivateKeySigner>,
139    tx_compose_channel_rx: Option<broadcast::Sender<MessageTxCompose>>,
140    allow_broadcast: bool,
141}
142
143impl FlashbotsBroadcastComponent {
144    pub fn new(signer: Option<PrivateKeySigner>, allow_broadcast: bool) -> Result<Self> {
145        let signer = Arc::new(signer.unwrap_or(PrivateKeySigner::random()));
146        let relays = Arc::new(Vec::new());
147
148        Ok(FlashbotsBroadcastComponent { relays, signer, tx_compose_channel_rx: None, allow_broadcast })
149    }
150
151    pub fn with_default_relays(mut self) -> Result<Self> {
152        let relay_configs = vec![
153            ("https://relay.flashbots.net", false),
154            ("https://rpc.beaverbuild.org/", false),
155            ("https://rpc.titanbuilder.xyz", false),
156            ("https://rsync-builder.xyz", false),
157            ("https://api.edennetwork.io/v1/bundle", false),
158            ("https://eth-builder.com", true),
159            ("https://api.securerpc.com/v1", true),
160            ("https://BuildAI.net", true),
161            ("https://rpc.payload.de", true),
162            ("https://rpc.f1b.io", false),
163            ("https://rpc.lokibuilder.xyz", false),
164            ("https://rpc.ibuilder.xyz", false),
165            ("https://rpc.jetbldr.xyz", false),
166            ("https://rpc.penguinbuild.org", false),
167            ("https://builder.gmbit.co/rpc", false),
168        ];
169
170        let mut relays = Vec::new();
171        for (url, no_sign) in relay_configs {
172            let signer = if no_sign { None } else { Some((*self.signer).clone()) };
173            relays.push(FlashbotsRelay::new(url, signer)?);
174        }
175
176        self.relays = Arc::new(relays);
177        Ok(self)
178    }
179
180    pub fn with_relay(mut self, url: &str) -> Result<Self> {
181        let mut relays = Arc::try_unwrap(self.relays).unwrap_or_else(|arc| (*arc).clone());
182
183        relays.push(FlashbotsRelay::new(url, Some((*self.signer).clone()))?);
184        self.relays = Arc::new(relays);
185        Ok(self)
186    }
187
188    pub fn with_relays(mut self, relay_configs: Vec<RelayConfig>) -> Result<Self> {
189        let mut relays = Vec::new();
190
191        for config in relay_configs {
192            let signer = if config.no_sign.unwrap_or(false) { None } else { Some((*self.signer).clone()) };
193            relays.push(FlashbotsRelay::new(&config.url, signer)?);
194        }
195
196        self.relays = Arc::new(relays);
197        Ok(self)
198    }
199
200    pub fn with_channel(self, channel: broadcast::Sender<MessageTxCompose>) -> Self {
201        Self { tx_compose_channel_rx: Some(channel), ..self }
202    }
203}
204
205impl Component for FlashbotsBroadcastComponent {
206    fn spawn(self, executor: TaskExecutor) -> Result<()> {
207        let name = self.name();
208        let bundle_rx = self.tx_compose_channel_rx.ok_or_else(|| eyre!("tx_compose_channel_rx not set"))?.subscribe();
209
210        executor.spawn_critical(name, flashbots_broadcaster_worker(self.relays.clone(), bundle_rx, self.allow_broadcast));
211
212        Ok(())
213    }
214    fn name(&self) -> &'static str {
215        "FlashbotsBroadcastComponent"
216    }
217}
kabu_broadcast_broadcaster/flashbots.rs

kabu_broadcast_broadcaster/
flashbots.rs
