Support aggregation in validator

program/rust/Cargo.toml

@@ -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/src/accounts.rs

@@ -34,7 +34,6 @@ use {
     std::borrow::BorrowMut,
 };
 
-
 mod mapping;
 mod permission;
 mod price;
@@ -53,6 +52,7 @@ pub use {
     permission::PermissionAccount,
     price::{
         PriceAccount,
+        PriceAccountFlags,
         PriceComponent,
         PriceCumulative,
         PriceEma,

program/rust/src/accounts/price.rs

@@ -33,6 +33,7 @@ mod price_pythnet {
             },
             error::OracleError,
         },
+        bitflags::bitflags,
     };
 
     /// Pythnet-only extended price account format. This extension is
@@ -65,8 +66,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 +93,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 {

program/rust/src/lib.rs

@@ -10,6 +10,12 @@ mod instruction;
 mod processor;
 mod utils;
 
+#[cfg(feature = "library")]
+pub mod validator;
+
+#[cfg(feature = "library")]
+pub use solana_program;
+
 #[cfg(test)]
 mod tests;
 

program/rust/src/processor.rs

@@ -30,7 +30,11 @@ mod upd_product;
 pub use {
     add_price::add_price,
     add_product::add_product,
-    add_publisher::add_publisher,
+    add_publisher::{
+        add_publisher,
+        DISABLE_ACCUMULATOR_V2,
+        ENABLE_ACCUMULATOR_V2,
+    },
     del_price::del_price,
     del_product::del_product,
     del_publisher::del_publisher,

program/rust/src/processor/add_publisher.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,7 +70,6 @@ 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
@@ -71,6 +78,18 @@ pub fn add_publisher(
         let num_comps = try_convert::<u32, usize>(price_data.num_)?;
         sort_price_comps(&mut price_data.comp_, num_comps)?;
         return Ok(());
+    } else 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(());
     }
 
     if price_data.num_ >= PC_NUM_COMP {
@@ -224,7 +243,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);
     }

program/rust/src/processor/upd_price.rs

