Merge pull request #53 from butaneprotocol/individual-price-feeds

Publish individual price feeds

Author: SupernaviX

Diff for: Cargo.lock

@@ -27,21 +27,22 @@ ed25519-dalek = { version = "2.1" }
 frost-ed25519 = { version = "2.1", features = ["serialization"] }
 futures = "0.3"
 hex = "0.4.3"
+itertools = "0.14"
 kupon = { git = "https://github.com/SundaeSwap-finance/kupon", rev = "ccc0b37" }
-minicbor = { version = "0.25", features = ["derive", "std"] }
-minicbor-io = { version = "0.20", features = ["async-io"] }
+minicbor = { version = "0.26", features = ["derive", "std"] }
+minicbor-io = { version = "0.21", features = ["async-io"] }
 num-bigint = "0.4"
 num-integer = "0.1"
 num-rational = "0.4"
 num-traits = "0.2"
-opentelemetry = "0.27"
-opentelemetry-otlp = { version = "0.27", features = ["gzip-tonic", "tls-webpki-roots"] }
-opentelemetry_sdk = { version = "0.27", features = ["rt-tokio"] }
-pallas-crypto = "0.32"
-pallas-primitives = "0.32"
+opentelemetry = "0.29"
+opentelemetry-otlp = { version = "0.29", features = ["grpc-tonic", "gzip-tonic", "tls-webpki-roots"] }
+opentelemetry_sdk = { version = "0.29", features = ["rt-tokio"] }
+pallas-crypto = { git = "https://github.com/txpipe/pallas", rev = "c3aad16" }
+pallas-primitives = { git = "https://github.com/txpipe/pallas", rev = "c3aad16" }
 rand = "0.8.5"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
-rust_decimal = "1.34.3"
+rust_decimal = "1"
 rustls = { version = "0.23", default-features = false, features = ["logging", "ring", "tls12"] }
 serde = "1"
 serde_json = "1"
@@ -51,8 +52,9 @@ tokio-tungstenite = { version = "0.26", features = ["rustls-tls-webpki-roots" ]}
 tokio-util = { version = "0.7", features = ["full"] }
 tonic = "0.12"
 tracing = "0.1.40"
-tracing-opentelemetry = "0.28"
+tracing-opentelemetry = "0.30"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
+urlencoding = "2"
 uuid = { version = "1", features = ["v4"] }
 x25519-dalek = { version = "2" }
 

Diff for: Cargo.toml

@@ -177,6 +177,9 @@ currencies:
     digits: 6
   - name: XAUt
     digits: 6
+feeds:
+  currencies:
+    - ADA
 binance:
   tokens:
     - token: ADA

Diff for: config.base.yaml

@@ -16,7 +16,7 @@ use crate::{
     config::OracleConfig,
     network::NodeId,
     price_aggregator::TokenPrice,
-    signature_aggregator::{Payload, TimestampedPayloadEntry},
+    signature_aggregator::{Payload, SyntheticPayloadEntry},
 };
 
 #[derive(Clone, Serialize)]
