feat(lazer): improve protocol types and modules (#2934)

* refactor(lazer): add some Price tests and error types

* refactor(lazer): make Price operations more explicit

* refactor(lazer): move Price and Rate out from router module, update Rate
API and add tests

* refactor(lazer): restructure protocol modules

* fix(lazer): use non-inverted exponent values in Price and Rate

* test(lazer): add more Price and Rate tests

Author: Riateche

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "pyth-lazer-solana-contract"
-version = "0.5.0"
+version = "0.6.0"
 edition = "2021"
 description = "Pyth Lazer Solana contract and SDK."
 license = "Apache-2.0"
@@ -19,7 +19,7 @@ no-log-ix-name = []
 idl-build = ["anchor-lang/idl-build"]
 
 [dependencies]
-pyth-lazer-protocol = { path = "../../../../sdk/rust/protocol", version = "0.10.1" }
+pyth-lazer-protocol = { path = "../../../../sdk/rust/protocol", version = "0.11.0" }
 
 anchor-lang = "0.31.1"
 bytemuck = { version = "1.20.0", features = ["derive"] }

lazer/contracts/solana/programs/pyth-lazer-solana-contract/Cargo.toml

@@ -1,13 +1,13 @@
 [package]
 name = "pyth-lazer-publisher-sdk"
-version = "0.5.0"
+version = "0.6.0"
 edition = "2021"
 description = "Pyth Lazer Publisher SDK types."
 license = "Apache-2.0"
 repository = "https://github.com/pyth-network/pyth-crosschain"
 
 [dependencies]
-pyth-lazer-protocol = { version = "0.10.2", path = "../../sdk/rust/protocol" }
+pyth-lazer-protocol = { version = "0.11.0", path = "../../sdk/rust/protocol" }
 anyhow = "1.0.98"
 protobuf = "3.7.2"
 serde_json = "1.0.140"

lazer/publisher_sdk/rust/Cargo.toml

@@ -3,8 +3,8 @@ use crate::publisher_update::{FeedUpdate, FundingRateUpdate, PriceUpdate};
 use crate::state::FeedState;
 use ::protobuf::MessageField;
 use pyth_lazer_protocol::jrpc::{FeedUpdateParams, UpdateParams};
-use pyth_lazer_protocol::symbol_state::SymbolState;
 use pyth_lazer_protocol::FeedKind;
+use pyth_lazer_protocol::SymbolState;
 
 pub mod transaction_envelope {
     pub use crate::protobuf::transaction_envelope::*;
@@ -56,18 +56,18 @@ impl From<UpdateParams> for Update {
                 best_bid_price,
                 best_ask_price,
             } => Update::PriceUpdate(PriceUpdate {
-                price: Some(price.0.into()),
-                best_bid_price: best_bid_price.map(|p| p.0.into()),
-                best_ask_price: best_ask_price.map(|p| p.0.into()),
+                price: Some(price.mantissa_i64()),
+                best_bid_price: best_bid_price.map(|p| p.mantissa_i64()),
+                best_ask_price: best_ask_price.map(|p| p.mantissa_i64()),
                 special_fields: Default::default(),
             }),
             UpdateParams::FundingRateUpdate {
                 price,
                 rate,
                 funding_rate_interval,
             } => Update::FundingRateUpdate(FundingRateUpdate {
-                price: price.map(|p| p.0.into()),
-                rate: Some(rate.0),
+                price: price.map(|p| p.mantissa_i64()),
+                rate: Some(rate.mantissa()),
                 funding_rate_interval: MessageField::from_option(
                     funding_rate_interval.map(|i| i.into()),
                 ),

lazer/publisher_sdk/rust/src/lib.rs

@@ -1,12 +1,12 @@
 [package]
 name = "pyth-lazer-client"
-version = "2.0.1"
+version = "3.0.0"
 edition = "2021"
 description = "A Rust client for Pyth Lazer"
 license = "Apache-2.0"
 
 [dependencies]
-pyth-lazer-protocol = { path = "../protocol", version = "0.10.2" }
+pyth-lazer-protocol = { path = "../protocol", version = "0.11.0" }
 tokio = { version = "1", features = ["full"] }
 tokio-tungstenite = { version = "0.20", features = ["native-tls"] }
 futures-util = "0.3"

lazer/sdk/rust/client/Cargo.toml

@@ -4,15 +4,16 @@ use base64::Engine;
 use pyth_lazer_client::backoff::PythLazerExponentialBackoffBuilder;
 use pyth_lazer_client::client::PythLazerClientBuilder;
 use pyth_lazer_client::ws_connection::AnyResponse;
+use pyth_lazer_protocol::api::{
+    Channel, DeliveryFormat, Format, JsonBinaryEncoding, SubscriptionParams, SubscriptionParamsRepr,
+};
+use pyth_lazer_protocol::api::{SubscribeRequest, SubscriptionId, WsResponse};
 use pyth_lazer_protocol::message::{
     EvmMessage, LeEcdsaMessage, LeUnsignedMessage, Message, SolanaMessage,
 };
 use pyth_lazer_protocol::payload::PayloadData;
-use pyth_lazer_protocol::router::{
-    Channel, DeliveryFormat, FixedRate, Format, JsonBinaryEncoding, PriceFeedId, PriceFeedProperty,
-    SubscriptionParams, SubscriptionParamsRepr,
-};
-use pyth_lazer_protocol::subscription::{Response, SubscribeRequest, SubscriptionId};
+use pyth_lazer_protocol::time::FixedRate;
+use pyth_lazer_protocol::{PriceFeedId, PriceFeedProperty};
 use tokio::pin;
 use tracing::level_filters::LevelFilter;
 use tracing_subscriber::EnvFilter;
@@ -109,7 +110,7 @@ async fn main() -> anyhow::Result<()> {
         // The stream gives us base64-encoded binary messages. We need to decode, parse, and verify them.
         match msg {
             AnyResponse::Json(msg) => match msg {
-                Response::StreamUpdated(update) => {
+                WsResponse::StreamUpdated(update) => {
                     println!("Received a JSON update for {:?}", update.subscription_id);
                     if let Some(evm_data) = update.payload.evm {
                         // Decode binary data

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

@@ -51,7 +51,7 @@ use crate::{
 };
 use anyhow::{bail, Result};
 use backoff::ExponentialBackoff;
-use pyth_lazer_protocol::subscription::{SubscribeRequest, SubscriptionId};
+use pyth_lazer_protocol::api::{SubscribeRequest, SubscriptionId};
 use tokio::sync::mpsc::{self, error::TrySendError};
 use tracing::{error, warn};
 use ttl_cache::TtlCache;

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

@@ -2,9 +2,7 @@ use std::time::Duration;
 
 use backoff::{backoff::Backoff, ExponentialBackoff};
 use futures_util::StreamExt;
-use pyth_lazer_protocol::subscription::{
-    Request, SubscribeRequest, SubscriptionId, UnsubscribeRequest,
-};
+use pyth_lazer_protocol::api::{SubscribeRequest, SubscriptionId, UnsubscribeRequest, WsRequest};
 use tokio::{pin, select, sync::mpsc, time::Instant};
 use tracing::{error, info, warn};
 use url::Url;
@@ -18,7 +16,7 @@ use anyhow::{bail, Context, Result};
 const BACKOFF_RESET_DURATION: Duration = Duration::from_secs(10);
 
 pub struct PythLazerResilientWSConnection {
-    request_sender: mpsc::Sender<Request>,
+    request_sender: mpsc::Sender<WsRequest>,
 }
 
 impl PythLazerResilientWSConnection {
@@ -53,15 +51,17 @@ impl PythLazerResilientWSConnection {
 
     pub async fn subscribe(&mut self, request: SubscribeRequest) -> Result<()> {
         self.request_sender
-            .send(Request::Subscribe(request))
+            .send(WsRequest::Subscribe(request))
             .await
             .context("Failed to send subscribe request")?;
         Ok(())
     }
 
     pub async fn unsubscribe(&mut self, subscription_id: SubscriptionId) -> Result<()> {
         self.request_sender
-            .send(Request::Unsubscribe(UnsubscribeRequest { subscription_id }))
+            .send(WsRequest::Unsubscribe(UnsubscribeRequest {
+                subscription_id,
+            }))
             .await
             .context("Failed to send unsubscribe request")?;
         Ok(())
@@ -95,7 +95,7 @@ impl PythLazerResilientWSConnectionTask {
     pub async fn run(
         &mut self,
         response_sender: mpsc::Sender<AnyResponse>,
-        request_receiver: &mut mpsc::Receiver<Request>,
+        request_receiver: &mut mpsc::Receiver<WsRequest>,
     ) -> Result<()> {
         loop {
             let start_time = Instant::now();
@@ -128,7 +128,7 @@ impl PythLazerResilientWSConnectionTask {
     pub async fn start(
         &mut self,
         sender: mpsc::Sender<AnyResponse>,
-        request_receiver: &mut mpsc::Receiver<Request>,
+        request_receiver: &mut mpsc::Receiver<WsRequest>,
     ) -> Result<()> {
         let mut ws_connection =
             PythLazerWSConnection::new(self.endpoint.clone(), self.access_token.clone())?;
@@ -137,7 +137,7 @@ impl PythLazerResilientWSConnectionTask {
 
         for subscription in self.subscriptions.clone() {
             ws_connection
-                .send_request(Request::Subscribe(subscription))
+                .send_request(WsRequest::Subscribe(subscription))
                 .await?;
         }
         loop {
@@ -167,10 +167,10 @@ impl PythLazerResilientWSConnectionTask {
                 }
                 Some(request) = request_receiver.recv() => {
                    match request {
-                        Request::Subscribe(request) => {
+                        WsRequest::Subscribe(request) => {
                             self.subscribe(&mut ws_connection, request).await?;
                         }
-                        Request::Unsubscribe(request) => {
+                        WsRequest::Unsubscribe(request) => {
                             self.unsubscribe(&mut ws_connection, request).await?;
                         }
                    }

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

@@ -4,8 +4,8 @@ use anyhow::Result;
 use derive_more::From;
 use futures_util::{SinkExt, StreamExt, TryStreamExt};
 use pyth_lazer_protocol::{
+    api::{ErrorResponse, SubscribeRequest, UnsubscribeRequest, WsRequest, WsResponse},
     binary_update::BinaryWsUpdate,
-    subscription::{ErrorResponse, Request, Response, SubscribeRequest, UnsubscribeRequest},
 };
 use tokio_tungstenite::{connect_async, tungstenite::Message};
 use url::Url;
@@ -32,7 +32,7 @@ pub struct PythLazerWSConnection {
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash, From)]
 pub enum AnyResponse {
-    Json(Response),
+    Json(WsResponse),
     Binary(BinaryWsUpdate),
 }
 
@@ -84,13 +84,13 @@ impl PythLazerWSConnection {
                 .try_filter_map(|msg| async {
                     let r: Result<Option<AnyResponse>> = match msg {
                         Message::Text(text) => {
-                            Ok(Some(serde_json::from_str::<Response>(&text)?.into()))
+                            Ok(Some(serde_json::from_str::<WsResponse>(&text)?.into()))
                         }
                         Message::Binary(data) => {
                             Ok(Some(BinaryWsUpdate::deserialize_slice(&data)?.into()))
                         }
                         Message::Close(_) => Ok(Some(
-                            Response::Error(ErrorResponse {
+                            WsResponse::Error(ErrorResponse {
                                 error: "WebSocket connection closed".to_string(),
                             })
                             .into(),
@@ -103,7 +103,7 @@ impl PythLazerWSConnection {
         Ok(response_stream)
     }
 
-    pub async fn send_request(&mut self, request: Request) -> Result<()> {
+    pub async fn send_request(&mut self, request: WsRequest) -> Result<()> {
         if let Some(sender) = &mut self.ws_sender {
             let msg = serde_json::to_string(&request)?;
             sender.send(Message::Text(msg)).await?;
@@ -118,7 +118,7 @@ impl PythLazerWSConnection {
     /// # Arguments
     /// * `request` - A subscription request containing feed IDs and parameters
     pub async fn subscribe(&mut self, request: SubscribeRequest) -> Result<()> {
-        let request = Request::Subscribe(request);
+        let request = WsRequest::Subscribe(request);
         self.send_request(request).await
     }
 
@@ -127,7 +127,7 @@ impl PythLazerWSConnection {
     /// # Arguments
     /// * `subscription_id` - The ID of the subscription to cancel
     pub async fn unsubscribe(&mut self, request: UnsubscribeRequest) -> Result<()> {
-        let request = Request::Unsubscribe(request);
+        let request = WsRequest::Unsubscribe(request);
         self.send_request(request).await
     }
 

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

@@ -1,6 +1,6 @@
 [package]
 name = "pyth-lazer-protocol"
-version = "0.10.2"
+version = "0.11.0"
 edition = "2021"
 description = "Pyth Lazer SDK - protocol types."
 license = "Apache-2.0"
@@ -20,6 +20,7 @@ mry = { version = "0.13.0", features = ["serde"], optional = true }
 chrono = "0.4.41"
 humantime = "2.2.0"
 hex = "0.4.3"
+thiserror = "2.0.12"
 
 [dev-dependencies]
 bincode = "1.3.3"
@@ -28,3 +29,4 @@ hex = "0.4.3"
 libsecp256k1 = "0.7.1"
 bs58 = "0.5.1"
 alloy-primitives = "0.8.19"
+assert_float_eq = "1.1.4"