configurable backoff

Author: keyvankhademi

lazer/sdk/rust/client/examples/subscribe_price_feeds.rs

@@ -1,3 +1,4 @@
+use backoff::ExponentialBackoffBuilder;
 use base64::Engine;
 use pyth_lazer_client::client::PythLazerClient;
 use pyth_lazer_client::ws_connection::AnyResponse;
@@ -36,7 +37,10 @@ async fn main() -> anyhow::Result<()> {
         vec!["wss://pyth-lazer.dourolabs.app/v1/stream".to_string()],
         get_lazer_access_token(),
         1,
-    );
+        ExponentialBackoffBuilder::default()
+            .with_max_elapsed_time(None) // max_elapsed_time is not supported in Pyth Lazer client
+            .build(),
+    )?;
 
     let stream = client.start().await?;
     pin!(stream);

lazer/sdk/rust/client/src/client.rs

@@ -2,7 +2,8 @@ use crate::{
     resilient_ws_connection::PythLazerResilientWSConnection, ws_connection::AnyResponse,
     CHANNEL_CAPACITY,
 };
-use anyhow::Result;
+use anyhow::{bail, Result};
+use backoff::ExponentialBackoff;
 use futures_util::stream;
 use pyth_lazer_protocol::subscription::{SubscribeRequest, SubscriptionId};
 use tokio::sync::mpsc::{self, error::TrySendError};
@@ -15,6 +16,7 @@ pub struct PythLazerClient {
     num_connections: usize,
     ws_connections: Vec<PythLazerResilientWSConnection>,
     receivers: Vec<mpsc::Receiver<AnyResponse>>,
+    backoff: ExponentialBackoff,
 }
 
 impl PythLazerClient {
@@ -24,14 +26,23 @@ impl PythLazerClient {
     /// * `endpoints` - A vector of endpoint URLs
     /// * `access_token` - The access token for authentication
     /// * `num_connections` - The number of WebSocket connections to maintain
-    pub fn new(endpoints: Vec<String>, access_token: String, num_connections: usize) -> Self {
-        Self {
+    pub fn new(
+        endpoints: Vec<String>,
+        access_token: String,
+        num_connections: usize,
+        backoff: ExponentialBackoff,
+    ) -> Result<Self> {
+        if backoff.max_elapsed_time.is_some() {
+            bail!("max_elapsed_time is not supported in Pyth Lazer client");
+        }
+        Ok(Self {
             endpoints,
             access_token,
             num_connections,
             ws_connections: Vec::with_capacity(num_connections),
             receivers: Vec::with_capacity(num_connections),
-        }
+            backoff,
+        })
     }
 
     pub async fn start(&mut self) -> Result<mpsc::Receiver<AnyResponse>> {
@@ -43,6 +54,7 @@ impl PythLazerClient {
             let connection = PythLazerResilientWSConnection::new(
                 endpoint,
                 self.access_token.clone(),
+                self.backoff.clone(),
                 sender.clone(),
             );
             self.ws_connections.push(connection);

lazer/sdk/rust/client/src/resilient_ws_connection.rs

@@ -1,6 +1,4 @@
-use backoff::{
-    backoff::Backoff, exponential::ExponentialBackoff, ExponentialBackoffBuilder, SystemClock,
-};
+use backoff::{backoff::Backoff, ExponentialBackoff};
 use futures_util::StreamExt;
 use pyth_lazer_protocol::subscription::{
     Request, SubscribeRequest, SubscriptionId, UnsubscribeRequest,
@@ -28,9 +26,14 @@ impl PythLazerResilientWSConnection {
     ///
     /// # Returns
     /// Returns a new client instance (not yet connected)
-    pub fn new(endpoint: String, access_token: String, sender: mpsc::Sender<AnyResponse>) -> Self {
+    pub fn new(
+        endpoint: String,
+        access_token: String,
+        backoff: ExponentialBackoff,
+        sender: mpsc::Sender<AnyResponse>,
+    ) -> Self {
         let (request_sender, mut request_receiver) = mpsc::channel(CHANNEL_CAPACITY);
-        let mut task = PythLazerResilientWSConnectionTask::new(endpoint, access_token);
+        let mut task = PythLazerResilientWSConnectionTask::new(endpoint, access_token, backoff);
 
         tokio::spawn(async move {
             if let Err(e) = task.run(sender, &mut request_receiver).await {
@@ -62,17 +65,16 @@ struct PythLazerResilientWSConnectionTask {
     endpoint: String,
     access_token: String,
     subscriptions: Vec<SubscribeRequest>,
-    backoff: ExponentialBackoff<SystemClock>,
+    backoff: ExponentialBackoff,
 }
 
 impl PythLazerResilientWSConnectionTask {
-    pub fn new(endpoint: String, access_token: String) -> Self {
+    pub fn new(endpoint: String, access_token: String, backoff: ExponentialBackoff) -> Self {
         Self {
             endpoint,
             access_token,
             subscriptions: Vec::new(),
-            // TODO: make backoff configurable
-            backoff: ExponentialBackoffBuilder::new().build(),
+            backoff,
         }
     }