Support aggregation in validator (#410)

* feat(program): match version with pythnet

* feat: validator access point (wip)

* feat: add price account flags

* feat(program): fill out validator aggregation code

* feat: don't aggregate on v2, clear message buffer, report validator aggregation errors

* feat: return price feed messages from validator aggregation

* test: enable v2 aggregation

* test: test aggregation toggle

* fix: don't aggregate in validator if already aggregated in this slot

* chore: reexport solana_program and add derive debug

* chore: add comments

* fix: don't revert if c_upd_aggregate returns false

* fix: make update_price_cumulative infallible; don't generate v2 messages until v1 buffer is cleared

* chore: remove publisher sorting hack

* chore: remove AggregationOutcome

* fix: remove debug_assert

* fix: allow clearing message buffers regardless of message_sent_ flag

* test: add test_upd_price_with_validator

* chore: add missing imports and fix warnings

* test: verify aggregation output

* chore: bump version

---------

Co-authored-by: Reisen <[email protected]>

Author: Riateche

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "pyth-oracle"
-version = "2.26.0"
+version = "2.32.0"
 edition = "2021"
 license = "Apache 2.0"
 publish = false
@@ -9,7 +9,7 @@ publish = false
 bindgen = "0.60.1"
 
 [dependencies]
-solana-program = "=1.13.3"
+solana-program = "=1.14.17"
 bytemuck = "1.11.0"
 thiserror = "1.0"
 num-derive = "0.3"
@@ -18,10 +18,12 @@ byteorder = "1.4.3"
 serde = { version = "1.0", features = ["derive"], optional = true }
 strum = { version = "0.24.1", features = ["derive"], optional = true }
 pythnet-sdk = { git = "https://github.com/pyth-network/pyth-crosschain", rev="60144002053a93f424be70decd8a8ccb8d618d81"}
+solana-sdk = { version = "=1.14.17", optional = true }
+bitflags = { version = "2.6.0", features = ["bytemuck"] }
 
 [dev-dependencies]
-solana-program-test = "=1.13.3"
-solana-sdk = "=1.13.3"
+solana-program-test = "=1.14.17"
+solana-sdk = "=1.14.17"
 tokio = "1.14.1"
 hex = "0.3.1"
 quickcheck = "1"
@@ -34,10 +36,15 @@ serde_json = "1.0"
 test-generator = "0.3.1"
 csv = "1.1"
 
+# Downgrade to be compatible with Rust 1.60
+tracing-subscriber = "=0.3.0"
+time-macros = "=0.2.3"
+time = "=0.3.7"
+
 [features]
 check = [] # Skips make build in build.rs, use with cargo-clippy and cargo-check
 debug = []
-library = []
+library = ["solana-sdk"]
 
 [lib]
 crate-type = ["cdylib", "lib"]

program/rust/Cargo.toml

@@ -34,7 +34,6 @@ use {
     std::borrow::BorrowMut,
 };
 
-
 mod mapping;
 mod permission;
 mod price;
@@ -44,6 +43,8 @@ mod product;
 #[cfg(feature = "strum")]
 pub use price::MessageType;
 #[cfg(test)]
+pub use price::PriceCumulative;
+#[cfg(test)]
 pub use product::{
     account_has_key_values,
     create_pc_str_t,
@@ -53,14 +54,13 @@ pub use {
     permission::PermissionAccount,
     price::{
         PriceAccount,
+        PriceAccountFlags,
         PriceComponent,
-        PriceCumulative,
         PriceEma,
         PriceInfo,
         PythOracleSerialize,
     },
     product::{
-        read_pc_str_t,
         update_product_metadata,
         ProductAccount,
     },

program/rust/src/accounts.rs

@@ -25,14 +25,12 @@ mod price_pythnet {
 
     use {
         super::*,
-        crate::{
-            c_oracle_header::{
-                PC_MAX_SEND_LATENCY,
-                PC_NUM_COMP_PYTHNET,
-                PC_STATUS_TRADING,
-            },
-            error::OracleError,
+        crate::c_oracle_header::{
+            PC_MAX_SEND_LATENCY,
+            PC_NUM_COMP_PYTHNET,
+            PC_STATUS_TRADING,
         },
+        bitflags::bitflags,
     };
 
     /// Pythnet-only extended price account format. This extension is
@@ -65,8 +63,9 @@ mod price_pythnet {
         pub message_sent_:      u8,
         /// Configurable max latency in slots between send and receive
         pub max_latency_:       u8,
+        /// Various flags
+        pub flags:              PriceAccountFlags,
         /// Unused placeholder for alignment
-        pub unused_2_:          i8,
         pub unused_3_:          i32,
         /// Corresponding product account
         pub product_account:    Pubkey,
@@ -91,6 +90,18 @@ mod price_pythnet {
         pub price_cumulative:   PriceCumulative,
     }
 
+    bitflags! {
+        #[repr(C)]
+        #[derive(Copy, Clone, Pod, Zeroable)]
+        pub struct PriceAccountFlags: u8 {
+            /// If set, the program doesn't do accumulation, but validator does.
+            const ACCUMULATOR_V2 = 0b1;
+            /// If unset, the program will remove old messages from its message buffer account
+            /// and set this flag.
+            const MESSAGE_BUFFER_CLEARED = 0b10;
+        }
+    }
+
     impl PriceAccountPythnet {
         pub fn as_price_feed_message(&self, key: &Pubkey) -> PriceFeedMessage {
             let (price, conf, publish_time) = if self.agg_.status_ == PC_STATUS_TRADING {
@@ -111,17 +122,14 @@ mod price_pythnet {
             }
         }
         /// This function gets triggered when there's a succesful aggregation and updates the cumulative sums
-        pub fn update_price_cumulative(&mut self) -> Result<(), OracleError> {
+        pub fn update_price_cumulative(&mut self) {
             if self.agg_.status_ == PC_STATUS_TRADING {
                 self.price_cumulative.update(
                     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 {
-                Err(OracleError::NeedsSuccesfulAggregation)
             }
         }
 

program/rust/src/accounts/price.rs

@@ -10,6 +10,12 @@ mod instruction;
 mod processor;
 mod utils;
 
+#[cfg(any(test, feature = "library"))]
+pub mod validator;
+
+#[cfg(feature = "library")]
+pub use solana_program;
+
 #[cfg(test)]
 mod tests;
 
@@ -31,11 +37,13 @@ pub use accounts::{
     MappingAccount,
     PermissionAccount,
     PriceAccount,
+    PriceAccountFlags,
     PriceComponent,
     PriceEma,
     PriceInfo,
     ProductAccount,
     PythAccount,
+    PythOracleSerialize,
 };
 use {
     crate::error::OracleError,

program/rust/src/lib.rs

@@ -27,6 +27,11 @@ mod upd_permissions;
 mod upd_price;
 mod upd_product;
 
+#[cfg(test)]
+pub use add_publisher::{
+    DISABLE_ACCUMULATOR_V2,
+    ENABLE_ACCUMULATOR_V2,
+};
 pub use {
     add_price::add_price,
     add_product::add_product,

program/rust/src/processor.rs

@@ -2,6 +2,7 @@ use {
     crate::{
         accounts::{
             PriceAccount,
+            PriceAccountFlags,
             PriceComponent,
             PythAccount,
         },
@@ -33,6 +34,13 @@ use {
     std::mem::size_of,
 };
 
+pub const ENABLE_ACCUMULATOR_V2: [u8; 32] = [
+    0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1,
+];
+pub const DISABLE_ACCUMULATOR_V2: [u8; 32] = [
+    0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 2,
+];
+
 /// Add publisher to symbol account
 // account[0] funding account       [signer writable]
 // account[1] price account         [signer writable]
@@ -62,14 +70,19 @@ pub fn add_publisher(
         &cmd_args.header,
     )?;
 
-
     let mut price_data = load_checked::<PriceAccount>(price_account, cmd_args.header.version)?;
 
-    // Use the call with the default pubkey (000..) as a trigger to sort the publishers as a
-    // migration step from unsorted list to sorted list.
-    if cmd_args.publisher == Pubkey::default() {
-        let num_comps = try_convert::<u32, usize>(price_data.num_)?;
-        sort_price_comps(&mut price_data.comp_, num_comps)?;
+    if cmd_args.publisher == Pubkey::from(ENABLE_ACCUMULATOR_V2) {
+        // Hack: we use add_publisher instruction to configure the `ACCUMULATOR_V2` flag. Using a new
+        // instruction would be cleaner but it would require more work in the tooling.
+        // These special cases can be removed along with the v1 aggregation code once the transition
+        // is complete.
+        price_data.flags.insert(PriceAccountFlags::ACCUMULATOR_V2);
+        return Ok(());
+    } else if cmd_args.publisher == Pubkey::from(DISABLE_ACCUMULATOR_V2) {
+        price_data
+            .flags
+            .remove(PriceAccountFlags::ACCUMULATOR_V2 | PriceAccountFlags::MESSAGE_BUFFER_CLEARED);
         return Ok(());
     }
 
@@ -224,7 +237,6 @@ mod test {
         let mut rust_std_sorted_comps = comps.get(..num_comps).unwrap().to_vec();
         rust_std_sorted_comps.sort_by_key(|x| x.pub_);
 
-
         assert_eq!(sort_price_comps(&mut comps, num_comps), Ok(()));
         assert_eq!(comps.get(..num_comps).unwrap(), rust_std_sorted_comps);
     }