@@ -2,6 +2,7 @@ use {
     crate::{
         accounts::{
             PriceAccount,
+            PriceAccountFlags,
             PriceComponent,
             PriceInfo,
             PythOracleSerialize,
@@ -131,6 +132,7 @@ pub fn upd_price(
 
     let publisher_index: usize;
     let latest_aggregate_price: PriceInfo;
+    let flags: PriceAccountFlags;
 
     // The price_data borrow happens in a scope because it must be
     // dropped before we borrow again as raw data pointer for the C
@@ -159,47 +161,59 @@ pub fn upd_price(
                     && cmd_args.publishing_slot <= clock.slot),
             ProgramError::InvalidArgument,
         )?;
-    }
 
-    // Try to update the aggregate
-    #[allow(unused_variables)]
-    if clock.slot > latest_aggregate_price.pub_slot_ {
-        let updated = unsafe {
-            // NOTE: c_upd_aggregate must use a raw pointer to price
-            // data. Solana's `<account>.borrow_*` methods require exclusive
-            // access, i.e. no other borrow can exist for the account.
-            c_upd_aggregate(
-                price_account.try_borrow_mut_data()?.as_mut_ptr(),
-                clock.slot,
-                clock.unix_timestamp,
-            )
-        };
+        flags = price_data.flags;
+    }
 
-        // If the aggregate was successfully updated, calculate the difference and update TWAP.
-        if updated {
-            let agg_diff = (clock.slot as i64)
-                - load_checked::<PriceAccount>(price_account, cmd_args.header.version)?.prev_slot_
-                    as i64;
-            // Encapsulate TWAP update logic in a function to minimize unsafe block scope.
-            unsafe {
-                c_upd_twap(price_account.try_borrow_mut_data()?.as_mut_ptr(), agg_diff);
+    if !flags.contains(PriceAccountFlags::ACCUMULATOR_V2) {
+        // Try to update the aggregate
+        #[allow(unused_variables)]
+        if clock.slot > latest_aggregate_price.pub_slot_ {
+            let updated = unsafe {
+                // NOTE: c_upd_aggregate must use a raw pointer to price
+                // data. Solana's `<account>.borrow_*` methods require exclusive
+                // access, i.e. no other borrow can exist for the account.
+                c_upd_aggregate(
+                    price_account.try_borrow_mut_data()?.as_mut_ptr(),
+                    clock.slot,
+                    clock.unix_timestamp,
+                )
+            };
+
+            // If the aggregate was successfully updated, calculate the difference and update TWAP.
+            if updated {
+                let agg_diff = (clock.slot as i64)
+                    - load_checked::<PriceAccount>(price_account, cmd_args.header.version)?
+                        .prev_slot_ as i64;
+                // Encapsulate TWAP update logic in a function to minimize unsafe block scope.
+                unsafe {
+                    c_upd_twap(price_account.try_borrow_mut_data()?.as_mut_ptr(), agg_diff);
+                }
+                let mut price_data =
+                    load_checked::<PriceAccount>(price_account, cmd_args.header.version)?;
+                // We want to send a message every time the aggregate price updates. However, during the migration,
+                // not every publisher will necessarily provide the accumulator accounts. The message_sent_ flag
+                // ensures that after every aggregate update, the next publisher who provides the accumulator accounts
+                // will send the message.
+                price_data.message_sent_ = 0;
+                price_data.update_price_cumulative()?;
             }
-            let mut price_data =
-                load_checked::<PriceAccount>(price_account, cmd_args.header.version)?;
-            // We want to send a message every time the aggregate price updates. However, during the migration,
-            // not every publisher will necessarily provide the accumulator accounts. The message_sent_ flag
-            // ensures that after every aggregate update, the next publisher who provides the accumulator accounts
-            // will send the message.
-            price_data.message_sent_ = 0;
-            price_data.update_price_cumulative()?;
         }
     }
 
     // Reload price data as a struct after c_upd_aggregate() borrow is dropped
     let mut price_data = load_checked::<PriceAccount>(price_account, cmd_args.header.version)?;
 
     // Feature-gated accumulator-specific code, used only on pythnet/pythtest
-    {
+    let need_message_buffer_update = if flags.contains(PriceAccountFlags::ACCUMULATOR_V2) {
+        // We need to clear old messages.
+        !flags.contains(PriceAccountFlags::MESSAGE_BUFFER_CLEARED)
+    } else {
+        // V1
+        true
+    };
+
+    if need_message_buffer_update {
         if let Some(accumulator_accounts) = maybe_accumulator_accounts {
             if price_data.message_sent_ == 0 {
                 // Check that the oracle PDA is correctly configured for the program we are calling.
@@ -232,12 +246,16 @@ pub fn upd_price(
                     },
                 ];
 
-                let message = vec![
-                    price_data
-                        .as_price_feed_message(price_account.key)
-                        .to_bytes(),
-                    price_data.as_twap_message(price_account.key).to_bytes(),
-                ];
+                let message = if flags.contains(PriceAccountFlags::ACCUMULATOR_V2) {
+                    vec![]
+                } else {
+                    vec![
+                        price_data
+                            .as_price_feed_message(price_account.key)
+                            .to_bytes(),
+                        price_data.as_twap_message(price_account.key).to_bytes(),
+                    ]
+                };
 
                 // Append a TWAP message if available
 
@@ -257,6 +275,11 @@ pub fn upd_price(
 
                 invoke_signed(&create_inputs_ix, accounts, &[auth_seeds_with_bump])?;
                 price_data.message_sent_ = 1;
+                if flags.contains(PriceAccountFlags::ACCUMULATOR_V2) {
+                    price_data
+                        .flags
+                        .insert(PriceAccountFlags::MESSAGE_BUFFER_CLEARED);
+                }
             }
         }
     }
@@ -287,7 +310,7 @@ pub fn upd_price(
 /// to get the result faster if the list is sorted. If the list is not sorted, it falls back to
 /// a linear search.
 #[inline(always)]
-fn find_publisher_index(comps: &[PriceComponent], key: &Pubkey) -> Option<usize> {
+pub fn find_publisher_index(comps: &[PriceComponent], key: &Pubkey) -> Option<usize> {
     // Verify that publisher is authorized by initially binary searching
     // for the publisher's component in the price account. The binary
     // search might not work if the publisher list is not sorted; therefore

program/rust/src/tests/mod.rs

@@ -2,6 +2,7 @@ mod pyth_simulator;
 mod test_add_price;
 mod test_add_product;
 mod test_add_publisher;
+mod test_aggregate_v2;
 mod test_aggregation;
 mod test_c_code;
 mod test_check_valid_signable_account_or_permissioned_funding_account;