From 97f7e1673e7f9b2e3be7b465f75a3b86bfaed506 Mon Sep 17 00:00:00 2001
From: Bryan Stitt <bryan@llamanodes.com>
Date: Fri, 6 Oct 2023 20:50:57 -0700
Subject: [PATCH] finish todo for waiting for head_block_sender

---
 web3_proxy/src/rpcs/one.rs | 25 +++++++++++++++++++++----
 1 file changed, 21 insertions(+), 4 deletions(-)

diff --git a/web3_proxy/src/rpcs/one.rs b/web3_proxy/src/rpcs/one.rs
index b415549d..f1e90af1 100644
--- a/web3_proxy/src/rpcs/one.rs
+++ b/web3_proxy/src/rpcs/one.rs
@@ -26,6 +26,7 @@ use std::fmt;
 use std::hash::{Hash, Hasher};
 use std::sync::atomic::{self, AtomicU32, AtomicU64, AtomicUsize};
 use std::{cmp::Ordering, sync::Arc};
+use tokio::select;
 use tokio::sync::{mpsc, watch, RwLock as AsyncRwLock};
 use tokio::time::{interval, sleep, sleep_until, Duration, Instant, MissedTickBehavior};
 use tracing::{debug, error, info, trace, warn, Level};
@@ -983,6 +984,8 @@ impl Web3Rpc {
         web3_request: &Arc<Web3Request>,
         error_handler: Option<RequestErrorHandler>,
     ) -> Web3ProxyResult<OpenRequestHandle> {
+        let mut head_block_sender = None;
+
         loop {
             match self.try_request_handle(web3_request, error_handler).await {
                 Ok(OpenRequestResult::Handle(handle)) => return Ok(handle),
@@ -1013,14 +1016,28 @@ impl Web3Rpc {
                     // TODO: when can this happen? log? emit a stat?
                     trace!("{} has no handle ready", self);
 
-                    // TODO: sleep how long? maybe just error?
-                    // TODO: instead of an arbitrary sleep, subscribe to the head block on this?
-                    // TODO: exponential backoff with jitter
-                    todo!();
+                    if head_block_sender.is_none() {
+                        head_block_sender = self.head_block_sender.as_ref().map(|x| x.subscribe());
+                    }
+
+                    if let Some(head_block_sender) = &mut head_block_sender {
+                        select! {
+                            _ = head_block_sender.changed() => {
+                                head_block_sender.borrow_and_update();
+                            }
+                            _ = sleep_until(web3_request.expire_instant) => {
+                                break;
+                            }
+                        }
+                    } else {
+                        break;
+                    }
                 }
                 Err(err) => return Err(err),
             }
         }
+
+        Err(Web3ProxyError::NoServersSynced)
     }
 
     async fn wait_for_throttle(self: &Arc<Self>, wait_until: Instant) -> Web3ProxyResult<u64> {