feat: bincode everything

Author: alexcambose

src/bin/pyth_reader.rs

@@ -30,7 +30,7 @@ use tracing::{debug, error, info, warn};
 use tracing_subscriber::{fmt, EnvFilter};
 use url::Url;
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, bincode::Encode, bincode::Decode)]
 struct PriceUpdate {
     #[serde(rename = "type")]
     update_type: String,
@@ -48,14 +48,14 @@ struct PublisherPriceUpdate {
 type PublisherKey = (String, String); // (feed_id, publisher)
 type PublisherBuffer = HashMap<PublisherKey, PublisherPriceUpdate>;
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, bincode::Encode, bincode::Decode)]
 struct PriceFeed {
     id: String,
     price: PriceInfo,
     ema_price: PriceInfo,
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, bincode::Encode, bincode::Decode)]
 struct PriceInfo {
     price: String,
     conf: String,
@@ -152,8 +152,8 @@ async fn publish_price_updates(
             },
         },
     };
-    let price_update_message = serde_json::to_string(&price_update).unwrap();
-
+    let price_update_message: Vec<u8> =
+        bincode::encode_to_vec(&price_update, bincode::config::standard()).unwrap();
     // Create a unique message ID
     let price_update_message_id = format!(
         "{}:{}",
@@ -213,11 +213,6 @@ async fn fetch_price_updates(jetstream: jetstream::Context, config: &AppConfig)
     info!("Starting Pyth reader");
     let mut publisher_buffer: PublisherBuffer = HashMap::new();
     let client = PubsubClient::new(config.pyth.websocket_addr.as_str()).await?;
-    info!(
-        "Connected to Pyth WebSocket at {}",
-        config.pyth.websocket_addr
-    );
-
     let rpc_config = RpcProgramAccountsConfig {
         account_config: RpcAccountInfoConfig {
             commitment: Some(CommitmentConfig::confirmed()),
@@ -230,63 +225,48 @@ async fn fetch_price_updates(jetstream: jetstream::Context, config: &AppConfig)
     };
 
     let (mut notif, _unsub) = client
-        .program_subscribe(&config.pyth.program_key, Some(rpc_config))
+        .program_subscribe(&config.pyth.program_key, Some(rpc_config.clone()))
         .await?;
 
-    let update_count = Arc::new(AtomicUsize::new(0));
-    let duration_count = Arc::new(AtomicUsize::new(0));
+    info!(
+        "Connected to Pyth WebSocket at {}",
+        config.pyth.websocket_addr
+    );
+
+    let mut update_count = 0;
+    let mut duration_count = 0;
     let jetstream_clone = jetstream.clone();
     let mut msg_id_counter = 0;
 
-    let update_count_clone = update_count.clone();
-    let duration_count_clone = duration_count.clone();
-    // tokio::spawn(async move {
-    //     loop {
-    //         tokio::time::sleep(Duration::from_secs(1)).await;
-    //         info!(
-    //             "Total duration: {:?}, Update Count: {:?}",
-    //             duration_count_clone.load(Ordering::Relaxed) as u128,
-    //             update_count_clone.load(Ordering::Relaxed),
-    //         );
-    //         if update_count_clone.load(Ordering::Relaxed) > 0 {
-    //             info!(
-    //                 "Average duration: {:?}",
-    //                 duration_count_clone.load(Ordering::Relaxed) as u128
-    //                     / update_count_clone.load(Ordering::Relaxed) as u128
-    //             );
-    //         }
-    //         duration_count_clone.store(0, Ordering::Relaxed);
-    //         update_count_clone.store(0, Ordering::Relaxed);
-    //     }
-    // });
-
-    let mut interval = tokio::time::interval(Duration::from_millis(100));
+    let mut interval = tokio::time::interval(Duration::from_millis(50));
     let mut last_seen_slot = 0;
 
     loop {
         tokio::select! {
             _ = interval.tick() => {
                 let instant = Instant::now();
-
                 let updates: Vec<PublisherPriceUpdate> = {
                     if publisher_buffer.is_empty() {
                         continue;
                     }
                     publisher_buffer.drain().map(|(_, v)| v).collect()
                 };
-                info!("Drained publisher buffer, size: {}, elapsed time: {:?}", updates.len(), instant.elapsed());
 
                 // Serialize as JSON array
                 let body: Vec<u8> = bincode::encode_to_vec(&updates, bincode::config::standard()).unwrap();
-
-                info!("Serialized publisher updates, size: {}, elapsed time: {:?}", body.len(), instant.elapsed());
-
                 // Use a random ID as Nats-Msg-Id for the batch
                 let msg_id = format!("publisher_batch:{}", msg_id_counter);
                 msg_id_counter += 1;
                 let mut headers = HeaderMap::new();
                 headers.insert("Nats-Msg-Id", msg_id.as_str());
-                info!("Publishing {} publisher updates in a batch, total size {}, elapsed time: {:?}", updates.len(), body.len(), instant.elapsed());
+                // info!("Serialized publisher updates, size: {}, elapsed time: {:?}, publisher buffer size: {:?}", body.len(), instant.elapsed(), publisher_buffer.len());
+                // info!("Publishing {} publisher updates in a batch, total size {}, elapsed time: {:?}", updates.len(), body.len(), instant.elapsed());
+                //  info!(
+                //     "Average duration: {:?}",
+                //     duration_count/ update_count
+                // );
+                duration_count = 0;
+                update_count = 0;
                 if let Err(e) = jetstream_clone
                     .publish_with_headers("pyth.publisher.updates", headers, body.into())
                     .await
@@ -295,20 +275,19 @@ async fn fetch_price_updates(jetstream: jetstream::Context, config: &AppConfig)
                 } else {
                     debug!("Published {} publisher updates in a batch", updates.len());
                 }
-
-                info!("Time taken to process publisher updates: {:?}", instant.elapsed());
             }
             maybe_update = notif.next() => {
+                let start_time = Instant::now();
+
                 let update = match maybe_update {
                     None => {
                         let error_msg = "Pythnet network listener stream ended unexpectedly";
                         error!("{}", error_msg);
-                        return Err(anyhow::anyhow!(error_msg));
+                        break Ok(());
                     },
                     Some(update) => update
                 };
 
-                let start_time = Instant::now();
                 debug!("Received price update");
                 let price_account: PythnetPriceAccount = match get_price_account_from_update(&update) {
                     Ok(account) => account,
@@ -329,27 +308,25 @@ async fn fetch_price_updates(jetstream: jetstream::Context, config: &AppConfig)
                     );
 
                     let jetstream_clone = jetstream.clone();
-                    let update_count_clone = update_count.clone();
-                    let duration_count_clone = duration_count.clone();
                     publish_price_updates(jetstream_clone, price_account, &update).await;
                     process_publisher_price_updates(price_account, &update, &mut publisher_buffer).await;
 
                     let end_time = Instant::now();
                     let duration = end_time.duration_since(start_time);
-                    update_count_clone.fetch_add(1, Ordering::Relaxed);
-                    duration_count_clone.fetch_add(duration.as_millis() as usize, Ordering::Relaxed);
+                    update_count += 1;
+                    duration_count += duration.as_micros();
                     if update.context.slot > last_seen_slot {
                         last_seen_slot = update.context.slot;
                         info!("Processing price update, slot: {}", update.context.slot);
                     }
+
                 } else {
                     debug!("Skipping price update due to invalid status or slot difference");
                 }
+
             }
         }
     }
