add support to configure max latency (#391)

* add initial code to support max latency

* fix precommit

* fix tests

* add more tests

* Refactor test_twap.rs to use random values for max_latency

* Refactor price aggregation test

* address comments

* bump

* add comment

* refactor

program/c/src/oracle/oracle.h

@@ -195,8 +195,9 @@ typedef struct pc_price
   pc_ema_t        twac_;              // time-weighted average conf interval
   int64_t         timestamp_;         // unix timestamp of aggregate price
   uint8_t         min_pub_;           // min publishers for valid price
-  int8_t          drv2_;              // space for future derived values
-  int16_t         drv3_;              // space for future derived values
+  int8_t          message_sent_;      // flag to indicate if the current aggregate price has been sent as a message to the message buffer, 0 if not sent, 1 if sent
+  uint8_t         max_latency_;       // configurable max latency in slots between send and receive
+  int8_t          drv3_;              // space for future derived values
   int32_t         drv4_;              // space for future derived values
   pc_pub_key_t    prod_;              // product id/ref-account
   pc_pub_key_t    next_;              // next price account in list

program/c/src/oracle/upd_aggregate.h

@@ -171,11 +171,13 @@ static inline bool upd_aggregate( pc_price_t *ptr, uint64_t slot, int64_t timest
       int64_t slot_diff = ( int64_t )slot - ( int64_t )( iptr->agg_.pub_slot_ );
       int64_t price     = iptr->agg_.price_;
       int64_t conf      = ( int64_t )( iptr->agg_.conf_ );
+      int64_t max_latency = ptr->max_latency_ ? ptr->max_latency_ : PC_MAX_SEND_LATENCY;
       if ( iptr->agg_.status_ == PC_STATUS_TRADING &&
            // No overflow for INT64_MIN+conf or INT64_MAX-conf as 0 < conf < INT64_MAX
            // These checks ensure that price - conf and price + conf do not overflow.
            (int64_t)0 < conf && (INT64_MIN + conf) <= price && price <= (INT64_MAX-conf) &&
-           slot_diff >= 0 && slot_diff <= PC_MAX_SEND_LATENCY ) {
+           // slot_diff is implicitly >= 0 due to the check in Rust code ensuring publishing_slot is always less than or equal to the current slot.
+           slot_diff <= max_latency ) {
         numv += 1;
         prcs[ nprcs++ ] = price - conf;
         prcs[ nprcs++ ] = price;

program/rust/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pyth-oracle"
-version = "2.25.0"
+version = "2.26.0"
 edition = "2021"
 license = "Apache 2.0"
 publish = false

program/rust/src/accounts/price.rs

@@ -65,7 +65,10 @@ mod price_pythnet {
         /// Minimum valid publisher quotes for a succesful aggregation
         pub min_pub_:           u8,
         pub message_sent_:      u8,
-        pub unused_2_:          i16,
+        /// Configurable max latency in slots between send and receive
+        pub max_latency_:       u8,
+        /// Unused placeholder for alignment
+        pub unused_2_:          i8,
         pub unused_3_:          i32,
         /// Corresponding product account
         pub product_account:    Pubkey,
@@ -116,6 +119,7 @@ mod price_pythnet {
                     self.agg_.price_,
                     self.agg_.conf_,
                     self.agg_.pub_slot_.saturating_sub(self.prev_slot_),
+                    self.max_latency_,
                 ); // pub_slot should always be >= prev_slot, but we protect ourselves against underflow just in case
                 Ok(())
             } else {
@@ -172,11 +176,17 @@ mod price_pythnet {
     }
 
     impl PriceCumulative {
-        pub fn update(&mut self, price: i64, conf: u64, slot_gap: u64) {
+        pub fn update(&mut self, price: i64, conf: u64, slot_gap: u64, max_latency: u8) {
             self.price += i128::from(price) * i128::from(slot_gap);
             self.conf += u128::from(conf) * u128::from(slot_gap);
+            // Use PC_MAX_SEND_LATENCY if max_latency is 0, otherwise use max_latency
+            let latency = if max_latency == 0 {
+                u64::from(PC_MAX_SEND_LATENCY)
+            } else {
+                u64::from(max_latency)
+            };
             // This is expected to saturate at 0 most of the time (while the feed is up).
-            self.num_down_slots += slot_gap.saturating_sub(PC_MAX_SEND_LATENCY.into());
+            self.num_down_slots += slot_gap.saturating_sub(latency);
         }
     }
 }
@@ -225,7 +235,10 @@ mod price_solana {
         /// Whether the current aggregate price has been sent as a message to the message buffer.
         /// 0 = false, 1 = true. (this is a u8 to make the Pod trait happy)
         pub message_sent_:      u8,
-        pub unused_2_:          i16,
+        /// Configurable max latency in slots between send and receive
+        pub max_latency_:       u8,
+        /// Unused placeholder for alignment
+        pub unused_2_:          i8,
         pub unused_3_:          i32,
         /// Corresponding product account
         pub product_account:    Pubkey,

program/rust/src/tests/test_twap.rs

@@ -21,18 +21,20 @@ use {
 
 #[derive(Clone, Debug, Copy)]
 pub struct DataEvent {
-    price:    i64,
-    conf:     u64,
-    slot_gap: u64,
+    price:       i64,
+    conf:        u64,
+    slot_gap:    u64,
+    max_latency: u8,
 }
 
 impl Arbitrary for DataEvent {
     fn arbitrary(g: &mut quickcheck::Gen) -> Self {
         DataEvent {
-            slot_gap: u64::from(u8::arbitrary(g)) + 1, /* Slot gap is always > 1, because there
-                                                        * has been a succesful aggregation */
-            price:    i64::arbitrary(g),
-            conf:     u64::arbitrary(g),
+            slot_gap:    u64::from(u8::arbitrary(g)) + 1, /* Slot gap is always > 1, because there
+                                                           * has been a succesful aggregation */
+            price:       i64::arbitrary(g),
+            conf:        u64::arbitrary(g),
+            max_latency: u8::arbitrary(g),
         }
     }
 }
@@ -44,6 +46,7 @@ impl Arbitrary for DataEvent {
 /// - slot_gap is a random number between 1 and u8::MAX + 1 (256)
 /// - price is a random i64
 /// - conf is a random u64
+/// - max_latency is a random u8
 #[quickcheck]
 fn test_twap(input: Vec<DataEvent>) -> bool {
     let mut price_cumulative = PriceCumulative {
@@ -56,7 +59,12 @@ fn test_twap(input: Vec<DataEvent>) -> bool {
     let mut data = Vec::<DataEvent>::new();
 
     for data_event in input {
-        price_cumulative.update(data_event.price, data_event.conf, data_event.slot_gap);
+        price_cumulative.update(
+            data_event.price,
+            data_event.conf,
+            data_event.slot_gap,
+            data_event.max_latency,
+        );
         data.push(data_event);
         price_cumulative.check_price(data.as_slice());
         price_cumulative.check_conf(data.as_slice());
@@ -67,7 +75,6 @@ fn test_twap(input: Vec<DataEvent>) -> bool {
     true
 }
 
-
 impl PriceCumulative {
     pub fn check_price(&self, data: &[DataEvent]) {
         assert_eq!(
@@ -87,12 +94,18 @@ impl PriceCumulative {
     }
     pub fn check_num_down_slots(&self, data: &[DataEvent]) {
         assert_eq!(
-            data.iter()
-                .fold(0, |acc, x| if x.slot_gap > PC_MAX_SEND_LATENCY.into() {
-                    acc + (x.slot_gap - PC_MAX_SEND_LATENCY as u64)
+            data.iter().fold(0, |acc, x| {
+                let latency_threshold = if x.max_latency == 0 {
+                    PC_MAX_SEND_LATENCY.into()
+                } else {
+                    x.max_latency.into()
+                };
+                if x.slot_gap > latency_threshold {
+                    acc + (x.slot_gap - latency_threshold)
                 } else {
                     acc
-                }),
+                }
+            }),
             self.num_down_slots
         );
     }
@@ -112,35 +125,65 @@ fn test_twap_unit() {
 
     let data = vec![
         DataEvent {
-            price:    1,
-            conf:     2,
-            slot_gap: 4,
+            price:       1,
+            conf:        2,
+            slot_gap:    4,
+            max_latency: 0,
+        },
+        DataEvent {
+            price:       i64::MAX,
+            conf:        u64::MAX,
+            slot_gap:    1,
+            max_latency: 0,
         },
         DataEvent {
-            price:    i64::MAX,
-            conf:     u64::MAX,
-            slot_gap: 1,
+            price:       -10,
+            conf:        4,
+            slot_gap:    30,
+            max_latency: 0,
         },
         DataEvent {
-            price:    -10,
-            conf:     4,
-            slot_gap: 30,
+            price:       1,
+            conf:        2,
+            slot_gap:    4,
+            max_latency: 5,
+        },
+        DataEvent {
+            price:       6,
+            conf:        7,
+            slot_gap:    8,
+            max_latency: 5,
         },
     ];
 
-    price_cumulative.update(data[0].price, data[0].conf, data[0].slot_gap);
+    price_cumulative.update(
+        data[0].price,
+        data[0].conf,
+        data[0].slot_gap,
+        data[0].max_latency,
+    );
     assert_eq!(price_cumulative.price, 5);
     assert_eq!(price_cumulative.conf, 10);
     assert_eq!(price_cumulative.num_down_slots, 3);
     assert_eq!(price_cumulative.unused, 0);
 
-    price_cumulative.update(data[1].price, data[1].conf, data[1].slot_gap);
+    price_cumulative.update(
+        data[1].price,
+        data[1].conf,
+        data[1].slot_gap,
+        data[1].max_latency,
+    );
     assert_eq!(price_cumulative.price, 9_223_372_036_854_775_812i128);
     assert_eq!(price_cumulative.conf, 18_446_744_073_709_551_625u128);
     assert_eq!(price_cumulative.num_down_slots, 3);
     assert_eq!(price_cumulative.unused, 0);
 
-    price_cumulative.update(data[2].price, data[2].conf, data[2].slot_gap);
+    price_cumulative.update(
+        data[2].price,
+        data[2].conf,
+        data[2].slot_gap,
+        data[2].max_latency,
+    );
     assert_eq!(price_cumulative.price, 9_223_372_036_854_775_512i128);
     assert_eq!(price_cumulative.conf, 18_446_744_073_709_551_745u128);
     assert_eq!(price_cumulative.num_down_slots, 8);
@@ -152,7 +195,7 @@ fn test_twap_unit() {
         num_down_slots: 0,
         unused:         0,
     };
-    price_cumulative_overflow.update(i64::MIN, u64::MAX, u64::MAX);
+    price_cumulative_overflow.update(i64::MIN, u64::MAX, u64::MAX, u8::MAX);
     assert_eq!(
         price_cumulative_overflow.price,
         i128::MIN - i128::from(i64::MIN)
@@ -163,9 +206,38 @@ fn test_twap_unit() {
     );
     assert_eq!(
         price_cumulative_overflow.num_down_slots,
-        u64::MAX - PC_MAX_SEND_LATENCY as u64
+        u64::MAX - u64::from(u8::MAX)
     );
     assert_eq!(price_cumulative_overflow.unused, 0);
+
+    let mut price_cumulative_nonzero_max_latency = PriceCumulative {
+        price:          1,
+        conf:           2,
+        num_down_slots: 3,
+        unused:         0,
+    };
+
+    price_cumulative_nonzero_max_latency.update(
+        data[3].price,
+        data[3].conf,
+        data[3].slot_gap,
+        data[3].max_latency,
+    );
+    assert_eq!(price_cumulative_nonzero_max_latency.price, 5);
+    assert_eq!(price_cumulative_nonzero_max_latency.conf, 10);
+    assert_eq!(price_cumulative_nonzero_max_latency.num_down_slots, 3);
+    assert_eq!(price_cumulative_nonzero_max_latency.unused, 0);
+
+    price_cumulative_nonzero_max_latency.update(
+        data[4].price,
+        data[4].conf,
+        data[4].slot_gap,
+        data[4].max_latency,
+    );
+    assert_eq!(price_cumulative_nonzero_max_latency.price, 53);
+    assert_eq!(price_cumulative_nonzero_max_latency.conf, 66);
+    assert_eq!(price_cumulative_nonzero_max_latency.num_down_slots, 6);
+    assert_eq!(price_cumulative_nonzero_max_latency.unused, 0);
 }
 
 #[test]
@@ -224,7 +296,6 @@ fn test_twap_with_price_account() {
         Err(OracleError::NeedsSuccesfulAggregation)
     );
 
-
     assert_eq!(price_data.price_cumulative.price, 1 - 2 * 10);
     assert_eq!(price_data.price_cumulative.conf, 2 + 2 * 5);
     assert_eq!(price_data.price_cumulative.num_down_slots, 3);