feat(lazer): add DynamicValue conversions, add source to GovernanceInstruction

Author: Riateche

lazer/Cargo.lock

@@ -14,19 +14,21 @@ package pyth_lazer_transaction;
 // Representation of a complete governance instruction. This value will be signed
 // by a governance source.
 message GovernanceInstruction {
+    // [required] Governance source that signed this instruction.
+    optional GovernanceSource source = 1;
     // Action requested by this instruction. For the instruction to be accepted, all directives
     // must be successfully applied. In case of any failure, the whole instruction is reverted.
     // However, note that if the instruction targets multiple (or all) shards, each shard will
     // accept or reject the instruction independently of other shards.
-    repeated GovernanceDirective directives = 1;
+    repeated GovernanceDirective directives = 2;
     // [optional] If specified, the instruction will be rejected if the current timestamp
     // is less than the specified value. In case of rejection, the same instruction can be resubmitted
     // and executed later once the time requirement is met.
-    optional google.protobuf.Timestamp min_execution_timestamp = 2;
+    optional google.protobuf.Timestamp min_execution_timestamp = 3;
     // [optional] If specified, the instruction will be rejected if the current timestamp
     // is greater than the specified value. After `max_execution_timestamp` is in the past,
     // it will no longer be possible to execute this instruction.
-    optional google.protobuf.Timestamp max_execution_timestamp = 3;
+    optional google.protobuf.Timestamp max_execution_timestamp = 4;
     // [required] Sequence number of this instruction. It must be greater than 0.
     // It must always be increasing, but not required to be
     // strictly sequential (i.e. gaps are allowed). Each shard separately keeps track of the last executed
@@ -35,7 +37,7 @@ message GovernanceInstruction {
     // rejected (e.g. if instruction #3 has been successfully processed before instruction #2 was observed,
     // #2 will always be rejected).
     // Sequence numbers are assigned and tracked separately for each governance source.
-    optional uint32 governance_sequence_no = 4;
+    optional uint32 governance_sequence_no = 5;
 }
 
 // Specifies which shards the governance instruction applies to.
@@ -160,7 +162,12 @@ message GovernanceSource {
     }
 }
 
-// Create a new shard. Shard name will be determined by the value of `GovernanceDirective.filter`.
+// Create a new shard. A shard is a partially isolated part of Lazer that has its own state and
+// cannot be directly influenced by other shards. The main purpose of shards in Lazer is
+// to allow horizontal scaling when the number of feeds grows. Feeds can be divided into subsets
+// and each subset will be assigned to a shard.
+//
+// Shard name will be determined by the value of `GovernanceDirective.filter`.
 // This action will be rejected unless `GovernanceDirective.filter` specified a single shard.
 // Shard name must be unique across all shards in all groups.
 // (Warning: it's not possible to enforce this rule within a shard!)

lazer/publisher_sdk/proto/governance_instruction.proto

@@ -7,7 +7,12 @@ license = "Apache-2.0"
 repository = "https://github.com/pyth-network/pyth-crosschain"
 
 [dependencies]
+pyth-lazer-protocol = { version = "0.7.2", path = "../../sdk/rust/protocol" }
+anyhow = "1.0.98"
 protobuf = "3.7.2"
+serde-value = "0.7.0"
+humantime = "2.2.0"
+tracing = "0.1.41"
 
 [build-dependencies]
 fs-err = "3.1.0"

lazer/publisher_sdk/rust/Cargo.toml

@@ -1,3 +1,11 @@
+use std::{collections::BTreeMap, time::Duration};
+
+use ::protobuf::MessageField;
+use anyhow::{bail, ensure, Context};
+use humantime::format_duration;
+use protobuf::dynamic_value::{dynamic_value, DynamicValue};
+use pyth_lazer_protocol::router::TimestampUs;
+
 pub mod transaction {
     pub use crate::protobuf::pyth_lazer_transaction::*;
 }
