From c3cdf7ef4360985448006f717508ab3b0401954f Mon Sep 17 00:00:00 2001
From: Bryan Stitt <bryan@llamanodes.com>
Date: Fri, 12 May 2023 23:40:15 -0700
Subject: [PATCH] fix consensus finding bug

---
 latency/src/peak_ewma/mod.rs      | 5 ++++-
 web3_proxy/src/rpcs/blockchain.rs | 2 +-
 web3_proxy/src/rpcs/consensus.rs  | 4 +++-
 web3_proxy/src/rpcs/one.rs        | 6 +++++-
 4 files changed, 13 insertions(+), 4 deletions(-)

diff --git a/latency/src/peak_ewma/mod.rs b/latency/src/peak_ewma/mod.rs
index 94dca1ab..640318c9 100644
--- a/latency/src/peak_ewma/mod.rs
+++ b/latency/src/peak_ewma/mod.rs
@@ -122,8 +122,11 @@ impl PeakEwmaLatencyTask {
             self.update_at,
         );
 
-        self.rtt_estimate
+        let x = self
+            .rtt_estimate
             .fetch_update(|mut rtt_estimate| rtt_estimate.update(rtt, self.decay_ns, now));
+
+        info!("x: {:?}", x);
     }
 }
 
diff --git a/web3_proxy/src/rpcs/blockchain.rs b/web3_proxy/src/rpcs/blockchain.rs
index dc1ef8c2..f0c1bdbd 100644
--- a/web3_proxy/src/rpcs/blockchain.rs
+++ b/web3_proxy/src/rpcs/blockchain.rs
@@ -435,7 +435,7 @@ impl Web3Rpcs {
             Ok(Some(x)) => x,
         };
 
-        trace!("new_synced_connections: {:?}", new_synced_connections);
+        trace!("new_synced_connections: {:#?}", new_synced_connections);
 
         let watch_consensus_head_sender = self.watch_consensus_head_sender.as_ref().unwrap();
         let consensus_tier = new_synced_connections.tier;
diff --git a/web3_proxy/src/rpcs/consensus.rs b/web3_proxy/src/rpcs/consensus.rs
index 8807895d..a5658522 100644
--- a/web3_proxy/src/rpcs/consensus.rs
+++ b/web3_proxy/src/rpcs/consensus.rs
@@ -375,9 +375,11 @@ impl ConsensusFinder {
 
         trace!("first_tier: {}", current_tier);
 
+        trace!("rpc_heads_by_tier: {:#?}", rpc_heads_by_tier);
+
         // loop over all the rpc heads (grouped by tier) and their parents to find consensus
         // TODO: i'm sure theres a lot of shortcuts that could be taken, but this is simplest to implement
-        for (rpc, rpc_head) in self.rpc_heads.iter() {
+        for (rpc, rpc_head) in rpc_heads_by_tier.into_iter() {
             if current_tier != rpc.tier {
                 // we finished processing a tier. check for primary results
                 if let Some(consensus) = self.count_votes(&primary_votes, web3_rpcs) {
diff --git a/web3_proxy/src/rpcs/one.rs b/web3_proxy/src/rpcs/one.rs
index ee792049..be0487e0 100644
--- a/web3_proxy/src/rpcs/one.rs
+++ b/web3_proxy/src/rpcs/one.rs
@@ -236,7 +236,11 @@ impl Web3Rpc {
     }
 
     pub fn peak_ewma(&self) -> OrderedFloat<f64> {
-        let peak_latency = self.peak_latency.as_ref().unwrap().latency().as_secs_f64();
+        let peak_latency = if let Some(peak_latency) = self.peak_latency.as_ref() {
+            peak_latency.latency().as_secs_f64()
+        } else {
+            0.0
+        };
 
         // TODO: what ordering?
         let active_requests = self.active_requests.load(atomic::Ordering::Acquire) as f64 + 1.0;