feat: add dynamic pricing based on staleness

This commit adds dynamic (exponential) pricing based on staleness
of the feeds that the publishers publish.

This feature, combined with the recent fees for the price updates
causes the spike on fee to last some time before going away.

Author: ali-behjati

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "pyth-agent"
-version = "2.2.0"
+version = "2.3.0"
 edition = "2021"
 
 [[bin]]

Cargo.toml

@@ -83,6 +83,16 @@ key_store.mapping_key = "RelevantOracleMappingAddress"
 # calculated based on the network previous prioritization fees.
 # exporter.dynamic_compute_unit_pricing_enabled = false
 
+# Maximum total compute unit fee paid for a single transaction. Defaults to 0.001 SOL. This
+# is a safety measure while using dynamic compute price to prevent the exporter from paying
+# too much for a single transaction. The default is 10**12 micro lamports (0.001 SOL).
+# exporter.maximum_total_compute_fee_micro_lamports = 1000000000000
+
+# Maximum slot gap between the current slot and the oldest slot amongst all the accounts in
+# the batch. This is used to calculate the dynamic price per compute unit. When the slot gap
+# reaches this number we will use the maximum total_compute_fee for the transaction.
+# exporter.maximum_slot_gap_for_dynamic_compute_unit_price = 25
+
 # Duration of the interval with which to poll the status of transactions.
 # It is recommended to set this to a value close to exporter.publish_interval_duration
 # exporter.transaction_monitor.poll_interval_duration = "4s"

config/config.toml

@@ -10,13 +10,13 @@
 |  +--------+    +----------+    | |  +----------+    +--------+    |
 |  | Oracle |    | Exporter |    | |  | Exporter |    | Oracle |    |
 |  +--------+    +----------+    | |  +----------+    +--------+    |
-|      |              ^  ^       | |     ^  ^             |         |
-+------|--------------|--|-------+ +-----|--|-------------|---------+
-       |              |  |               |  |             |      +------------------------+
-       |        +--------|---------------|---------+      |      |  Pythd Websocket API   |
-       |        |            Local Store |         |<---------------+-------+   +------+  |
-       |        +--------|---------------|---------+      |      |  |       |<--|JRPC  |  |
-       v                 |               |        |       v      |  |Adapter|   | WS   |  |
+|      |           ^  ^  ^       | |     ^  ^  ^          |         |
++------|-----------|--|--|-------+ +-----|--|--|----------|---------+
+       |           |  |  |               |  |  |          |      +------------------------+
+       |        +--|-----|---------------|-----|---+      |      |  Pythd Websocket API   |
+       |        |  |         Local Store |     |   |<---------------+-------+   +------+  |
+       |        +--|-----|---------------|-----|---+      |      |  |       |<--|JRPC  |  |
+       v           |     |               |     |  |       v      |  |Adapter|   | WS   |  |
     +--------------------|---------------|--------|-----------+  |  |       |-->|Server|  |
     |                    |  Global Store |        |           |---->+-------+   +------+  |
     +--------------------|---------------|--------|-----------+  |               ^    |   |
@@ -39,7 +39,8 @@ Publisher data write path:
 - The Adapter then transforms this into the Pyth SDK data structures and sends it to the Local Store.
 - The Local Store holds the latest price data the user has submitted for each price feed.
 - The Exporters periodically query the Local Store for the latest user-submitted data,
-and send it to the RPC node.
+and send it to the RPC node. They query the Global Store to get the on-chain status to dynamically
+adjust the compute unit price (if enabled).
 
 Publisher data read path:
 - The Oracles continually fetch data from the RPC node, and pass this to the Global Store.
@@ -128,7 +129,9 @@ impl Agent {
         // Spawn the primary network
         jhs.extend(network::spawn_network(
             self.config.primary_network.clone(),
+            network::Network::Primary,
             local_store_tx.clone(),
+            global_store_lookup_tx.clone(),
             primary_oracle_updates_tx,
             primary_keypair_loader_tx,
             logger.new(o!("primary" => true)),
@@ -138,7 +141,9 @@ impl Agent {
         if let Some(config) = &self.config.secondary_network {
             jhs.extend(network::spawn_network(
                 config.clone(),
+                network::Network::Secondary,
                 local_store_tx.clone(),
+                global_store_lookup_tx.clone(),
                 secondary_oracle_updates_tx,
                 secondary_keypair_loader_tx,
                 logger.new(o!("primary" => false)),

src/agent.rs

@@ -56,6 +56,7 @@ impl MetricsServer {
 
         self.global_store_lookup_tx
             .send(Lookup::LookupAllAccountsData {
+                network:   super::solana::network::Network::Primary,
                 result_tx: global_data_tx,
             })
             .await?;

src/agent/dashboard.rs

@@ -359,7 +359,10 @@ impl Adapter {
     async fn lookup_all_accounts_data(&self) -> Result<AllAccountsData> {
         let (result_tx, result_rx) = oneshot::channel();
         self.global_store_lookup_tx
-            .send(global::Lookup::LookupAllAccountsData { result_tx })
+            .send(global::Lookup::LookupAllAccountsData {
+                network: solana::network::Network::Primary,
+                result_tx,
+            })
             .await?;
         result_rx.await?
     }

src/agent/pythd/adapter.rs

@@ -36,6 +36,12 @@ pub mod network {
         },
     };
 
+    #[derive(Clone, Copy, Serialize, Deserialize, Debug)]
+    pub enum Network {
+        Primary,
+        Secondary,
+    }
+
     pub fn default_rpc_url() -> String {
         "http://localhost:8899".to_string()
     }
@@ -72,9 +78,11 @@ pub mod network {
 
     pub fn spawn_network(
         config: Config,
+        network: Network,
         local_store_tx: Sender<store::local::Message>,
-        global_store_update_tx: mpsc::Sender<global::Update>,
-        keypair_request_tx: mpsc::Sender<KeypairRequest>,
+        global_store_lookup_tx: Sender<global::Lookup>,
+        global_store_update_tx: Sender<global::Update>,
+        keypair_request_tx: Sender<KeypairRequest>,
         logger: Logger,
     ) -> Result<Vec<JoinHandle<()>>> {
         // Publisher permissions updates between oracle and exporter
@@ -96,11 +104,13 @@ pub mod network {
         // Spawn the Exporter
         let exporter_jhs = exporter::spawn_exporter(
             config.exporter,
+            network,
             &config.rpc_url,
             config.rpc_timeout,
             publisher_permissions_rx,
             KeyStore::new(config.key_store.clone(), &logger)?,
             local_store_tx,
+            global_store_lookup_tx,
             keypair_request_tx,
             logger,
         )?;