feat(lazer): add resilient client in rust (#2859)

* feat(lazer): add resilient client in rust

* configurable backoff

* add backoff reset

* impl dedup

* fix

* use Url

* asset non empty endponts in constructor

* configurable channel capacity

* use single channel for connections

* add expo backoff builder wrapper

* add timeout

* add pyth lazer client builder

* fix backoff reset logic

Author: keyvankhademi

Cargo.lock

@@ -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/Cargo.toml

@@ -1,6 +1,9 @@
+use std::time::Duration;
+
 use base64::Engine;
-use futures_util::StreamExt;
-use pyth_lazer_client::{AnyResponse, LazerClient};
+use pyth_lazer_client::backoff::PythLazerExponentialBackoffBuilder;
+use pyth_lazer_client::client::PythLazerClientBuilder;
+use pyth_lazer_client::ws_connection::AnyResponse;
 use pyth_lazer_protocol::message::{
     EvmMessage, LeEcdsaMessage, LeUnsignedMessage, Message, SolanaMessage,
 };
@@ -9,8 +12,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,11 +25,32 @@ 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 = PythLazerClientBuilder::new(get_lazer_access_token())
+        // Optionally override the default endpoints
+        .with_endpoints(vec![
+            "wss://pyth-lazer-0.dourolabs.app/v1/stream".parse()?,
+            "wss://pyth-lazer-1.dourolabs.app/v1/stream".parse()?,
+        ])
+        // Optionally set the number of connections
+        .with_num_connections(4)
+        // Optionally set the backoff strategy
+        .with_backoff(PythLazerExponentialBackoffBuilder::default().build())
+        // Optionally set the timeout for each connection
+        .with_timeout(Duration::from_secs(5))
+        // Optionally set the channel capacity for responses
+        .with_channel_capacity(1000)
+        .build()?;
+
     let stream = client.start().await?;
     pin!(stream);
 
@@ -72,16 +98,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 +215,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/examples/subscribe_price_feeds.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()
+    }
+}