feat: add serializers for pyth formats

feat: use pythnet serialization in hermes

Fix vaa validation

Clippy

Update config names

Wrap Store with Arc

Store works perfectly without Arc as all it's elements are behind an Arc
or something similar to that, however a developer might make
mistake to add a field and missing it.

Improve error handling

Update metadata struct

Add metadata

Update Eth listener

Pin wormhole to a version

Fix ws dispatcher

fix: blocking in go recv corrupts tokio runtime

Make network <> store message passing non-blocking

Update logs and revert debug changes

Author: Reisen

hermes/.gitignore

@@ -7,4 +7,4 @@ src/network/p2p.proto
 tools/
 
 # Ignore Wormhole cloned repo
-wormhole/
+wormhole*/

hermes/Cargo.lock

@@ -69,6 +69,8 @@ pyth-oracle = { git = "https://github.com/pyth-network/pyth-client", rev = "7d59
 
 strum = { version = "0.24", features = ["derive"] }
 ethabi = { version = "18.0.0", features = ["serde"] }
+sha3 = "0.10.4"
+humantime = "2.1.0"
 
 [patch.crates-io]
 serde_wormhole = { git = "https://github.com/wormhole-foundation/wormhole", tag = "v2.17.1" }

hermes/Cargo.toml

@@ -1,63 +1,79 @@
 use std::{
     env,
     path::PathBuf,
-    process::Command,
+    process::{
+        Command,
+        Stdio,
+    },
 };
 
 fn main() {
     let out_dir = PathBuf::from(env::var("OUT_DIR").unwrap());
     let out_var = env::var("OUT_DIR").unwrap();
 
-    // Clone the Wormhole repository, which we need to access the protobuf definitions for Wormhole
-    // P2P message types.
+    // Download the Wormhole repository at a certain tag, which we need to access the protobuf definitions
+    // for Wormhole P2P message types.
     //
-    // TODO: This is ugly and costly, and requires git. Instead of this we should have our own tool
+    // TODO: This is ugly. Instead of this we should have our own tool
     // build process that can generate protobuf definitions for this and other user cases. For now
     // this is easy and works and matches upstream Wormhole's `Makefile`.
-    let _ = Command::new("git")
+
+    const WORMHOLE_VERSION: &str = "2.18.1";
+
+    let wh_curl = Command::new("curl")
         .args([
-            "clone",
-            "https://github.com/wormhole-foundation/wormhole",
-            "wormhole",
+            "-s",
+            "-L",
+            format!("https://github.com/wormhole-foundation/wormhole/archive/refs/tags/v{WORMHOLE_VERSION}.tar.gz").as_str(),
         ])
+        .stdout(Stdio::piped())
+        .spawn()
+        .expect("failed to download wormhole archive");
+
+    let _ = Command::new("tar")
+        .args(["xvz"])
+        .stdin(Stdio::from(wh_curl.stdout.unwrap()))
         .output()
-        .expect("failed to execute process");
+        .expect("failed to extract wormhole archive");
 
     // Move the tools directory to the root of the repo because that's where the build script
     // expects it to be, paths get hardcoded into the binaries.
     let _ = Command::new("mv")
-        .args(["wormhole/tools", "tools"])
+        .args([
+            format!("wormhole-{WORMHOLE_VERSION}/tools").as_str(),
+            "tools",
+        ])
         .output()
-        .expect("failed to execute process");
+        .expect("failed to move wormhole tools directory");
 
     // Move the protobuf definitions to the src/network directory, we don't have to do this
     // but it is more intuitive when debugging.
     let _ = Command::new("mv")
         .args([
-            "wormhole/proto/gossip/v1/gossip.proto",
+            format!("wormhole-{WORMHOLE_VERSION}/proto/gossip/v1/gossip.proto").as_str(),
             "src/network/p2p.proto",
         ])
         .output()
-        .expect("failed to execute process");
+        .expect("failed to move wormhole protobuf definitions");
 
     // Build the protobuf compiler.
     let _ = Command::new("./build.sh")
         .current_dir("tools")
         .output()
-        .expect("failed to execute process");
+        .expect("failed to run protobuf compiler build script");
 
     // Make the protobuf compiler executable.
     let _ = Command::new("chmod")
         .args(["+x", "tools/bin/*"])
         .output()
-        .expect("failed to execute process");
+        .expect("failed to make protofuf compiler executable");
 
     // Generate the protobuf definitions. See buf.gen.yaml to see how we rename the module for our
     // particular use case.
     let _ = Command::new("./tools/bin/buf")
         .args(["generate", "--path", "src"])
         .output()
-        .expect("failed to execute process");
+        .expect("failed to generate protobuf definitions");
 
     // Build the Go library.
     let mut cmd = Command::new("go");

hermes/build.rs

@@ -7,7 +7,7 @@ with pkgs; mkShell {
     clang
     llvmPackages.libclang
     nettle
-    openssl
+    openssl_1_1
     pkgconfig
     rustup
     systemd

hermes/shell.nix

@@ -7,6 +7,10 @@ use {
         Router,
     },
     std::sync::Arc,
+    tokio::{
+        signal,
+        sync::mpsc::Receiver,
+    },
 };
 
 mod rest;
@@ -15,12 +19,12 @@ mod ws;
 
 #[derive(Clone)]
 pub struct State {
-    pub store: Store,
+    pub store: Arc<Store>,
     pub ws:    Arc<ws::WsState>,
 }
 
 impl State {
-    pub fn new(store: Store) -> Self {
+    pub fn new(store: Arc<Store>) -> Self {
         Self {
             store,
             ws: Arc::new(ws::WsState::new()),
@@ -32,7 +36,7 @@ impl State {
 ///
 /// Currently this is based on Axum due to the simplicity and strong ecosystem support for the
 /// packages they are based on (tokio & hyper).
-pub async fn spawn(store: Store, rpc_addr: String) -> Result<()> {
+pub async fn run(store: Arc<Store>, mut update_rx: Receiver<()>, rpc_addr: String) -> Result<()> {
     let state = State::new(store);
 
     // Initialize Axum Router. Note the type here is a `Router<State>` due to the use of the
@@ -50,28 +54,31 @@ pub async fn spawn(store: Store, rpc_addr: String) -> Result<()> {
         .with_state(state.clone());
 
 
-    // Binds the axum's server to the configured address and port. This is a blocking call and will
-    // not return until the server is shutdown.
-    tokio::spawn(async move {
-        // FIXME handle errors properly
-        axum::Server::bind(&rpc_addr.parse().unwrap())
-            .serve(app.into_make_service())
-            .await
-            .unwrap();
-    });
-
     // Call dispatch updates to websocket every 1 seconds
     // FIXME use a channel to get updates from the store
     tokio::spawn(async move {
         loop {
-            dispatch_updates(
-                state.store.get_price_feed_ids().into_iter().collect(),
-                state.clone(),
-            )
-            .await;
-            tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+            // Panics if the update channel is closed, which should never happen.
+            // If it happens we have no way to recover, so we just panic.
+            update_rx
+                .recv()
+                .await
+                .expect("state update channel is closed");
+
+            dispatch_updates(state.clone()).await;
         }
     });
 
+    // Binds the axum's server to the configured address and port. This is a blocking call and will
+    // not return until the server is shutdown.
+    axum::Server::try_bind(&rpc_addr.parse()?)?
+        .serve(app.into_make_service())
+        .with_graceful_shutdown(async {
+            signal::ctrl_c()
+                .await
+                .expect("Ctrl-c signal handler failed.");
+        })
+        .await?;
+
     Ok(())
 }

hermes/src/api.rs

@@ -36,6 +36,7 @@ use {
 pub enum RestError {
     UpdateDataNotFound,
     CcipUpdateDataNotFound,
+    InvalidCCIPInput,
 }
 
 impl IntoResponse for RestError {
@@ -54,6 +55,9 @@ impl IntoResponse for RestError {
 
                 (StatusCode::BAD_GATEWAY, "CCIP update data not found").into_response()
             }
+            RestError::InvalidCCIPInput => {
+                (StatusCode::BAD_REQUEST, "Invalid CCIP input").into_response()
+            }
         }
     }
 }
@@ -113,7 +117,7 @@ pub async fn latest_price_feeds(
             .price_feeds
             .into_iter()
             .map(|price_feed| {
-                RpcPriceFeed::from_price_feed_message(price_feed, params.verbose, params.binary)
+                RpcPriceFeed::from_price_feed_update(price_feed, params.verbose, params.binary)
             })
             .collect(),
     ))
@@ -156,7 +160,8 @@ pub async fn get_vaa(
         .price_feeds
         .get(0)
         .ok_or(RestError::UpdateDataNotFound)?
-        .publish_time; // TODO: This should never happen.
+        .price_feed
+        .publish_time;
 
     Ok(Json(GetVaaResponse { vaa, publish_time }))
 }
@@ -179,8 +184,16 @@ pub async fn get_vaa_ccip(
     State(state): State<super::State>,
     QsQuery(params): QsQuery<GetVaaCcipQueryParams>,
 ) -> Result<Json<GetVaaCcipResponse>, RestError> {
-    let price_id: PriceIdentifier = PriceIdentifier::new(params.data[0..32].try_into().unwrap());
-    let publish_time = UnixTimestamp::from_be_bytes(params.data[32..40].try_into().unwrap());
+    let price_id: PriceIdentifier = PriceIdentifier::new(
+        params.data[0..32]
+            .try_into()
+            .map_err(|_| RestError::InvalidCCIPInput)?,
+    );
+    let publish_time = UnixTimestamp::from_be_bytes(
+        params.data[32..40]
+            .try_into()
+            .map_err(|_| RestError::InvalidCCIPInput)?,
+    );
 
     let price_feeds_with_update_data = state
         .store

hermes/src/api/rest.rs

@@ -1,17 +1,25 @@
 use {
     crate::{
         impl_deserialize_for_hex_string_wrapper,
-        store::types::UnixTimestamp,
+        store::types::{
+            PriceFeedUpdate,
+            Slot,
+            UnixTimestamp,
+        },
+    },
+    base64::{
+        engine::general_purpose::STANDARD as base64_standard_engine,
+        Engine as _,
     },
     derive_more::{
         Deref,
         DerefMut,
     },
-    pyth_oracle::PriceFeedMessage,
     pyth_sdk::{
         Price,
         PriceIdentifier,
     },
+    wormhole_sdk::Chain,
 };
 
 
@@ -34,8 +42,8 @@ type Base64String = String;
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct RpcPriceFeedMetadata {
+    pub slot:                       Slot,
     pub emitter_chain:              u16,
-    pub sequence_number:            u64,
     pub price_service_receive_time: UnixTimestamp,
 }
 
@@ -54,11 +62,13 @@ pub struct RpcPriceFeed {
 impl RpcPriceFeed {
     // TODO: Use a Encoding type to have None, Base64, and Hex variants instead of binary flag.
     // TODO: Use a Verbosity type to define None, or Full instead of verbose flag.
-    pub fn from_price_feed_message(
-        price_feed_message: PriceFeedMessage,
-        _verbose: bool,
-        _binary: bool,
+    pub fn from_price_feed_update(
+        price_feed_update: PriceFeedUpdate,
+        verbose: bool,
+        binary: bool,
     ) -> Self {
+        let price_feed_message = price_feed_update.price_feed;
+
         Self {
             id:        PriceIdentifier::new(price_feed_message.id),
             price:     Price {
@@ -73,16 +83,14 @@ impl RpcPriceFeed {
                 expo:         price_feed_message.exponent,
                 publish_time: price_feed_message.publish_time,
             },
-            // FIXME: Handle verbose flag properly.
-            // metadata:  verbose.then_some(RpcPriceFeedMetadata {
-            //     emitter_chain:              price_feed_message.emitter_chain,
-            //     sequence_number:            price_feed_message.sequence_number,
-            //     price_service_receive_time: price_feed_message.receive_time,
-            // }),
-            metadata:  None,
-            // FIXME: The vaa is wrong, fix it
-            // vaa:       binary.then_some(base64_standard_engine.encode(message_state.proof_set.wormhole_merkle_proof.vaa)),
-            vaa:       None,
+            metadata:  verbose.then_some(RpcPriceFeedMetadata {
+                emitter_chain:              Chain::Pythnet.into(),
+                price_service_receive_time: price_feed_update.received_at,
+                slot:                       price_feed_update.slot,
+            }),
+            vaa:       binary.then_some(
+                base64_standard_engine.encode(price_feed_update.wormhole_merkle_update_data),
+            ),
         }
     }
 }