feat(oauth): caching, shareable authenticator

bin/builder.rs

@@ -31,14 +31,15 @@ async fn main() -> eyre::Result<()> {
     let builder = builder::tasks::block::BlockBuilder::new(&config);
 
     let submit = builder::tasks::submit::SubmitTask {
-        authenticator,
+        authenticator: authenticator.clone(),
         provider,
         zenith,
         client: reqwest::Client::new(),
         sequencer_signer,
         config: config.clone(),
     };
 
+    let authenticator_jh = authenticator.spawn();
     let (submit_channel, submit_jh) = submit.spawn();
     let (tx_channel, bundle_channel, build_jh) = builder.spawn(submit_channel);
     let tx_poller_jh = tx_poller.spawn(tx_channel.clone());
@@ -62,6 +63,9 @@ async fn main() -> eyre::Result<()> {
         _ = bundle_poller_jh => {
             tracing::info!("bundle_poller finished");
         }
+        _ = authenticator_jh => {
+            tracing::info!("authenticator finished");
+        }
     }
 
     tracing::info!("shutting down");

src/config.rs

@@ -26,6 +26,7 @@ const BUILDER_REWARDS_ADDRESS: &str = "BUILDER_REWARDS_ADDRESS";
 const ROLLUP_BLOCK_GAS_LIMIT: &str = "ROLLUP_BLOCK_GAS_LIMIT";
 const TX_POOL_URL: &str = "TX_POOL_URL";
 const TX_POOL_POLL_INTERVAL: &str = "TX_POOL_POLL_INTERVAL";
+const AUTH_TOKEN_REFRESH_INTERVAL: &str = "AUTH_TOKEN_REFRESH_INTERVAL";
 const TX_POOL_CACHE_DURATION: &str = "TX_POOL_CACHE_DURATION";
 const OAUTH_CLIENT_ID: &str = "OAUTH_CLIENT_ID";
 const OAUTH_CLIENT_SECRET: &str = "OAUTH_CLIENT_SECRET";
@@ -82,6 +83,8 @@ pub struct BuilderConfig {
     pub oauth_token_url: String,
     /// OAuth audience for the builder.
     pub oauth_audience: String,
+    /// The oauth token refresh interval in seconds.
+    pub oauth_token_refresh_interval: u64,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -159,6 +162,7 @@ impl BuilderConfig {
             oauth_authenticate_url: load_string(OAUTH_AUTHENTICATE_URL)?,
             oauth_token_url: load_string(OAUTH_TOKEN_URL)?,
             oauth_audience: load_string(OAUTH_AUDIENCE)?,
+            oauth_token_refresh_interval: load_u64(AUTH_TOKEN_REFRESH_INTERVAL)?,
         })
     }
 

src/tasks/oauth.rs

@@ -1,69 +1,97 @@
 //! Service responsible for authenticating with the cache with Oauth tokens.
+//! This authenticator periodically fetches a new token every set amount of seconds.
+use std::sync::Arc;
+
 use crate::config::BuilderConfig;
 use oauth2::{
     basic::{BasicClient, BasicTokenType},
     reqwest::http_client,
     AuthUrl, ClientId, ClientSecret, EmptyExtraTokenFields, StandardTokenResponse, TokenUrl,
 };
+use tokio::{sync::RwLock, task::JoinHandle};
 
 const OAUTH_AUDIENCE_CLAIM: &str = "audience";
 
 type Token = StandardTokenResponse<EmptyExtraTokenFields, BasicTokenType>;
 
-/// Holds a reference to the current oauth token and the builder config
+/// A self-refreshing, periodically fetching authenticator for the block builder.
+/// It is architected as a shareable struct that can be used across all the multiple builder tasks.
+/// It fetches a new token every set amount of seconds, configured through the general builder config.
+/// Readers are guaranteed to not read stale tokens as the [RwLock] guarantees that write tasks (refreshing the token) will claim priority over read access.
 #[derive(Debug, Clone)]
 pub struct Authenticator {
     pub config: BuilderConfig,
+    inner: Arc<RwLock<AuthenticatorInner>>,
+}
+
+/// Inner state of the Authenticator.
+/// Contains the token that is being used for authentication.
+#[derive(Debug)]
+pub struct AuthenticatorInner {
     pub token: Option<Token>,
 }
 