@@ -13,3 +21,137 @@ pub mod governance_instruction {
 mod protobuf {
     include!(concat!(env!("OUT_DIR"), "/protobuf/mod.rs"));
 }
+
+impl DynamicValue {
+    pub fn try_option_from_serde(value: serde_value::Value) -> anyhow::Result<Option<Self>> {
+        match value {
+            serde_value::Value::Option(value) => {
+                if let Some(value) = value {
+                    Ok(Some((*value).try_into()?))
+                } else {
+                    Ok(None)
+                }
+            }
+            value => Ok(Some(value.try_into()?)),
+        }
+    }
+
+    pub fn to_timestamp(&self) -> anyhow::Result<TimestampUs> {
+        let value = self.value.as_ref().context("missing DynamicValue.value")?;
+        match value {
+            dynamic_value::Value::TimestampValue(ts) => Ok(ts.try_into()?),
+            _ => bail!("expected timestamp, got {:?}", self),
+        }
+    }
+
+    pub fn to_duration(&self) -> anyhow::Result<Duration> {
+        let value = self.value.as_ref().context("missing DynamicValue.value")?;
+        match value {
+            dynamic_value::Value::DurationValue(duration) => Ok(duration.clone().into()),
+            _ => bail!("expected duration, got {:?}", self),
+        }
+    }
+}
+
+impl TryFrom<serde_value::Value> for DynamicValue {
+    type Error = anyhow::Error;
+
+    fn try_from(value: serde_value::Value) -> Result<Self, Self::Error> {
+        let converted = match value {
+            serde_value::Value::Bool(value) => dynamic_value::Value::BoolValue(value),
+            serde_value::Value::U8(value) => dynamic_value::Value::UintValue(value.into()),
+            serde_value::Value::U16(value) => dynamic_value::Value::UintValue(value.into()),
+            serde_value::Value::U32(value) => dynamic_value::Value::UintValue(value.into()),
+            serde_value::Value::U64(value) => dynamic_value::Value::UintValue(value),
+            serde_value::Value::I8(value) => dynamic_value::Value::IntValue(value.into()),
+            serde_value::Value::I16(value) => dynamic_value::Value::IntValue(value.into()),
+            serde_value::Value::I32(value) => dynamic_value::Value::IntValue(value.into()),
+            serde_value::Value::I64(value) => dynamic_value::Value::IntValue(value),
+            serde_value::Value::F32(value) => dynamic_value::Value::DoubleValue(value.into()),
+            serde_value::Value::F64(value) => dynamic_value::Value::DoubleValue(value),
+            serde_value::Value::Char(value) => dynamic_value::Value::StringValue(value.to_string()),
+            serde_value::Value::String(value) => dynamic_value::Value::StringValue(value),
+            serde_value::Value::Bytes(value) => dynamic_value::Value::BytesValue(value),
+            serde_value::Value::Seq(values) => {
+                let mut items = Vec::new();
+                for value in values {
+                    items.push(value.try_into()?);
+                }
+                dynamic_value::Value::List(dynamic_value::List {
+                    items,
+                    special_fields: Default::default(),
+                })
+            }
+            serde_value::Value::Map(values) => {
+                let mut items = Vec::new();
+                for (key, value) in values {
+                    let key = match key {
+                        serde_value::Value::String(key) => key,
+                        _ => bail!("unsupported key type: expected string, got {:?}", key),
+                    };
+                    items.push(dynamic_value::MapItem {
+                        key: Some(key),
+                        value: MessageField::some(value.try_into()?),
+                        special_fields: Default::default(),
+                    })
+                }
+                dynamic_value::Value::Map(dynamic_value::Map {
+                    items,
+                    special_fields: Default::default(),
+                })
+            }
+            serde_value::Value::Unit
+            | serde_value::Value::Option(_)
+            | serde_value::Value::Newtype(_) => bail!("unsupported type: {:?}", value),
+        };
+        Ok(DynamicValue {
+            value: Some(converted),
+            special_fields: Default::default(),
+        })
+    }
+}
+
+impl TryFrom<DynamicValue> for serde_value::Value {
+    type Error = anyhow::Error;
+
+    fn try_from(value: DynamicValue) -> Result<Self, Self::Error> {
+        let value = value.value.context("missing DynamicValue.value")?;
+        match value {
+            dynamic_value::Value::StringValue(value) => Ok(serde_value::Value::String(value)),
+            dynamic_value::Value::DoubleValue(value) => Ok(serde_value::Value::F64(value)),
+            dynamic_value::Value::UintValue(value) => Ok(serde_value::Value::U64(value)),
+            dynamic_value::Value::IntValue(value) => Ok(serde_value::Value::I64(value)),
+            dynamic_value::Value::BoolValue(value) => Ok(serde_value::Value::Bool(value)),
+            dynamic_value::Value::BytesValue(value) => Ok(serde_value::Value::Bytes(value)),
+            dynamic_value::Value::DurationValue(duration) => {
+                let s: Duration = duration.into();
+                Ok(serde_value::Value::String(format_duration(s).to_string()))
+            }
+            dynamic_value::Value::TimestampValue(ts) => {
+                let ts = TimestampUs::try_from(&ts)?;
+                Ok(serde_value::Value::U64(ts.0))
+            }
+            dynamic_value::Value::List(list) => {
+                let mut output = Vec::new();
+                for item in list.items {
+                    output.push(item.try_into()?);
+                }
+                Ok(serde_value::Value::Seq(output).into())
+            }
+            dynamic_value::Value::Map(map) => {
+                let mut output = BTreeMap::new();
+                for item in map.items {
+                    let key = item.key.context("missing DynamicValue.MapItem.key")?;
+                    let value = item
+                        .value
+                        .into_option()
+                        .context("missing DynamicValue.MapItem.value")?
+                        .try_into()?;
+                    let old = output.insert(serde_value::Value::String(key), value);
+                    ensure!(old.is_none(), "duplicate DynamicValue.MapItem.key");
+                }
+                Ok(serde_value::Value::Map(output).into())
+            }
+        }
+    }
+}