-
-    // If we exit the loop, it means the stream has ended
 }
 
 fn load_config(args: &Args) -> Result<AppConfig> {

src/bin/websocket_server.rs

@@ -327,7 +327,7 @@ struct PriceUpdate {
     price_feed: PriceFeed,
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, bincode::Encode, bincode::Decode)]
 struct PublisherPriceUpdate {
     publisher: String,
     feed_id: String,
@@ -380,14 +380,15 @@ async fn handle_nats_publisher_updates_messages(
             match msg {
                 Ok(msg) => {
                     let updates: Vec<PublisherPriceUpdate> =
-                        match serde_json::from_slice(&msg.payload) {
-                            Ok(u) => u,
+                        match bincode::decode_from_slice(&msg.payload, bincode::config::standard())
+                        {
+                            Ok((updates, _)) => updates,
                             Err(e) => {
                                 warn!(error = %e, "Failed to parse publisher price update batch");
                                 continue;
                             }
                         };
-                    debug!("Parsed {} publisher updates in batch", updates.len());
+                    info!("Parsed {} publisher updates in batch", updates.len());
                     // Build per-client payloads while holding the lock,
                     // but DO NOT send while holding it.
                     let mut to_send: Vec<(String, mpsc::UnboundedSender<Message>, String)> =
@@ -403,9 +404,6 @@ async fn handle_nats_publisher_updates_messages(
                                     client_data.publisher_subscriptions.contains(&u.publisher)
                                 })
                                 .map(|u| {
-                                    if u.feed_id == "7jAVut34sgRj6erznsYvLYvjc9GJwXTpN88ThZSDJ65G" {
-                                        info!("publisher_price_update: {:#?}", u);
-                                    }
                                     json!({
                                         "publisher": u.publisher,
                                         "feed_id": u.feed_id,
@@ -438,7 +436,7 @@ async fn handle_nats_publisher_updates_messages(
                                 batch_json,
                             ));
                         }
-                    } // lock dropped here
+                    }
 
                     // Now send the prepared batches
                     for (client_addr, sender, batch_json) in to_send {