feat(lazer): add resilient client in rust

lazer/sdk/rust/client/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pyth-lazer-client"
-version = "0.1.3"
+version = "1.0.0"
 edition = "2021"
 description = "A Rust client for Pyth Lazer"
 license = "Apache-2.0"
@@ -17,6 +17,9 @@ anyhow = "1.0"
 tracing = "0.1"
 url = "2.4"
 derive_more = { version = "1.0.0", features = ["from"] }
+backoff = { version = "0.4.0", features = ["futures", "tokio"] }
+ttl_cache = "0.5.1"
+
 
 [dev-dependencies]
 bincode = "1.3.3"
@@ -25,3 +28,4 @@ hex = "0.4.3"
 libsecp256k1 = "0.7.1"
 bs58 = "0.5.1"
 alloy-primitives = "0.8.19"
+tracing-subscriber = { version = "0.3.19", features = ["env-filter", "json"] }

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

@@ -1,6 +1,7 @@
 use base64::Engine;
-use futures_util::StreamExt;
-use pyth_lazer_client::{AnyResponse, LazerClient};
+use pyth_lazer_client::backoff::PythLazerExponentialBackoffBuilder;
+use pyth_lazer_client::client::PythLazerClient;
+use pyth_lazer_client::ws_connection::AnyResponse;
 use pyth_lazer_protocol::message::{
     EvmMessage, LeEcdsaMessage, LeUnsignedMessage, Message, SolanaMessage,
 };
@@ -9,8 +10,10 @@ use pyth_lazer_protocol::router::{
     Channel, DeliveryFormat, FixedRate, Format, JsonBinaryEncoding, PriceFeedId, PriceFeedProperty,
     SubscriptionParams, SubscriptionParamsRepr,
 };
-use pyth_lazer_protocol::subscription::{Request, Response, SubscribeRequest, SubscriptionId};
+use pyth_lazer_protocol::subscription::{Response, SubscribeRequest, SubscriptionId};
 use tokio::pin;
