feat(pyth-lazer-agent) Allow deduplicating updates within each batch

Author: bplatak

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "pyth-lazer-agent"
-version = "0.4.0"
+version = "0.4.1"
 edition = "2024"
 description = "Pyth Lazer Agent"
 license = "Apache-2.0"

apps/pyth-lazer-agent/Cargo.toml

@@ -49,10 +49,12 @@ publish_keypair_path = "/path/to/keypair.json"
 authorization_token = "your_token"
 listen_address = "0.0.0.0:8910"
 publish_interval_duration = "25ms"
+enable_update_deduplication = false
 ```
 
 - `relayers_urls`: The Lazer team will provide these.
 - `publish_keypair_path`: The keypair file generated with `solana-keygen` or similar.
 - `authorization_token`: The Lazer team will provide this or instruct that it can be omitted.
 - `listen_address`: The local port the agent will be listening on; can be anything you want.
 - `publisher_interval`: The agent will batch and send transaction bundles at this interval. The Lazer team will provide guidance here.
+- `enable_update_deduplication`: The agent will deduplicate updates based inside each batch before sending it to Lazer.

apps/pyth-lazer-agent/README.md

@@ -1,5 +1,5 @@
-relayer_urls = ["wss://relayer.pyth-lazer-staging.dourolabs.app/v1/transaction", "wss://relayer-1.pyth-lazer-staging.dourolabs.app/v1/transaction"]
-publish_keypair_path = "/path/to/solana/id.json"
+relayer_urls = ["ws://localhost:10001/v1/transaction"]
+publish_keypair_path = "/tmp/keypair.json"
 listen_address = "0.0.0.0:8910"
 publish_interval_duration = "25ms"
 authorization_token="token1"

apps/pyth-lazer-agent/config/config.toml

@@ -19,6 +19,8 @@ pub struct Config {
     #[serde(with = "humantime_serde", default = "default_publish_interval")]
     pub publish_interval_duration: Duration,
     pub history_service_url: Option<Url>,
+    #[serde(default)]
+    pub enable_update_deduplication: bool
 }
 
 #[derive(Deserialize, Derivative, Clone, PartialEq)]

apps/pyth-lazer-agent/src/config.rs

@@ -299,6 +299,7 @@ pub mod tests {
             publish_keypair_path: Default::default(),
             publish_interval_duration: Default::default(),
             history_service_url: None,
+            enable_update_deduplication: false,
         };
 
         println!("{:?}", get_metadata(config).await.unwrap());

apps/pyth-lazer-agent/src/jrpc_handle.rs

@@ -1,3 +1,4 @@
+use std::collections::HashMap;
 use crate::config::{CHANNEL_CAPACITY, Config};
 use crate::relayer_session::RelayerSessionTask;
 use anyhow::{Context, Result, bail};
@@ -132,8 +133,14 @@ impl LazerPublisherTask {
             return Ok(());
         }
 
+        let updates = if self.config.enable_update_deduplication {
+            deduplicate_feed_updates(&self.pending_updates.drain(..).collect())?
+        } else {
+            self.pending_updates.drain(..).collect()
+        };
+
         let publisher_update = PublisherUpdate {
-            updates: self.pending_updates.drain(..).collect(),
+            updates,
             publisher_timestamp: MessageField::some(Timestamp::now()),
             special_fields: Default::default(),
         };
@@ -173,10 +180,33 @@ impl LazerPublisherTask {
     }
 }
 
+fn deduplicate_feed_updates(feed_updates: &Vec<FeedUpdate>) -> Result<Vec<FeedUpdate>> {
+    let mut deduped_feed_updates = Vec::new();
+    let mut last_feed_update = HashMap::new();
+
+    // assume that feed_updates is already sorted by ts (within feed_update_id groups)
+    for feed_update in feed_updates {
+        let feed_id = feed_update.feed_id.context("feed_id is required")?;
+
+        if let Some(update) = feed_update.update.as_ref() {
+            if let Some(last_update) = last_feed_update.get(&feed_id) {
+                if update == last_update {
+                    continue;
+                }
+            }
+
+            deduped_feed_updates.push(feed_update.clone());
+            last_feed_update.insert(feed_id, update.clone());
+        }
+    }
+
+    Ok(deduped_feed_updates)
+}
+
 #[cfg(test)]
 mod tests {
     use crate::config::{CHANNEL_CAPACITY, Config};
-    use crate::lazer_publisher::LazerPublisherTask;
+    use crate::lazer_publisher::{deduplicate_feed_updates, LazerPublisherTask};
     use ed25519_dalek::SigningKey;
     use protobuf::well_known_types::timestamp::Timestamp;
     use protobuf::{Message, MessageField};
@@ -186,6 +216,7 @@ mod tests {
     use std::io::Write;
     use std::path::PathBuf;
     use std::time::Duration;
+    use pyth_lazer_protocol::time::TimestampUs;
     use tempfile::NamedTempFile;
     use tokio::sync::broadcast::error::TryRecvError;
     use tokio::sync::{broadcast, mpsc};
@@ -212,6 +243,18 @@ mod tests {
         temp_file
     }
 
+    fn test_feed_update(feed_id: u32, timestamp: TimestampUs, price: i64) -> FeedUpdate {
+        FeedUpdate {
+            feed_id: Some(feed_id),
+            source_timestamp: MessageField::some(timestamp.into()),
+            update: Some(Update::PriceUpdate(PriceUpdate {
+                price: Some(price),
+                ..PriceUpdate::default()
+            })),
+            special_fields: Default::default(),
+        }
+    }
+
     #[tokio::test]
     async fn test_lazer_exporter_task() {
         let signing_key_file = get_private_key_file();
@@ -224,6 +267,7 @@ mod tests {
             publish_keypair_path: PathBuf::from(signing_key_file.path()),
             publish_interval_duration: Duration::from_millis(25),
             history_service_url: None,
+            enable_update_deduplication: false,
         };
 
         let (relayer_sender, mut relayer_receiver) = broadcast::channel(CHANNEL_CAPACITY);
@@ -274,4 +318,53 @@ mod tests {
             _ => panic!("channel should have a transaction waiting"),
         }
     }
+
+    #[test]
+    fn test_deduplicate_feed_updates() {
+        // let's consider a batch containing updates for a single feed. the updates are (ts, price):
+        //   - (1, 10)
+        //   - (2, 10)
+        //   - (3, 10)
+        //   - (4, 15)
+        //   - (5, 15)
+        //   - (6, 10)
+        // we should only return (1, 10), (4, 15), (6, 10)
+
+        let updates = vec![
+            test_feed_update(1, TimestampUs::from_millis(1).unwrap(), 10),
+            test_feed_update(1, TimestampUs::from_millis(2).unwrap(), 10),
+            test_feed_update(1, TimestampUs::from_millis(3).unwrap(), 10),
+            test_feed_update(1, TimestampUs::from_millis(4).unwrap(), 15),
+            test_feed_update(1, TimestampUs::from_millis(5).unwrap(), 15),
+            test_feed_update(1, TimestampUs::from_millis(6).unwrap(), 10),
+        ];
+
+        let expected_updates = vec![
+            test_feed_update(1, TimestampUs::from_millis(1).unwrap(), 10),
+            test_feed_update(1, TimestampUs::from_millis(4).unwrap(), 15),
+            test_feed_update(1, TimestampUs::from_millis(6).unwrap(), 10),
+        ];
+
+        assert_eq!(deduplicate_feed_updates(&updates).unwrap(), expected_updates);
+    }
+
+    #[test]
+    fn test_deduplicate_feed_updates_multiple_feeds() {
+        let updates = vec![
+            test_feed_update(1, TimestampUs::from_millis(1).unwrap(), 10),
+            test_feed_update(1, TimestampUs::from_millis(2).unwrap(), 10),
+            test_feed_update(1, TimestampUs::from_millis(3).unwrap(), 10),
+            test_feed_update(2, TimestampUs::from_millis(4).unwrap(), 15),
+            test_feed_update(2, TimestampUs::from_millis(5).unwrap(), 15),
+            test_feed_update(2, TimestampUs::from_millis(6).unwrap(), 10),
+        ];
+
+        let expected_updates = vec![
+            test_feed_update(1, TimestampUs::from_millis(1).unwrap(), 10),
+            test_feed_update(2, TimestampUs::from_millis(4).unwrap(), 15),
+            test_feed_update(2, TimestampUs::from_millis(6).unwrap(), 10),
+        ];
+
+        assert_eq!(deduplicate_feed_updates(&updates).unwrap(), expected_updates);
+    }
 }