fix consensus finding bug

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);
     }
 }
 

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;

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) {

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;