@@ -105,7 +105,7 @@ async fn report_all_prices(State(state): State<APIState>) -> impl IntoResponse {
 
 #[allow(clippy::large_enum_variant)]
 pub enum Response {
-    Ok(TimestampedPayloadEntry),
+    Ok(SyntheticPayloadEntry),
     NotFound,
 }
 impl IntoResponse for Response {
@@ -125,9 +125,9 @@ async fn report_payload(
 ) -> impl IntoResponse {
     let payload = state.payload_source.borrow().clone();
     match payload
-        .entries
+        .synthetics
         .iter()
-        .find(|&p| p.entry.synthetic == synthetic)
+        .find(|&p| p.synthetic == synthetic)
     {
         Some(entry) => Response::Ok(entry.clone()),
         None => Response::NotFound,

Diff for: src/api.rs

@@ -32,11 +32,13 @@ struct RawOracleConfig {
     pub use_persisted_prices: bool,
     pub max_synthetic_divergence: Decimal,
     pub publish_url: Option<String>,
+    pub publish_feed_base_url: Option<String>,
     pub logs: RawLogConfig,
     pub frost_address: Option<String>,
     pub keygen: KeygenConfig,
     pub synthetics: Vec<SyntheticConfig>,
     pub currencies: Vec<CurrencyConfig>,
+    pub feeds: FeedConfig,
     pub binance: BinanceConfig,
     pub bybit: ByBitConfig,
     pub coinbase: CoinbaseConfig,
@@ -66,11 +68,13 @@ pub struct OracleConfig {
     pub use_persisted_prices: bool,
     pub max_synthetic_divergence: Decimal,
     pub publish_url: Option<String>,
+    pub publish_feed_base_url: Option<String>,
     pub logs: LogConfig,
     pub frost_address: Option<String>,
     pub keygen: KeygenConfig,
     pub synthetics: Vec<SyntheticConfig>,
     pub currencies: Vec<CurrencyConfig>,
+    pub feeds: FeedConfig,
     pub bybit: ByBitConfig,
     pub binance: BinanceConfig,
     pub coinbase: CoinbaseConfig,
@@ -157,6 +161,11 @@ impl TryFrom<RawOracleConfig> for OracleConfig {
             uptrace_dsn: raw.logs.uptrace_dsn,
         };
 
+        let publish_feed_base_url = raw.publish_feed_base_url.or_else(|| {
+            let base_url = raw.publish_url.as_ref()?.strip_suffix("/oraclePrices")?;
+            Some(format!("{base_url}/oracles"))
+        });
+
         Ok(Self {
             id,
             label,
@@ -173,11 +182,13 @@ impl TryFrom<RawOracleConfig> for OracleConfig {
             use_persisted_prices: raw.use_persisted_prices,
             max_synthetic_divergence: raw.max_synthetic_divergence,
             publish_url: raw.publish_url,
+            publish_feed_base_url,
             logs,
             frost_address: raw.frost_address,
             keygen: raw.keygen,
             synthetics: raw.synthetics,
             currencies: raw.currencies,
+            feeds: raw.feeds,
             binance: raw.binance,
             bybit: raw.bybit,
             coinbase: raw.coinbase,
@@ -275,6 +286,11 @@ pub struct CurrencyConfig {
     pub digits: u32,
 }
 
+#[derive(Debug, Deserialize)]
+pub struct FeedConfig {
+    pub currencies: Vec<String>,
+}
+
 #[derive(Debug, Deserialize)]
 pub struct BinanceConfig {
     pub tokens: Vec<BinanceTokenConfig>,

Diff for: src/config.rs

@@ -14,16 +14,18 @@ use tracing_subscriber::{
 use crate::{config::LogConfig, network::NodeId};
 
 pub enum OtelGuard {
-    Enabled(metrics::SdkMeterProvider),
+    Enabled(trace::SdkTracerProvider, metrics::SdkMeterProvider),
     Disabled,
 }
 impl Drop for OtelGuard {
     fn drop(&mut self) {
-        if let Self::Enabled(meter_provider) = &self {
+        if let Self::Enabled(tracer_provider, meter_provider) = &self {
             if let Err(err) = meter_provider.shutdown() {
                 eprintln!("{:?}", err);
             }
-            global::shutdown_tracer_provider();
+            if let Err(err) = tracer_provider.shutdown() {
+                eprintln!("{:?}", err);
+            }
         }
     }
 }
@@ -57,7 +59,7 @@ where
             let metrics_layer = tracing_opentelemetry::MetricsLayer::new(meter_provider.clone());
 
             Dispatch::new(subscriber.with(metrics_layer).with(tracer_layer)).init();
-            Ok(OtelGuard::Enabled(meter_provider))
+            Ok(OtelGuard::Enabled(tracer_provider, meter_provider))
         }
         None => {
             Dispatch::new(subscriber).init();
@@ -77,13 +79,15 @@ fn init_providers(
     name: &str,
     endpoint: &str,
     uptrace_dsn: Option<&String>,
-) -> Result<(trace::TracerProvider, metrics::SdkMeterProvider)> {
-    let resource = Resource::default().merge(&Resource::new([
-        KeyValue::new("service.name", name.to_string()),
-        KeyValue::new("service.namespace", "oracles"),
-        KeyValue::new("service.instance.id", id.to_string()),
-        KeyValue::new("service.version", env!("CARGO_PKG_VERSION")),
-    ]));
+) -> Result<(trace::SdkTracerProvider, metrics::SdkMeterProvider)> {
+    let resource = Resource::builder()
+        .with_attributes([
+            KeyValue::new("service.name", name.to_string()),
+            KeyValue::new("service.namespace", "oracles"),
+            KeyValue::new("service.instance.id", id.to_string()),
+            KeyValue::new("service.version", env!("CARGO_PKG_VERSION")),
+        ])
+        .build();
 
     let exporter_provider = ExporterProvider::new(endpoint, uptrace_dsn)?;
 
@@ -96,9 +100,9 @@ fn init_providers(
 fn init_tracer_provider(
     resource: Resource,
     exporter_provider: &ExporterProvider,
-) -> Result<trace::TracerProvider> {
+) -> Result<trace::SdkTracerProvider> {
     let exporter = exporter_provider.span()?;
-    let processor = trace::BatchSpanProcessor::builder(exporter, opentelemetry_sdk::runtime::Tokio)
+    let processor = trace::BatchSpanProcessor::builder(exporter)
         .with_batch_config(
             opentelemetry_sdk::trace::BatchConfigBuilder::default()
                 .with_max_queue_size(30000)
@@ -108,7 +112,7 @@ fn init_tracer_provider(
         )
         .build();
 
-    let provider = trace::TracerProvider::builder()
+    let provider = trace::SdkTracerProvider::builder()
         .with_span_processor(processor)
         .with_resource(resource)
         .build();
@@ -123,10 +127,8 @@ fn init_meter_provider(
     exporter_provider: &ExporterProvider,
 ) -> Result<metrics::SdkMeterProvider> {
     let exporter = exporter_provider.metric()?;
-    let reader =
-        metrics::PeriodicReader::builder(exporter, opentelemetry_sdk::runtime::Tokio).build();
     let provider = metrics::SdkMeterProvider::builder()
-        .with_reader(reader)
+        .with_periodic_exporter(exporter)
         .with_resource(resource)
         .build();
 

Diff for: src/instrumentation.rs

@@ -10,6 +10,7 @@ use oracles::{
     instrumentation,
     network::Network,
     price_aggregator::PriceAggregator,
+    price_feed::PriceData,
     publisher::Publisher,
     raft::{Raft, RaftClient, RaftLeader},
     signature_aggregator::SignatureAggregator,
@@ -45,7 +46,10 @@ struct Node {
 impl Node {
     pub fn new(config: Arc<OracleConfig>) -> Result<Self> {
         let (leader_sink, leader_source) = watch::channel(RaftLeader::Unknown);
-        let (price_feed_sink, price_feed_source) = watch::channel(vec![]);
+        let (price_sink, price_source) = watch::channel(PriceData {
+            synthetics: vec![],
+            generics: vec![],
+        });
         let (price_audit_sink, price_audit_source) = watch::channel(vec![]);
         let (raft_client, raft_source) = RaftClient::new();
 
@@ -63,7 +67,7 @@ impl Node {
             &config,
             &mut network,
             leader_sink,
-            price_feed_source.clone(),
+            price_source.clone(),
             raft_source,
         );
 
@@ -72,15 +76,15 @@ impl Node {
                 &config,
                 &mut network,
                 raft_client,
-                price_feed_source,
+                price_source,
                 leader_source,
             )?
         } else {
-            SignatureAggregator::single(&config, raft_client, price_feed_source, leader_source)?
+            SignatureAggregator::single(&config, raft_client, price_source, leader_source)?
         };
 
         let price_aggregator = PriceAggregator::new(
-            price_feed_sink,
+            price_sink,
             price_audit_sink,
             payload_source.clone(),
             config.clone(),