+use tracing::level_filters::LevelFilter;
+use tracing_subscriber::EnvFilter;
 
 fn get_lazer_access_token() -> String {
     // Place your access token in your env at LAZER_ACCESS_TOKEN or set it here
@@ -20,12 +23,31 @@ fn get_lazer_access_token() -> String {
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            EnvFilter::builder()
+                .with_default_directive(LevelFilter::INFO.into())
+                .from_env()?,
+        )
+        .json()
+        .init();
+
     // Create and start the client
-    let mut client = LazerClient::new(
-        "wss://pyth-lazer.dourolabs.app/v1/stream",
-        &get_lazer_access_token(),
+    let mut client = PythLazerClient::new(
+        vec![
+            "wss://pyth-lazer-0.dourolabs.app/v1/stream".parse()?,
+            "wss://pyth-lazer-1.dourolabs.app/v1/stream".parse()?,
+        ],
+        get_lazer_access_token(),
+        4,
+        PythLazerExponentialBackoffBuilder::default().build(),
     )?;
-    let stream = client.start().await?;
+
+    let stream = client
+        .start(
+            1000, // Use a channel capacity of 1000
+        )
+        .await?;
     pin!(stream);
 
     let subscription_requests = vec![
@@ -72,16 +94,16 @@ async fn main() -> anyhow::Result<()> {
     ];
 
     for req in subscription_requests {
-        client.subscribe(Request::Subscribe(req)).await?;
+        client.subscribe(req).await?;
     }
 
     println!("Subscribed to price feeds. Waiting for updates...");
 
     // Process the first few updates
     let mut count = 0;
-    while let Some(msg) = stream.next().await {
+    while let Some(msg) = stream.recv().await {
         // The stream gives us base64-encoded binary messages. We need to decode, parse, and verify them.
-        match msg? {
+        match msg {
             AnyResponse::Json(msg) => match msg {
                 Response::StreamUpdated(update) => {
                     println!("Received a JSON update for {:?}", update.subscription_id);
@@ -189,8 +211,6 @@ async fn main() -> anyhow::Result<()> {
         println!("Unsubscribed from {sub_id:?}");
     }
 
-    tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
-    client.close().await?;
     Ok(())
 }
 

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

@@ -0,0 +1,69 @@
+use std::time::Duration;
+
+use backoff::{
+    default::{INITIAL_INTERVAL_MILLIS, MAX_INTERVAL_MILLIS, MULTIPLIER, RANDOMIZATION_FACTOR},
+    ExponentialBackoff, ExponentialBackoffBuilder,
+};
+
+#[derive(Debug)]
+pub struct PythLazerExponentialBackoffBuilder {
+    initial_interval: Duration,
+    randomization_factor: f64,
+    multiplier: f64,
+    max_interval: Duration,
+}
+
+impl Default for PythLazerExponentialBackoffBuilder {
+    fn default() -> Self {
+        Self {
+            initial_interval: Duration::from_millis(INITIAL_INTERVAL_MILLIS),
+            randomization_factor: RANDOMIZATION_FACTOR,
+            multiplier: MULTIPLIER,
+            max_interval: Duration::from_millis(MAX_INTERVAL_MILLIS),
+        }
+    }
+}
+
+impl PythLazerExponentialBackoffBuilder {
+    pub fn new() -> Self {
+        Default::default()
+    }
+
+    /// The initial retry interval.
+    pub fn with_initial_interval(&mut self, initial_interval: Duration) -> &mut Self {
+        self.initial_interval = initial_interval;
+        self
+    }
+
+    /// The randomization factor to use for creating a range around the retry interval.
+    ///
+    /// A randomization factor of 0.5 results in a random period ranging between 50% below and 50%
+    /// above the retry interval.
+    pub fn with_randomization_factor(&mut self, randomization_factor: f64) -> &mut Self {
+        self.randomization_factor = randomization_factor;
+        self
+    }
+
+    /// The value to multiply the current interval with for each retry attempt.
+    pub fn with_multiplier(&mut self, multiplier: f64) -> &mut Self {
+        self.multiplier = multiplier;
+        self
+    }
+
+    /// The maximum value of the back off period. Once the retry interval reaches this
+    /// value it stops increasing.
+    pub fn with_max_interval(&mut self, max_interval: Duration) -> &mut Self {
+        self.max_interval = max_interval;
+        self
+    }
+
+    pub fn build(&self) -> ExponentialBackoff {
+        ExponentialBackoffBuilder::default()
+            .with_initial_interval(self.initial_interval)
+            .with_randomization_factor(self.randomization_factor)
+            .with_multiplier(self.multiplier)
+            .with_max_interval(self.max_interval)
+            .with_max_elapsed_time(None)
+            .build()
+    }
+}

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

@@ -0,0 +1,111 @@
+use std::time::Duration;
+
+use crate::{
+    resilient_ws_connection::PythLazerResilientWSConnection, ws_connection::AnyResponse,
+    CHANNEL_CAPACITY,
+};
+use anyhow::{bail, Result};
+use backoff::ExponentialBackoff;
+use pyth_lazer_protocol::subscription::{SubscribeRequest, SubscriptionId};
+use tokio::sync::mpsc::{self, error::TrySendError};
+use tracing::{error, warn};
+use ttl_cache::TtlCache;
+use url::Url;
+
+const DEDUP_CACHE_SIZE: usize = 100_000;
+const DEDUP_TTL: Duration = Duration::from_secs(10);
+
+pub struct PythLazerClient {
+    endpoints: Vec<Url>,
+    access_token: String,
+    num_connections: usize,
+    ws_connections: Vec<PythLazerResilientWSConnection>,
+    backoff: ExponentialBackoff,
+}
+
+impl PythLazerClient {
+    /// Creates a new client instance
+    ///
+    /// # Arguments
+    /// * `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<Url>,
+        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");
+        }
+        if endpoints.is_empty() {
+            bail!("At least one endpoint must be provided");
+        }
+        Ok(Self {
+            endpoints,
+            access_token,
+            num_connections,
+            ws_connections: Vec::with_capacity(num_connections),
+            backoff,
+        })
+    }
+
+    pub async fn start(&mut self, channel_capacity: usize) -> Result<mpsc::Receiver<AnyResponse>> {
+        let (sender, receiver) = mpsc::channel::<AnyResponse>(channel_capacity);
+        let (ws_connection_sender, mut ws_connection_receiver) =
+            mpsc::channel::<AnyResponse>(CHANNEL_CAPACITY);
+
+        for i in 0..self.num_connections {
+            let endpoint = self.endpoints[i % self.endpoints.len()].clone();
+            let connection = PythLazerResilientWSConnection::new(
+                endpoint,
+                self.access_token.clone(),
+                self.backoff.clone(),
+                ws_connection_sender.clone(),
+            );
+            self.ws_connections.push(connection);
+        }
+
+        let mut seen_updates = TtlCache::new(DEDUP_CACHE_SIZE);
+
+        tokio::spawn(async move {
+            while let Some(response) = ws_connection_receiver.recv().await {
+                let cache_key = response.cache_key();
+                if seen_updates.contains_key(&cache_key) {
+                    continue;
+                }
+                seen_updates.insert(cache_key, response.clone(), DEDUP_TTL);
+
+                match sender.try_send(response) {
+                    Ok(_) => (),
+                    Err(TrySendError::Full(r)) => {
+                        warn!("Sender channel is full, responses will be delayed");
+                        if sender.send(r).await.is_err() {
+                            error!("Sender channel is closed, stopping client");
+                        }
+                    }
+                    Err(TrySendError::Closed(_)) => {
+                        error!("Sender channel is closed, stopping client");
+                    }
+                }
+            }
+        });
+
+        Ok(receiver)
+    }
+
+    pub async fn subscribe(&mut self, subscribe_request: SubscribeRequest) -> Result<()> {
+        for connection in &mut self.ws_connections {
+            connection.subscribe(subscribe_request.clone()).await?;
+        }
+        Ok(())
+    }
+
+    pub async fn unsubscribe(&mut self, subscription_id: SubscriptionId) -> Result<()> {
+        for connection in &mut self.ws_connections {
+            connection.unsubscribe(subscription_id).await?;
+        }
+        Ok(())
+    }
+}