Batch publish (pyth-network#313, pyth-network#314, pyth-network#315)

---

Batch publish (pyth-network#313)

* feat: batch publish

* feat: batch publish 2 (wip)

* feat: verify that price feed index is set

* refactor: add pyth module

* feat: integrate pyth-price-publisher

* refactor: rename and move get_accumulator_keys

* test: make sure batch publish does not apply prices when price_index == 0

* chore: fix format

* fix: update deps

---

Update dependency, program id and test for pyth-price-store (pyth-network#314)

* test: modify batch publish tests

* chore: use real batch publish pid as default

* chore: update pyth-price-store dependency

---

Rename price store env var and bump version (pyth-network#315)

* chore: rename variable for pyth store pid

* chore: bump pythnet version to 1.14.179

Author: Riateche

Cargo.lock

@@ -74,7 +74,7 @@ use {
         accounts_db::{AccountShrinkThreshold, AccountsDbConfig},
         accounts_index::AccountSecondaryIndexes,
         accounts_update_notifier_interface::AccountsUpdateNotifier,
-        bank::{pyth_accumulator, Bank},
+        bank::{pyth, Bank},
         bank_forks::BankForks,
         commitment::BlockCommitmentCache,
         cost_model::CostModel,
@@ -1521,14 +1521,8 @@ fn load_blockstore(
         }
     }
 
-    for (key_name, pk_res) in pyth_accumulator::get_accumulator_keys() {
-        match pk_res {
-            Ok(pk) => info!("Accumulator {}: {}", key_name, pk),
-            Err(err) => {
-                error!("Failed to get Accumulator {}: {:?}", key_name, err);
-                std::process::abort();
-            }
-        }
+    for (key_name, pk) in pyth::get_pyth_keys() {
+        info!("Pyth key {}: {}", key_name, pk);
     }
 
     leader_schedule_cache.set_fixed_leader_schedule(config.fixed_leader_schedule.clone());

core/src/validator.rs

@@ -36,8 +36,9 @@ num-traits = { version = "0.2" }
 num_cpus = "1.13.1"
 once_cell = "1.12.0"
 ouroboros = "0.15.0"
-pyth-oracle = { git = "https://github.com/pyth-network/pyth-client", tag = "oracle-v2.33.0", features = ["library"] }
+pyth-oracle = { git = "https://github.com/pyth-network/pyth-client", rev = "256b57", features = ["library"] }
 pythnet-sdk = { git = "https://github.com/pyth-network/pyth-crosschain", version = "1.13.6", rev = "33f901aa45f4f0005aa5a84a1479b78ca9033074" }
+pyth-price-store = "0.1.0"
 rand = "0.7.0"
 rayon = "1.5.3"
 regex = "1.5.6"

programs/bpf/Cargo.lock

@@ -198,10 +198,7 @@ mod builtin_programs;
 mod sysvar_cache;
 mod transaction_account_state_info;
 
-pub mod pyth_accumulator;
-
-#[cfg(test)]
-mod pyth_accumulator_tests;
+pub mod pyth;
 
 pub const SECONDS_PER_YEAR: f64 = 365.25 * 24.0 * 60.0 * 60.0;
 
@@ -1410,7 +1407,7 @@ impl Bank {
         // state before the accumulator is used.  bank is in a fully
         // updated state before the accumulator is used.
         if !accumulator_moved_to_end_of_block {
-            pyth_accumulator::update_accumulator(&bank);
+            pyth::accumulator::update_accumulator(&bank);
         }
 
         bank
@@ -1796,7 +1793,7 @@ impl Bank {
             // the accumulator sysvar updates.  sysvars are in a fully updated
             // state before the accumulator sysvar updates.
             if !accumulator_moved_to_end_of_block {
-                pyth_accumulator::update_accumulator(&new);
+                pyth::accumulator::update_accumulator(&new);
             }
         });
 
@@ -3238,7 +3235,7 @@ impl Bank {
             // other tasks when freezing to avoid any conflicts.
             if accumulator_moved_to_end_of_block {
                 let mut measure = Measure::start("accumulator");
-                pyth_accumulator::update_accumulator(self);
+                pyth::accumulator::update_accumulator(self);
                 measure.stop();
 
                 debug!(

runtime/Cargo.toml

@@ -1,8 +1,12 @@
 use {
-    super::Bank,
-    crate::accounts_index::{IndexKey, ScanConfig, ScanError},
+    super::batch_publish,
+    crate::{
+        accounts_index::{IndexKey, ScanConfig, ScanError},
+        bank::Bank,
+    },
     borsh::BorshSerialize,
     byteorder::{LittleEndian, ReadBytesExt},
+    itertools::Itertools,
     log::*,
     pyth_oracle::validator::AggregationError,
     pythnet_sdk::{
@@ -19,7 +23,10 @@ use {
         hash::hashv,
         pubkey::Pubkey,
     },
-    std::env::{self, VarError},
+    std::{
+        collections::HashMap,
+        env::{self, VarError},
+    },
 };
 
 pub const ACCUMULATOR_RING_SIZE: u32 = 10_000;
@@ -53,6 +60,12 @@ lazy_static! {
             .parse()
             .unwrap(),
     );
+    pub static ref PRICE_STORE_PID: Pubkey = env_pubkey_or(
+        "PRICE_STORE_PID",
+        "3m6sv6HGqEbuyLV84mD7rJn4MAC9LhUa1y1AUNVqcPfr"
+            .parse()
+            .unwrap(),
+    );
 }
 
 /// Accumulator specific error type. It would be nice to use `transaction::Error` but it does
@@ -118,25 +131,6 @@ fn env_pubkey_or(var: &str, default: Pubkey) -> Pubkey {
     }
 }
 
-/// Get all accumulator related pubkeys from environment variables
-/// or return default if the variable is not set.
-pub fn get_accumulator_keys() -> Vec<(
-    &'static str,
-    std::result::Result<Pubkey, AccumulatorUpdateErrorV1>,
-)> {
-    vec![
-        ("MESSAGE_BUFFER_PID", Ok(*MESSAGE_BUFFER_PID)),
-        ("ACCUMULATOR_EMITTER_ADDR", Ok(*ACCUMULATOR_EMITTER_ADDR)),
-        ("ACCUMULATOR_SEQUENCE_ADDR", Ok(*ACCUMULATOR_SEQUENCE_ADDR)),
-        ("WORMHOLE_PID", Ok(*WORMHOLE_PID)),
-        ("ORACLE_PID", Ok(*ORACLE_PID)),
-        (
-            "STAKE_CAPS_PARAMETERS_ADDR",
-            Ok(*STAKE_CAPS_PARAMETERS_ADDR),
-        ),
-    ]
-}
-
 pub fn update_v1(
     bank: &Bank,
     v2_messages: &[Vec<u8>],
@@ -424,21 +418,37 @@ pub fn update_v2(bank: &Bank) -> std::result::Result<(), AccumulatorUpdateErrorV
         v2_messages.push(publisher_stake_caps_message);
     }
 
-    let mut measure = Measure::start("update_v2_aggregate_price");
+    let mut measure = Measure::start("extract_batch_publish_prices");
+    let mut new_prices = batch_publish::extract_batch_publish_prices(bank).unwrap_or_else(|err| {
+        warn!("extract_batch_publish_prices failed: {}", err);
+        HashMap::new()
+    });
+    measure.stop();
+    debug!("batch publish: loaded prices in {}us", measure.as_us());
 
+    let mut measure = Measure::start("update_v2_aggregate_price");
     for (pubkey, mut account) in accounts {
         let mut price_account_data = account.data().to_owned();
+        let price_account =
+            match pyth_oracle::validator::checked_load_price_account_mut(&mut price_account_data) {
+                Ok(data) => data,
+                Err(_err) => {
+                    continue;
+                }
+            };
+
+        let mut need_save =
+            batch_publish::apply_published_prices(price_account, &mut new_prices, bank.slot());
 
         // Perform Accumulation
         match pyth_oracle::validator::aggregate_price(
             bank.slot(),
             bank.clock().unix_timestamp,
             &pubkey.to_bytes().into(),
-            &mut price_account_data,
+            price_account,
         ) {
             Ok(messages) => {
-                account.set_data(price_account_data);
-                bank.store_account_and_update_capitalization(&pubkey, &account);
+                need_save = true;
                 v2_messages.extend(messages);
             }
             Err(err) => match err {
@@ -448,6 +458,16 @@ pub fn update_v2(bank: &Bank) -> std::result::Result<(), AccumulatorUpdateErrorV
                 }
             },
         }
+        if need_save {
+            account.set_data(price_account_data);
+            bank.store_account_and_update_capitalization(&pubkey, &account);
+        }
+    }
+    if !new_prices.is_empty() {
+        warn!(
+            "pyth batch publish: missing price feed accounts for indexes: {}",
+            new_prices.keys().join(", ")
+        );
     }
 
     measure.stop();