+impl Default for AuthenticatorInner {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl AuthenticatorInner {
+    pub fn new() -> Self {
+        Self { token: None }
+    }
+}
+
 impl Authenticator {
     /// Creates a new Authenticator from the provided builder config.
     pub fn new(config: &BuilderConfig) -> Self {
         Self {
             config: config.clone(),
-            token: None,
+            inner: Arc::new(RwLock::new(AuthenticatorInner::new())),
         }
     }
 
     /// Requests a new authentication token and, if successful, sets it to as the token
-    pub async fn authenticate(&mut self) -> eyre::Result<()> {
+    pub async fn authenticate(&self) -> eyre::Result<()> {
         let token = self.fetch_oauth_token().await?;
-        dbg!(&token);
-        self.set_token(token);
+        self.set_token(token).await;
         Ok(())
     }
 
     /// Returns true if there is Some token set
-    pub fn is_authenticated(&self) -> bool {
-        self.token.is_some()
+    pub async fn is_authenticated(&self) -> bool {
+        let lock = self.inner.read().await;
+
+        lock.token.is_some()
     }
 
     /// Sets the Authenticator's token to the provided value
-    pub fn set_token(
-        &mut self,
+    pub async fn set_token(
+        &self,
         token: StandardTokenResponse<EmptyExtraTokenFields, BasicTokenType>,
     ) {
-        self.token = Some(token);
+        let mut lock = self.inner.write().await;
+        lock.token = Some(token);
     }
 
     /// Returns the currently set token
-    pub fn token(&self) -> Option<Token> {
-        self.token.clone()
+    pub async fn token(&self) -> Option<Token> {
+        let lock = self.inner.read().await;
+        lock.token.clone()
     }
 
     /// Fetches an oauth token
     pub async fn fetch_oauth_token(
         &self,
     ) -> eyre::Result<StandardTokenResponse<EmptyExtraTokenFields, BasicTokenType>> {
         let config = self.config.clone();
-        dbg!(config.clone());
 
         let client = BasicClient::new(
-            ClientId::new(self.config.oauth_client_id.clone()),
-            Some(ClientSecret::new(self.config.oauth_client_secret.clone())),
-            AuthUrl::new(self.config.oauth_authenticate_url.clone())?,
-            Some(TokenUrl::new(self.config.oauth_token_url.clone())?),
+            ClientId::new(config.oauth_client_id.clone()),
+            Some(ClientSecret::new(config.oauth_client_secret.clone())),
+            AuthUrl::new(config.oauth_authenticate_url.clone())?,
+            Some(TokenUrl::new(config.oauth_token_url.clone())?),
         );
 
         let token_result = client
@@ -73,6 +101,19 @@ impl Authenticator {
 
         Ok(token_result)
     }
+
+    /// Spawns a task that periodically fetches a new token every 300 seconds.
+    pub async fn spawn(self) -> JoinHandle<()> {
+        let interval = self.config.oauth_token_refresh_interval;
+
+        tokio::spawn(async move {
+            loop {
+                tokio::time::sleep(tokio::time::Duration::from_secs(interval)).await;
+                tracing::info!("Refreshing oauth token");
+                self.authenticate().await.unwrap();
+            }
+        })
+    }
 }
 
 mod tests {
@@ -90,7 +131,7 @@ mod tests {
         let auth = Authenticator::new(&config);
         let token = auth.fetch_oauth_token().await?;
         dbg!(&token);
-        let token = auth.token().unwrap();
+        let token = auth.token().await.unwrap();
         println!("{:?}", token);
         assert!(!token.access_token().secret().is_empty());
         Ok(())
@@ -120,6 +161,7 @@ mod tests {
             oauth_token_url: "http://localhost:9000".into(),
             oauth_audience: "https://transactions.holesky.signet.sh".into(),
             tx_broadcast_urls: vec!["http://localhost:9000".into()],
+            oauth_token_refresh_interval: 300, // 5 minutes
         };
         Ok((BlockBuilder::new(&config), config))
     }

tests/bundle_poller_test.rs

@@ -43,6 +43,7 @@ mod tests {
             oauth_token_url: "http://localhost:8080".into(),
             oauth_audience: "https://transactions.holesky.signet.sh".into(),
             tx_broadcast_urls: vec!["http://localhost:9000".into()],
+            oauth_token_refresh_interval: 300, // 5 minutes
         };
         Ok((BlockBuilder::new(&config), config))
     }

tests/tx_poller_test.rs

@@ -86,6 +86,7 @@ mod tests {
             oauth_authenticate_url: "http://localhost:8080".into(),
             oauth_token_url: "http://localhost:8080".into(),
             oauth_audience: "https://transactions.holesky.signet.sh".into(),
+            oauth_token_refresh_interval: 300, // 5 minutes
         };
         Ok((BlockBuilder::new(&config), config))
     }