feat: timings

Author: alexcambose

src/bin/pyth_reader.rs

@@ -126,6 +126,7 @@ async fn fetch_price_updates(jetstream: jetstream::Context, config: &AppConfig)
         .await?;
 
     let mut update_count = 0;
+    let mut duration_count = 0;
     let mut unique_price_feeds = HashSet::new();
     let mut last_report_time = Instant::now();
     let jetstream_clone = jetstream.clone();
@@ -158,7 +159,6 @@ async fn fetch_price_updates(jetstream: jetstream::Context, config: &AppConfig)
             msg_id_counter += 1;
             let mut headers = HeaderMap::new();
             headers.insert("Nats-Msg-Id", msg_id.as_str());
-            info!("body: {:#?},msg_id: {}", body.len(), msg_id);
             if let Err(e) = jetstream_clone
                 .publish_with_headers("pyth.publisher.updates", headers, body.into())
                 .await
@@ -172,6 +172,8 @@ async fn fetch_price_updates(jetstream: jetstream::Context, config: &AppConfig)
 
     while let Some(update) = notif.next().await {
         debug!("Received price update");
+        let start_time = Instant::now();
+
         let account: Account = match update.value.account.decode() {
             Some(account) => account,
             _none => {
@@ -248,11 +250,7 @@ async fn fetch_price_updates(jetstream: jetstream::Context, config: &AppConfig)
                     Err(e) => warn!("Failed to publish price update to JetStream: {}", e),
                 }
             });
-            info!(
-                "update: {}, comp: {}",
-                update.context.slot,
-                price_account.comp.len()
-            );
+
             for component in price_account.comp {
                 let publisher = component.publisher.to_string();
                 let publisher_price_update = PublisherPriceUpdate {
@@ -261,28 +259,35 @@ async fn fetch_price_updates(jetstream: jetstream::Context, config: &AppConfig)
                     price: price_account.agg.price.to_string(),
                     slot: update.context.slot, // Add this field
                 };
-                if publisher_price_update.feed_id == "7jAVut34sgRj6erznsYvLYvjc9GJwXTpN88ThZSDJ65G"
-                    && publisher == "6DNocjFJjocPLZnKBZyEJAC5o2QaiT5Mx8AkphfxDm5i"
-                {
-                    info!("publisher_price_update: {:#?}", publisher_price_update);
-                }
+
                 let key = (publisher_price_update.feed_id.clone(), publisher);
                 let mut buf = publisher_buffer.lock().await;
                 buf.insert(key, publisher_price_update);
             }
+
+            let end_time = Instant::now();
+            let duration = end_time.duration_since(start_time);
+            duration_count += duration.as_micros();
             update_count += 1;
             unique_price_feeds.insert(price_account.prod);
         } else {
             debug!("Skipping price update due to invalid status or slot difference");
         }
 
         // Report aggregate information every minute and emit metrics
-        if last_report_time.elapsed() >= Duration::from_secs(60) {
+        if last_report_time.elapsed() >= Duration::from_secs(1) {
             info!(
-                "Processed {} updates from {} unique price feeds in the last minute",
+                "Processed {} updates from {} unique price feeds in the last 1 secs",
                 update_count,
                 unique_price_feeds.len()
             );
+            info!(
+                "last slot: {}, comp: {}",
+                update.context.slot,
+                price_account.comp.len()
+            );
+            info!("Average duration: {:?}", duration_count / update_count);
+            duration_count = 0;
             update_count = 0;
             unique_price_feeds.clear();
             last_report_time = Instant::now();