feat(target_chains/fuel): add governance contract (#1518)

* add governance contract

* add fuel ci

* add rust-toolchain

* add executes_governance_instruction test

* add test for SetValidPeriod

* add test for AuthorizeGovernanceDataSourceTransfer

* remove SetWormholeAddress

* add test for SetDataSources

* remove WormholeAddressSetEvent

* remove SetWormholeAddress

* remove SetWormholeAddressPayload

* remove SetWormholeAddressPayload and SetWormholeAddress imports

* remove GovernanceAction::SetWormholeAddress

* address comments

* refactor test

* add comments

.dockerignore

@@ -15,5 +15,4 @@
 
 .git
 
-hermes/wormhole
-!hermes/src/state/cache.rs
+!apps/hermes/src/state/cache.rs

.github/workflows/ci-aptos-contract.yml

@@ -21,10 +21,10 @@ jobs:
       - uses: actions/checkout@v3
 
       - name: Download CLI
-        run: wget https://github.com/aptos-labs/aptos-core/releases/download/aptos-cli-v1.0.4/aptos-cli-1.0.4-Ubuntu-22.04-x86_64.zip
+        run: wget https://github.com/aptos-labs/aptos-core/releases/download/aptos-cli-v3.1.0/aptos-cli-3.1.0-Ubuntu-22.04-x86_64.zip
 
       - name: Unzip CLI
-        run: unzip aptos-cli-1.0.4-Ubuntu-22.04-x86_64.zip
+        run: unzip aptos-cli-3.1.0-Ubuntu-22.04-x86_64.zip
 
       - name: Run tests
         run: ./aptos move test

.github/workflows/ci-fortuna.yml

@@ -2,10 +2,10 @@ name: Check Fortuna
 
 on:
   pull_request:
-    paths: [fortuna/**]
+    paths: [apps/fortuna/**]
   push:
     branches: [main]
-    paths: [fortuna/**]
+    paths: [apps/fortuna/**]
 jobs:
   test:
     runs-on: ubuntu-latest
@@ -17,4 +17,4 @@ jobs:
           toolchain: nightly-2023-07-23
           override: true
       - name: Run executor tests
-        run: cargo test --manifest-path ./fortuna/Cargo.toml
+        run: cargo test --manifest-path ./apps/fortuna/Cargo.toml

.github/workflows/ci-fuel-contract.yml

@@ -0,0 +1,35 @@
+name: Test Fuel Contract
+
+on:
+  pull_request:
+    paths:
+      - target_chains/fuel/**
+  push:
+    branches:
+      - main
+    paths:
+      - target_chains/fuel/**
+
+env:
+  CARGO_TERM_COLOR: always
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: target_chains/fuel/contracts/
+    steps:
+      - uses: actions/checkout@v2
+      - name: Install Fuel toolchain
+        run: |
+          curl https://install.fuel.network | sh
+          echo "$HOME/.fuelup/bin" >> $GITHUB_PATH
+      - name: Build with Forc
+        run: forc build --verbose
+      - name: Run tests with Forc
+        run: forc test --verbose
+      - name: Build
+        run: cargo build --verbose
+      - name: Run tests
+        run: cargo test --verbose

.github/workflows/ci-hermes.yml

@@ -2,10 +2,10 @@ name: Check Hermes
 
 on:
   pull_request:
-    paths: [hermes/**]
+    paths: [apps/hermes/**]
   push:
     branches: [main]
-    paths: [hermes/**]
+    paths: [apps/hermes/**]
 jobs:
   test:
     runs-on: ubuntu-latest
@@ -20,4 +20,4 @@ jobs:
       - name: Install protoc
         uses: arduino/setup-protoc@v3
       - name: Run executor tests
-        run: cargo test --manifest-path ./hermes/Cargo.toml
+        run: cargo test --manifest-path ./apps/hermes/Cargo.toml

.github/workflows/ci-starknet-contract.yml

@@ -0,0 +1,37 @@
+name: Starknet contract
+on:
+  pull_request:
+    paths:
+      - target_chains/starknet/contracts/**
+  push:
+    branches:
+      - main
+    paths:
+      - target_chains/starknet/contracts/**
+jobs:
+  check:
+    name: Starknet Foundry tests
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: target_chains/starknet/contracts/
+    steps:
+      - uses: actions/checkout@v3
+      - name: Install Scarb
+        uses: software-mansion/setup-scarb@v1
+        with:
+          tool-versions: target_chains/starknet/contracts/.tool-versions
+      - name: Install Starknet Foundry
+        uses: foundry-rs/setup-snfoundry@v3
+        with:
+          tool-versions: target_chains/starknet/contracts/.tool-versions
+      - name: Install Starkli
+        run: curl https://get.starkli.sh | sh && . ~/.config/.starkli/env && starkliup -v $(awk '/starkli/{print $2}' .tool-versions)
+      - name: Install Katana
+        run: curl -L https://install.dojoengine.org | bash && PATH="$PATH:$HOME/.config/.dojo/bin" dojoup -v $(awk '/dojo/{print $2}' .tool-versions)
+      - name: Check formatting
+        run: scarb fmt --check
+      - name: Run tests
+        run: snforge test
+      - name: Test local deployment script
+        run: bash -c 'PATH="$PATH:$HOME/.config/.dojo/bin" katana & . ~/.config/.starkli/env && deploy/local_deploy'

.github/workflows/publish-js.yml

@@ -12,7 +12,7 @@ jobs:
       - uses: actions/checkout@v2
       - uses: actions/setup-node@v2
         with:
-          node-version: "16"
+          node-version: "18"
           registry-url: "https://registry.npmjs.org"
       - run: npm ci
       - run: npx lerna run build --no-private

.github/workflows/publish-pyth-solana-receiver-state.yml

@@ -11,8 +11,14 @@ jobs:
     steps:
       - name: Checkout sources
         uses: actions/checkout@v2
+      - name: Install Rust
+        uses: actions-rs/toolchain@v1
+        with:
+          toolchain: stable
+          default: true
+          profile: minimal
 
-      - run: cargo publish --token ${CARGO_REGISTRY_TOKEN}
+      - run: cargo +stable-x86_64-unknown-linux-gnu publish --token ${CARGO_REGISTRY_TOKEN}
         env:
           CARGO_REGISTRY_TOKEN: ${{ secrets.CARGO_REGISTRY_TOKEN }}
         working-directory: "target_chains/solana/pyth_solana_receiver_sdk"

.github/workflows/push-fortuna-image.yml

@@ -46,7 +46,7 @@ jobs:
         uses: docker/build-push-action@f2a1d5e99d037542a71f64918e516c093c6f3fc4
         with:
           context: .
-          file: "./fortuna/Dockerfile"
+          file: "./apps/fortuna/Dockerfile"
           push: true
           tags: ${{ steps.metadata_fortuna.outputs.tags }}
           labels: ${{ steps.metadata_fortuna.outputs.labels }}

.github/workflows/push-hermes-image.yml

@@ -37,7 +37,7 @@ jobs:
         env:
           AWS_REGION: us-east-1
       - run: |
-          DOCKER_BUILDKIT=1 docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG -f hermes/Dockerfile .
+          DOCKER_BUILDKIT=1 docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG -f apps/hermes/Dockerfile .
           docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
         env:
           ECR_REGISTRY: public.ecr.aws

.github/workflows/push-price-pusher-image.yml

@@ -40,7 +40,7 @@ jobs:
         id: ecr_login
       - run: |
           DOCKER_BUILDKIT=1 docker build -t lerna -f Dockerfile.lerna .
-          DOCKER_BUILDKIT=1 docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG -f price_pusher/Dockerfile .
+          DOCKER_BUILDKIT=1 docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG -f apps/price_pusher/Dockerfile .
           docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
         env:
           ECR_REGISTRY: public.ecr.aws

.github/workflows/pypi-express-relay-utils.yml

@@ -5,12 +5,17 @@ on:
     tags:
       - "python-v*"
 
+env:
+  PYTHON_VERSION: "3.11"
+
 jobs:
   deploy:
     runs-on: ubuntu-20.04
     steps:
       - uses: actions/checkout@v2
       - uses: actions/setup-python@v2
+        with:
+          python-version: ${{ env.PYTHON_VERSION }}
       - name: Install dependencies
         run: |
           python3 -m pip install --upgrade poetry

.npmrc

@@ -0,0 +1 @@
+engine-strict=true

.pre-commit-config.yaml

@@ -47,22 +47,22 @@ repos:
       - id: cargo-fmt-hermes
         name: Cargo format for Hermes
         language: "rust"
-        entry: cargo +nightly-2024-03-26 fmt --manifest-path ./hermes/Cargo.toml --all -- --config-path rustfmt.toml
+        entry: cargo +nightly-2024-03-26 fmt --manifest-path ./apps/hermes/Cargo.toml --all -- --config-path rustfmt.toml
         pass_filenames: false
-        files: hermes
+        files: apps/hermes
       - id: cargo-clippy-hermes
         name: Cargo clippy for Hermes
         language: "rust"
-        entry: cargo +nightly-2024-03-26 clippy --manifest-path ./hermes/Cargo.toml --tests --fix --allow-dirty --allow-staged -- -D warnings
+        entry: cargo +nightly-2024-03-26 clippy --manifest-path ./apps/hermes/Cargo.toml --tests --fix --allow-dirty --allow-staged -- -D warnings
         pass_filenames: false
-        files: hermes
+        files: apps/hermes
       # Hooks for Fortuna
       - id: cargo-fmt-fortuna
         name: Cargo format for Fortuna
         language: "rust"
-        entry: cargo +nightly-2023-07-23 fmt --manifest-path ./fortuna/Cargo.toml --all -- --config-path rustfmt.toml
+        entry: cargo +nightly-2023-07-23 fmt --manifest-path ./apps/fortuna/Cargo.toml --all -- --config-path rustfmt.toml
         pass_filenames: false
-        files: fortuna
+        files: apps/fortuna
       # Hooks for message buffer contract
       - id: cargo-fmt-message-buffer
         name: Cargo format for message buffer contract
@@ -80,13 +80,13 @@ repos:
       - id: cargo-fmt-pythnet-sdk
         name: Cargo format for pythnet SDK
         language: "rust"
-        entry: cargo +nightly-2023-07-23 fmt --manifest-path ./pythnet/pythnet_sdk/Cargo.toml --all -- --config-path rustfmt.toml
+        entry: cargo +nightly-2024-03-26 fmt --manifest-path ./pythnet/pythnet_sdk/Cargo.toml --all -- --config-path rustfmt.toml
         pass_filenames: false
         files: pythnet/pythnet_sdk
       - id: cargo-clippy-pythnet-sdk
         name: Cargo clippy for pythnet SDK
         language: "rust"
-        entry: cargo +nightly-2023-07-23 clippy --manifest-path ./pythnet/pythnet_sdk/Cargo.toml --tests --fix --allow-dirty --allow-staged -- -D warnings
+        entry: cargo +nightly-2024-03-26 clippy --manifest-path ./pythnet/pythnet_sdk/Cargo.toml --tests --fix --allow-dirty --allow-staged -- -D warnings
         pass_filenames: false
         files: pythnet/pythnet_sdk
       # Hooks for solana receiver contract

README.md

@@ -16,7 +16,7 @@ contracts, SDKs, and examples.
 
 ## Hermes
 
-> [hermes](./hermes/)
+> [hermes](./apps/hermes/)
 
 Hermes is an off-chain service which constantly observes Pythnet and the
 Wormhole network watching for price updates emitted from the Pyth contract. It
@@ -79,10 +79,11 @@ Lerna has some common failure modes that you may encounter:
 1. `npm ci` fails with a typescript compilation error about a missing package.
    This error likely means that the failing package has a `prepare` entry compiling the typescript in its `package.json`.
    Fix this error by moving that logic to the `prepublishOnly` entry.
-1. The software builds locally but fails in CI, or vice-versa.
+2. The software builds locally but fails in CI, or vice-versa.
    This error likely means that some local build caches need to be cleaned.
    The build error may not indicate that this is a caching issue, e.g., it may appear that the packages are being built in the wrong order.
    Delete `node_modules/`, `lib/` and `tsconfig.tsbuildinfo` from each package's subdirectory. then try again.
+3. `npm ci` fails due to wrong node version. Make sure to be using `v18`. Node version `v21` is not supported and known to cause issues.
 
 ## Audit / Feature Status
 

apps/fortuna/.gitignore

@@ -1,4 +1,4 @@
 /target
-config.yaml
+*config.yaml
 *secret*
 *private-key*

apps/fortuna/Cargo.lock

@@ -1488,7 +1488,7 @@ dependencies = [
 
 [[package]]
 name = "fortuna"
-version = "4.0.0"
+version = "5.2.2"
 dependencies = [
  "anyhow",
  "axum",
@@ -2822,7 +2822,7 @@ dependencies = [
 
 [[package]]
 name = "pythnet-sdk"
-version = "2.0.0"
+version = "2.1.0"
 dependencies = [
  "bincode",
  "borsh",

apps/fortuna/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name    = "fortuna"
-version = "4.0.0"
+version = "5.2.2"
 edition = "2021"
 
 [dependencies]
@@ -14,9 +14,9 @@ clap        = { version = "4.4.6", features = ["derive", "cargo", "env"] }
 ethabi      = "18.0.0"
 ethers      = { version = "2.0.14", features = ["ws"] }
 futures            = { version = "0.3.28" }
-hex         = "0.4.3"
+hex = "0.4.3"
 prometheus-client  = { version = "0.21.2" }
-pythnet-sdk = { path = "../pythnet/pythnet_sdk", features = ["strum"] }
+pythnet-sdk = { path = "../../pythnet/pythnet_sdk", features = ["strum"] }
 rand        = "0.8.5"
 reqwest     = { version = "0.11.22", features = ["json", "blocking"] }
 serde       = { version = "1.0.188", features = ["derive"] }

apps/fortuna/Dockerfile

@@ -7,15 +7,15 @@ RUN rustup default nightly-2023-07-23
 
 # Build
 WORKDIR /src
-COPY fortuna fortuna
+COPY apps/fortuna apps/fortuna
 COPY pythnet pythnet
 COPY target_chains/ethereum/entropy_sdk/solidity/abis target_chains/ethereum/entropy_sdk/solidity/abis
 
-WORKDIR /src/fortuna
+WORKDIR /src/apps/fortuna
 
 RUN --mount=type=cache,target=/root/.cargo/registry cargo build --release
 
 
 FROM rust:${RUST_VERSION}
 # Copy artifacts from other images
-COPY --from=build /src/fortuna/target/release/fortuna /usr/local/bin/
+COPY --from=build /src/apps/fortuna/target/release/fortuna /usr/local/bin/

apps/fortuna/config.sample.yaml

@@ -4,3 +4,4 @@ chains:
     contract_addr: 0x8250f4aF4B972684F7b336503E2D6dFeDeB1487a
     reveal_delay_blocks: 0
     legacy_tx: true
+    gas_limit: 500000

apps/fortuna/provider-config.sample.yaml

@@ -0,0 +1,7 @@
+chains:
+  lightlink-pegasus:
+    commitments:
+      # prettier-ignore
+      - seed: [219,125,217,197,234,88,208,120,21,181,172,143,239,102,41,233,167,212,237,106,37,255,184,165,238,121,230,155,116,158,173,48]
+        chain_length: 10000
+        original_commitment_sequence_number: 104

apps/fortuna/rust-toolchain

@@ -0,0 +1 @@
+nightly-2023-07-23

apps/fortuna/src/chain/ethereum.rs

@@ -59,7 +59,7 @@ use {
 // contract in the same repo.
 abigen!(
     PythRandom,
-    "../target_chains/ethereum/entropy_sdk/solidity/abis/IEntropy.json"
+    "../../target_chains/ethereum/entropy_sdk/solidity/abis/IEntropy.json"
 );
 
 pub type SignablePythContract = PythRandom<
@@ -277,7 +277,11 @@ impl EntropyReader for PythContract {
             Err(e) => match e {
                 ContractError::ProviderError { e } => Err(anyhow!(e)),
                 _ => {
-                    tracing::info!("Gas estimation for reveal with callback failed: {:?}", e);
+                    tracing::info!(
+                        sequence_number = sequence_number,
+                        "Gas estimation failed. error: {:?}",
+                        e
+                    );
                     Ok(None)
                 }
             },

apps/fortuna/src/command/run.rs

@@ -8,7 +8,9 @@ use {
         chain::ethereum::PythContract,
         command::register_provider::CommitmentMetadata,
         config::{
+            Commitment,
             Config,
+            ProviderConfig,
             RunOptions,
         },
         keeper,
@@ -27,7 +29,6 @@ use {
         collections::HashMap,
         net::SocketAddr,
         sync::Arc,
-        vec,
     },
     tokio::{
         spawn,
@@ -121,38 +122,67 @@ pub async fn run_keeper(
 
 pub async fn run(opts: &RunOptions) -> Result<()> {
     let config = Config::load(&opts.config.config)?;
-    let private_key = opts.load_private_key()?;
+    let provider_config = opts
+        .provider_config
+        .provider_config
+        .as_ref()
+        .map(|path| ProviderConfig::load(&path).expect("Failed to load provider config"));
     let secret = opts.randomness.load_secret()?;
     let (tx_exit, rx_exit) = watch::channel(false);
 
     let mut chains: HashMap<ChainId, BlockchainState> = HashMap::new();
     for (chain_id, chain_config) in &config.chains {
         let contract = Arc::new(PythContract::from_config(&chain_config)?);
-        let provider_info = contract.get_provider_info(opts.provider).call().await?;
+        let provider_chain_config = provider_config
+            .as_ref()
+            .and_then(|c| c.get_chain_config(chain_id));
+        let mut provider_commitments = provider_chain_config
+            .as_ref()
+            .map(|c| c.get_sorted_commitments())
+            .unwrap_or_else(|| Vec::new());
+
+        let provider_info = contract.get_provider_info(opts.provider).call().await?;
+        let latest_metadata =
+            bincode::deserialize::<CommitmentMetadata>(&provider_info.commitment_metadata)
+                .map_err(|e| {
+                    anyhow!(
+                        "Chain: {} - Failed to deserialize commitment metadata: {}",
+                        &chain_id,
+                        e
+                    )
+                })?;
+
+        provider_commitments.push(Commitment {
+            seed:                                latest_metadata.seed,
+            chain_length:                        latest_metadata.chain_length,
+            original_commitment_sequence_number: provider_info.original_commitment_sequence_number,
+        });
 
-        // Reconstruct the hash chain based on the metadata and check that it matches the on-chain commitment.
-        // TODO: we should instantiate the state here with multiple hash chains.
-        // This approach works fine as long as we haven't rotated the commitment (i.e., all user requests
-        // are for the most recent chain).
         // TODO: we may want to load the hash chain in a lazy/fault-tolerant way. If there are many blockchains,
         // then it's more likely that some RPC fails. We should tolerate these faults and generate the hash chain
         // later when a user request comes in for that chain.
-        let metadata =
-            bincode::deserialize::<CommitmentMetadata>(&provider_info.commitment_metadata)?;
 
-        let hash_chain = PebbleHashChain::from_config(
-            &secret,
-            &chain_id,
-            &opts.provider,
-            &chain_config.contract_addr,
-            &metadata.seed,
-            metadata.chain_length,
-        )?;
+        let mut offsets = Vec::<usize>::new();
+        let mut hash_chains = Vec::<PebbleHashChain>::new();
+
+        for commitment in &provider_commitments {
+            let offset = commitment.original_commitment_sequence_number.try_into()?;
+            offsets.push(offset);
+
+            let pebble_hash_chain = PebbleHashChain::from_config(
+                &secret,
+                &chain_id,
+                &opts.provider,
+                &chain_config.contract_addr,
+                &commitment.seed,
+                commitment.chain_length,
+            )?;
+            hash_chains.push(pebble_hash_chain);
+        }
+
         let chain_state = HashChainState {
-            offsets:     vec![provider_info
-                .original_commitment_sequence_number
-                .try_into()?],
-            hash_chains: vec![hash_chain],
+            offsets,
+            hash_chains,
         };
 
         if chain_state.reveal(provider_info.original_commitment_sequence_number)?
@@ -187,7 +217,10 @@ pub async fn run(opts: &RunOptions) -> Result<()> {
 
         Ok::<(), Error>(())
     });
-    spawn(run_keeper(chains.clone(), config, private_key));
+
+    if let Some(keeper_private_key) = opts.load_keeper_private_key()? {
+        spawn(run_keeper(chains.clone(), config, keeper_private_key));
+    }
 
     run_api(opts.addr.clone(), chains, rx_exit).await?;
 

apps/fortuna/src/command/setup_provider.rs

@@ -16,7 +16,10 @@ use {
             PebbleHashChain,
         },
     },
-    anyhow::Result,
+    anyhow::{
+        anyhow,
+        Result,
+    },
     ethers::{
         abi::Bytes as AbiBytes,
         signers::{
@@ -66,7 +69,14 @@ pub async fn setup_provider(opts: &SetupProviderOptions) -> Result<()> {
             register = true;
         } else {
             let metadata =
-                bincode::deserialize::<CommitmentMetadata>(&provider_info.commitment_metadata)?;
+                bincode::deserialize::<CommitmentMetadata>(&provider_info.commitment_metadata)
+                    .map_err(|e| {
+                        anyhow!(
+                            "Chain: {} - Failed to deserialize commitment metadata: {}",
+                            &chain_id,
+                            e
+                        )
+                    })?;
 
             let hash_chain = PebbleHashChain::from_config(
                 &secret,
@@ -74,7 +84,7 @@ pub async fn setup_provider(opts: &SetupProviderOptions) -> Result<()> {
                 &provider_address,
                 &chain_config.contract_addr,
                 &metadata.seed,
-                metadata.chain_length,
+                opts.randomness.chain_length,
             )?;
             let chain_state = HashChainState {
                 offsets:     vec![provider_info
@@ -105,7 +115,8 @@ pub async fn setup_provider(opts: &SetupProviderOptions) -> Result<()> {
                 fee: opts.fee,
                 uri,
             })
-            .await?;
+            .await
+            .map_err(|e| anyhow!("Chain: {} - Failed to register provider: {}", &chain_id, e))?;
             tracing::info!("{}: registered", &chain_id);
         } else {
             if provider_info.fee_in_wei != opts.fee {

apps/fortuna/src/config.rs

@@ -97,7 +97,7 @@ pub struct RandomnessOptions {
     /// The length of the hash chain to generate.
     #[arg(long = "chain-length")]
     #[arg(env = "FORTUNA_CHAIN_LENGTH")]
-    #[arg(default_value = "10000")]
+    #[arg(default_value = "100000")]
     pub chain_length: u64,
 }
 
@@ -158,3 +158,57 @@ pub struct EthereumConfig {
     /// The gas limit to use for entropy callback transactions.
     pub gas_limit: U256,
 }
+
+#[derive(Args, Clone, Debug)]
+#[command(next_help_heading = "Provider Config Options")]
+#[group(id = "ProviderConfig")]
+pub struct ProviderConfigOptions {
+    #[arg(long = "provider-config")]
+    #[arg(env = "FORTUNA_PROVIDER_CONFIG")]
+    pub provider_config: Option<String>,
+}
+
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct ProviderConfig {
+    pub chains: HashMap<ChainId, ProviderChainConfig>,
+}
+
+impl ProviderConfig {
+    pub fn load(path: &str) -> Result<ProviderConfig> {
+        // Open and read the YAML file
+        let yaml_content = fs::read_to_string(path)?;
+        let config: ProviderConfig = serde_yaml::from_str(&yaml_content)?;
+        Ok(config)
+    }
+
+    /// Get the provider chain config. The method returns an Option for ProviderChainConfig.
+    /// We may not have past any commitments for a chain. For example, for a new chain
+    pub fn get_chain_config(&self, chain_id: &ChainId) -> Option<ProviderChainConfig> {
+        self.chains.get(chain_id).map(|x| x.clone())
+    }
+}
+
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct ProviderChainConfig {
+    commitments: Vec<Commitment>,
+}
+
+impl ProviderChainConfig {
+    /// Returns a clone of the commitments in the sorted order.
+    /// `HashChainState`  requires offsets to be in order.
+    pub fn get_sorted_commitments(&self) -> Vec<Commitment> {
+        let mut sorted_commitments = self.commitments.clone();
+        sorted_commitments.sort_by(|c1, c2| {
+            c1.original_commitment_sequence_number
+                .cmp(&c2.original_commitment_sequence_number)
+        });
+        sorted_commitments
+    }
+}
+
+#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
+pub struct Commitment {
+    pub seed:                                [u8; 32],
+    pub chain_length:                        u64,
+    pub original_commitment_sequence_number: u64,
+}

apps/fortuna/src/config/run.rs

@@ -1,6 +1,7 @@
 use {
     crate::config::{
         ConfigOptions,
+        ProviderConfigOptions,
         RandomnessOptions,
     },
     anyhow::Result,
@@ -18,6 +19,9 @@ pub struct RunOptions {
     #[command(flatten)]
     pub config: ConfigOptions,
 
+    #[command(flatten)]
+    pub provider_config: ProviderConfigOptions,
+
     #[command(flatten)]
     pub randomness: RandomnessOptions,
 
@@ -32,16 +36,20 @@ pub struct RunOptions {
     #[arg(env = "FORTUNA_PROVIDER")]
     pub provider: Address,
 
-    /// Path to a file containing a 20-byte (40 char) hex encoded Ethereum private key.
+    /// If provided, the keeper will run alongside the Fortuna API service.
+    /// It should be a path to a file containing a 20-byte (40 char) hex encoded Ethereum private key.
     /// This key is required to submit transactions for entropy callback requests.
     /// This key should not be a registered provider.
     #[arg(long = "keeper-private-key")]
     #[arg(env = "KEEPER_PRIVATE_KEY")]
-    pub keeper_private_key_file: String,
+    pub keeper_private_key_file: Option<String>,
 }
 
 impl RunOptions {
-    pub fn load_private_key(&self) -> Result<String> {
-        return Ok((fs::read_to_string(&self.keeper_private_key_file))?);
+    pub fn load_keeper_private_key(&self) -> Result<Option<String>> {
+        if let Some(ref keeper_private_key_file) = self.keeper_private_key_file {
+            return Ok(Some(fs::read_to_string(keeper_private_key_file)?));
+        }
+        return Ok(None);
     }
 }

apps/fortuna/src/keeper.rs

@@ -0,0 +1,487 @@
+use {
+    crate::{
+        api::{
+            self,
+            BlockchainState,
+        },
+        chain::{
+            ethereum::SignablePythContract,
+            reader::{
+                BlockNumber,
+                RequestedWithCallbackEvent,
+            },
+        },
+        config::EthereumConfig,
+    },
+    anyhow::{
+        anyhow,
+        Result,
+    },
+    ethers::{
+        contract::ContractError,
+        providers::{
+            Middleware,
+            Provider,
+            Ws,
+        },
+        types::U256,
+    },
+    futures::StreamExt,
+    std::sync::Arc,
+    tokio::{
+        spawn,
+        sync::mpsc,
+        time::{
+            self,
+            Duration,
+        },
+    },
+    tracing::{
+        self,
+        Instrument,
+    },
+};
+
+#[derive(Debug)]
+pub struct BlockRange {
+    pub from: BlockNumber,
+    pub to:   BlockNumber,
+}
+
+/// How much to wait before retrying in case of an RPC error
+const RETRY_INTERVAL: Duration = Duration::from_secs(5);
+/// How many blocks to look back for events that might be missed when starting the keeper
+const BACKLOG_RANGE: u64 = 1000;
+/// How many blocks to fetch events for in a single rpc call
+const BLOCK_BATCH_SIZE: u64 = 100;
+/// How much to wait before polling the next latest block
+const POLL_INTERVAL: Duration = Duration::from_secs(5);
+
+
+/// Get the latest safe block number for the chain. Retry internally if there is an error.
+async fn get_latest_safe_block(chain_state: &BlockchainState) -> BlockNumber {
+    loop {
+        match chain_state
+            .contract
+            .get_block_number(chain_state.confirmed_block_status)
+            .await
+        {
+            Ok(latest_confirmed_block) => {
+                tracing::info!(
+                    "Fetched latest safe block {}",
+                    latest_confirmed_block - chain_state.reveal_delay_blocks
+                );
+                return latest_confirmed_block - chain_state.reveal_delay_blocks;
+            }
+            Err(e) => {
+                tracing::error!("Error while getting block number. error: {:?}", e);
+                time::sleep(RETRY_INTERVAL).await;
+            }
+        }
+    }
+}
+
+/// Run threads to handle events for the last `BACKLOG_RANGE` blocks, watch for new blocks and
+/// handle any events for the new blocks.
+#[tracing::instrument(name="keeper", skip_all, fields(chain_id=chain_state.id))]
+pub async fn run_keeper_threads(
+    private_key: String,
+    chain_eth_config: EthereumConfig,
+    chain_state: BlockchainState,
+) {
+    tracing::info!("starting keeper");
+    let latest_safe_block = get_latest_safe_block(&chain_state).in_current_span().await;
+    tracing::info!("latest safe block: {}", &latest_safe_block);
+
+    let contract = Arc::new(
+        SignablePythContract::from_config(&chain_eth_config, &private_key)
+            .await
+            .expect("Chain config should be valid"),
+    );
+
+    // Spawn a thread to handle the events from last BACKLOG_RANGE blocks.
+    spawn(
+        process_backlog(
+            BlockRange {
+                from: latest_safe_block.saturating_sub(BACKLOG_RANGE),
+                to:   latest_safe_block,
+            },
+            contract.clone(),
+            chain_eth_config.gas_limit,
+            chain_state.clone(),
+        )
+        .in_current_span(),
+    );
+
+    let (tx, rx) = mpsc::channel::<BlockRange>(1000);
+    // Spawn a thread to watch for new blocks and send the range of blocks for which events has not been handled to the `tx` channel.
+    spawn(
+        watch_blocks_wrapper(
+            chain_state.clone(),
+            latest_safe_block,
+            tx,
+            chain_eth_config.geth_rpc_wss.clone(),
+        )
+        .in_current_span(),
+    );
+    // Spawn a thread that listens for block ranges on the `rx` channel and processes the events for those blocks.
+    spawn(
+        process_new_blocks(
+            chain_state.clone(),
+            rx,
+            Arc::clone(&contract),
+            chain_eth_config.gas_limit,
+        )
+        .in_current_span(),
+    );
+}
+
+
+/// Process an event for a chain. It estimates the gas for the reveal with callback and
+/// submits the transaction if the gas estimate is below the gas limit.
+/// It will return an Error if the gas estimation failed with a provider error or if the
+/// reveal with callback failed with a provider error.
+pub async fn process_event(
+    event: RequestedWithCallbackEvent,
+    chain_config: &BlockchainState,
+    contract: &Arc<SignablePythContract>,
+    gas_limit: U256,
+) -> Result<()> {
+    if chain_config.provider_address != event.provider_address {
+        return Ok(());
+    }
+    let provider_revelation = match chain_config.state.reveal(event.sequence_number) {
+        Ok(result) => result,
+        Err(e) => {
+            tracing::error!(
+                sequence_number = &event.sequence_number,
+                "Error while revealing with error: {:?}",
+                e
+            );
+            return Ok(());
+        }
+    };
+
+    let gas_estimate_res = chain_config
+        .contract
+        .estimate_reveal_with_callback_gas(
+            event.provider_address,
+            event.sequence_number,
+            event.user_random_number,
+            provider_revelation,
+        )
+        .in_current_span()
+        .await;
+
+    match gas_estimate_res {
+        Ok(gas_estimate_option) => match gas_estimate_option {
+            Some(gas_estimate) => {
+                // Pad the gas estimate by 33%
+                let (gas_estimate, _) = gas_estimate
+                    .saturating_mul(U256::from(4))
+                    .div_mod(U256::from(3));
+
+                if gas_estimate > gas_limit {
+                    tracing::error!(
+                        sequence_number = &event.sequence_number,
+                        "Gas estimate for reveal with callback is higher than the gas limit"
+                    );
+                    return Ok(());
+                }
+
+                let contract_call = contract
+                    .reveal_with_callback(
+                        event.provider_address,
+                        event.sequence_number,
+                        event.user_random_number,
+                        provider_revelation,
+                    )
+                    .gas(gas_estimate);
+
+                let res = contract_call.send().await;
+
+                let pending_tx = match res {
+                    Ok(pending_tx) => pending_tx,
+                    Err(e) => match e {
+                        // If there is a provider error, we weren't able to send the transaction.
+                        // We will return an error. So, that the caller can decide what to do (retry).
+                        ContractError::ProviderError { e } => return Err(e.into()),
+                        // For all the other errors, it is likely the case we won't be able to reveal for
+                        // ever. We will return an Ok(()) to signal that we have processed this reveal
+                        // and concluded that its Ok to not reveal.
+                        _ => {
+                            tracing::error!(
+                                sequence_number = &event.sequence_number,
+                                "Error while revealing with error: {:?}",
+                                e
+                            );
+                            return Ok(());
+                        }
+                    },
+                };
+
+                match pending_tx.await {
+                    Ok(res) => {
+                        tracing::info!(
+                            sequence_number = &event.sequence_number,
+                            "Revealed with res: {:?}",
+                            res
+                        );
+                        Ok(())
+                    }
+                    Err(e) => {
+                        tracing::error!(
+                            sequence_number = &event.sequence_number,
+                            "Error while revealing with error: {:?}",
+                            e
+                        );
+                        Err(e.into())
+                    }
+                }
+            }
+            None => {
+                tracing::info!(
+                    sequence_number = &event.sequence_number,
+                    "Not processing event"
+                );
+                Ok(())
+            }
+        },
+        Err(e) => {
+            tracing::error!(
+                sequence_number = &event.sequence_number,
+                "Error while simulating reveal with error: {:?}",
+                e
+            );
+            Err(e)
+        }
+    }
+}
+
+
+/// Process a range of blocks in batches. It calls the `process_single_block_batch` method for each batch.
+#[tracing::instrument(skip_all, fields(range_from_block=block_range.from, range_to_block=block_range.to))]
+pub async fn process_block_range(
+    block_range: BlockRange,
+    contract: Arc<SignablePythContract>,
+    gas_limit: U256,
+    chain_state: api::BlockchainState,
+) {
+    let BlockRange {
+        from: first_block,
+        to: last_block,
+    } = block_range;
+    let mut current_block = first_block;
+    while current_block <= last_block {
+        let mut to_block = current_block + BLOCK_BATCH_SIZE;
+        if to_block > last_block {
+            to_block = last_block;
+        }
+
+        process_single_block_batch(
+            BlockRange {
+                from: current_block,
+                to:   to_block,
+            },
+            contract.clone(),
+            gas_limit,
+            chain_state.clone(),
+        )
+        .in_current_span()
+        .await;
+
+        current_block = to_block + 1;
+    }
+}
+
+/// Process a batch of blocks for a chain. It will fetch events for all the blocks in a single call for the provided batch
+/// and then try to process them one by one. If the process fails, it will retry indefinitely.
+#[tracing::instrument(name="batch", skip_all, fields(batch_from_block=block_range.from, batch_to_block=block_range.to))]
+pub async fn process_single_block_batch(
+    block_range: BlockRange,
+    contract: Arc<SignablePythContract>,
+    gas_limit: U256,
+    chain_state: api::BlockchainState,
+) {
+    loop {
+        let events_res = chain_state
+            .contract
+            .get_request_with_callback_events(block_range.from, block_range.to)
+            .await;
+
+        match events_res {
+            Ok(events) => {
+                tracing::info!(num_of_events = &events.len(), "Processing",);
+                for event in &events {
+                    tracing::info!(sequence_number = &event.sequence_number, "Processing event",);
+                    while let Err(e) =
+                        process_event(event.clone(), &chain_state, &contract, gas_limit)
+                            .in_current_span()
+                            .await
+                    {
+                        tracing::error!(
+                            sequence_number = &event.sequence_number,
+                            "Error while processing event. Waiting for {} seconds before retry. error: {:?}",
+                            RETRY_INTERVAL.as_secs(),
+                            e
+                        );
+                        time::sleep(RETRY_INTERVAL).await;
+                    }
+                    tracing::info!(sequence_number = &event.sequence_number, "Processed event",);
+                }
+                tracing::info!(num_of_events = &events.len(), "Processed",);
+                break;
+            }
+            Err(e) => {
+                tracing::error!(
+                    "Error while getting events. Waiting for {} seconds before retry. error: {:?}",
+                    RETRY_INTERVAL.as_secs(),
+                    e
+                );
+                time::sleep(RETRY_INTERVAL).await;
+            }
+        }
+    }
+}
+
+/// Wrapper for the `watch_blocks` method. If there was an error while watching, it will retry after a delay.
+/// It retries indefinitely.
+#[tracing::instrument(name="watch_blocks", skip_all, fields(initial_safe_block=latest_safe_block))]
+pub async fn watch_blocks_wrapper(
+    chain_state: BlockchainState,
+    latest_safe_block: BlockNumber,
+    tx: mpsc::Sender<BlockRange>,
+    geth_rpc_wss: Option<String>,
+) {
+    let mut last_safe_block_processed = latest_safe_block;
+    loop {
+        if let Err(e) = watch_blocks(
+            chain_state.clone(),
+            &mut last_safe_block_processed,
+            tx.clone(),
+            geth_rpc_wss.clone(),
+        )
+        .in_current_span()
+        .await
+        {
+            tracing::error!("watching blocks. error: {:?}", e);
+            time::sleep(RETRY_INTERVAL).await;
+        }
+    }
+}
+
+/// Watch for new blocks and send the range of blocks for which events have not been handled to the `tx` channel.
+/// We are subscribing to new blocks instead of events. If we miss some blocks, it will be fine as we are sending
+/// block ranges to the `tx` channel. If we have subscribed to events, we could have missed those and won't even
+/// know about it.
+pub async fn watch_blocks(
+    chain_state: BlockchainState,
+    last_safe_block_processed: &mut BlockNumber,
+    tx: mpsc::Sender<BlockRange>,
+    geth_rpc_wss: Option<String>,
+) -> Result<()> {
+    tracing::info!("Watching blocks to handle new events");
+
+    let provider_option = match geth_rpc_wss {
+        Some(wss) => Some(match Provider::<Ws>::connect(wss.clone()).await {
+            Ok(provider) => provider,
+            Err(e) => {
+                tracing::error!("Error while connecting to wss: {}. error: {:?}", wss, e);
+                return Err(e.into());
+            }
+        }),
+        None => {
+            tracing::info!("No wss provided");
+            None
+        }
+    };
+
+    let mut stream_option = match provider_option {
+        Some(ref provider) => Some(match provider.subscribe_blocks().await {
+            Ok(client) => client,
+            Err(e) => {
+                tracing::error!("Error while subscribing to blocks. error {:?}", e);
+                return Err(e.into());
+            }
+        }),
+        None => None,
+    };
+
+    loop {
+        match stream_option {
+            Some(ref mut stream) => {
+                if let None = stream.next().await {
+                    tracing::error!("Error blocks subscription stream ended");
+                    return Err(anyhow!("Error blocks subscription stream ended"));
+                }
+            }
+            None => {
+                time::sleep(POLL_INTERVAL).await;
+            }
+        }
+
+        let latest_safe_block = get_latest_safe_block(&chain_state).in_current_span().await;
+        if latest_safe_block > *last_safe_block_processed {
+            match tx
+                .send(BlockRange {
+                    from: *last_safe_block_processed + 1,
+                    to:   latest_safe_block,
+                })
+                .await
+            {
+                Ok(_) => {
+                    tracing::info!(
+                        from_block = *last_safe_block_processed + 1,
+                        to_block = &latest_safe_block,
+                        "Block range sent to handle events",
+                    );
+                    *last_safe_block_processed = latest_safe_block;
+                }
+                Err(e) => {
+                    tracing::error!(
+                            "Error while sending block range to handle events. These will be handled in next call. error: {:?}",
+                            e
+                        );
+                }
+            };
+        }
+    }
+}
+
+/// It waits on rx channel to receive block ranges and then calls process_block_range to process them.
+#[tracing::instrument(skip_all)]
+pub async fn process_new_blocks(
+    chain_state: BlockchainState,
+    mut rx: mpsc::Receiver<BlockRange>,
+    contract: Arc<SignablePythContract>,
+    gas_limit: U256,
+) {
+    tracing::info!("Waiting for new block ranges to process");
+    loop {
+        if let Some(block_range) = rx.recv().await {
+            process_block_range(
+                block_range,
+                Arc::clone(&contract),
+                gas_limit,
+                chain_state.clone(),
+            )
+            .in_current_span()
+            .await;
+        }
+    }
+}
+
+/// Processes the backlog_range for a chain.
+#[tracing::instrument(skip_all)]
+pub async fn process_backlog(
+    backlog_range: BlockRange,
+    contract: Arc<SignablePythContract>,
+    gas_limit: U256,
+    chain_state: BlockchainState,
+) {
+    tracing::info!("Processing backlog");
+    process_block_range(backlog_range, contract, gas_limit, chain_state)
+        .in_current_span()
+        .await;
+    tracing::info!("Backlog processed");
+}

apps/hermes/Cargo.lock

@@ -1796,7 +1796,7 @@ checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
 
 [[package]]
 name = "hermes"
-version = "0.5.4"
+version = "0.5.9"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -3138,7 +3138,7 @@ dependencies = [
 
 [[package]]
 name = "pythnet-sdk"
-version = "2.0.0"
+version = "2.1.0"
 dependencies = [
  "bincode",
  "borsh 0.10.3",

apps/hermes/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name        = "hermes"
-version     = "0.5.4"
+version     = "0.5.9"
 description = "Hermes is an agent that provides Verified Prices from the Pythnet Pyth Oracle."
 edition     = "2021"
 
@@ -31,7 +31,7 @@ prometheus-client  = { version = "0.21.2" }
 prost              = { version = "0.12.1" }
 pyth-sdk           = { version = "0.8.0" }
 pyth-sdk-solana    = { version = "0.9.0" }
-pythnet-sdk        = { path = "../pythnet/pythnet_sdk/", version = "2.0.0", features = ["strum"] }
+pythnet-sdk        = { path = "../../pythnet/pythnet_sdk/", version = "2.0.0", features = ["strum"] }
 rand               = { version = "0.8.5" }
 reqwest            = { version = "0.11.14", features = ["blocking", "json"] }
 secp256k1          = { version = "0.27.0", features = ["rand", "recovery", "serde"] }

apps/hermes/Dockerfile

@@ -12,15 +12,15 @@ RUN rustup default nightly-2024-03-26
 
 # Build
 WORKDIR /src
-COPY hermes hermes
+COPY apps/hermes apps/hermes
 COPY pythnet/pythnet_sdk pythnet/pythnet_sdk
 
 
-WORKDIR /src/hermes
+WORKDIR /src/apps/hermes
 
 RUN --mount=type=cache,target=/root/.cargo/registry cargo build --release
 
 FROM rust:1.77.0
 
 # Copy artifacts from other images
-COPY --from=build /src/hermes/target/release/hermes /usr/local/bin/
+COPY --from=build /src/apps/hermes/target/release/hermes /usr/local/bin/

apps/hermes/README.md

@@ -35,14 +35,14 @@ To set up and run a Hermes node, follow the steps below:
    ```
 4. **Build the project**: Navigate to the project directory and run the following command to build the project:
    ```bash
-   cd hermes
+   cd apps/hermes
    cargo build --release
    ```
    This will create a binary in the target/release directory.
 5. **Run the node**: To run Hermes for Pythnet, use the following command:
 
    ```bash
-   ./target/release/hermes run \
+   cargo run --release -- run \
      --pythnet-http-addr https://pythnet-rpc/ \
      --pythnet-ws-addr wss://pythnet-rpc/ \
      --wormhole-spy-rpc-addr https://wormhole-spy-rpc/

apps/hermes/src/api.rs

@@ -1,6 +1,5 @@
 use {
     crate::{
-        aggregate::AggregationEvent,
         config::RunOptions,
         state::State,
     },
@@ -14,7 +13,6 @@ use {
     ipnet::IpNet,
     serde_qs::axum::QsQueryConfig,
     std::sync::Arc,
-    tokio::sync::broadcast::Sender,
     tower_http::cors::CorsLayer,
     utoipa::OpenApi,
     utoipa_swagger_ui::SwaggerUi,
@@ -23,15 +21,13 @@ use {
 mod doc_examples;
 mod metrics_middleware;
 mod rest;
-mod sse;
 pub mod types;
 mod ws;
 
 pub struct ApiState<S = State> {
-    pub state:     Arc<S>,
-    pub ws:        Arc<ws::WsState>,
-    pub metrics:   Arc<metrics_middleware::Metrics>,
-    pub update_tx: Sender<AggregationEvent>,
+    pub state:   Arc<S>,
+    pub ws:      Arc<ws::WsState>,
+    pub metrics: Arc<metrics_middleware::Metrics>,
 }
 
 /// Manually implement `Clone` as the derive macro will try and slap `Clone` on
@@ -39,10 +35,9 @@ pub struct ApiState<S = State> {
 impl<S> Clone for ApiState<S> {
     fn clone(&self) -> Self {
         Self {
-            state:     self.state.clone(),
-            ws:        self.ws.clone(),
-            metrics:   self.metrics.clone(),
-            update_tx: self.update_tx.clone(),
+            state:   self.state.clone(),
+            ws:      self.ws.clone(),
+            metrics: self.metrics.clone(),
         }
     }
 }
@@ -52,7 +47,6 @@ impl ApiState<State> {
         state: Arc<State>,
         ws_whitelist: Vec<IpNet>,
         requester_ip_header_name: String,
-        update_tx: Sender<AggregationEvent>,
     ) -> Self {
         Self {
             metrics: Arc::new(metrics_middleware::Metrics::new(state.clone())),
@@ -62,24 +56,18 @@ impl ApiState<State> {
                 state.clone(),
             )),
             state,
-            update_tx,
         }
     }
 }
 
-#[tracing::instrument(skip(opts, state, update_tx))]
-pub async fn spawn(
-    opts: RunOptions,
-    state: Arc<State>,
-    update_tx: Sender<AggregationEvent>,
-) -> Result<()> {
+#[tracing::instrument(skip(opts, state))]
+pub async fn spawn(opts: RunOptions, state: Arc<State>) -> Result<()> {
     let state = {
         let opts = opts.clone();
         ApiState::new(
             state,
             opts.rpc.ws_whitelist,
             opts.rpc.requester_ip_header_name,
-            update_tx,
         )
     };
 
@@ -106,6 +94,7 @@ pub async fn run(opts: RunOptions, state: ApiState) -> Result<()> {
             rest::latest_price_updates,
             rest::timestamp_price_updates,
             rest::price_feeds_metadata,
+            rest::price_stream_sse_handler,
         ),
         components(
             schemas(
@@ -135,6 +124,7 @@ pub async fn run(opts: RunOptions, state: ApiState) -> Result<()> {
     // Initialize Axum Router. Note the type here is a `Router<State>` due to the use of the
     // `with_state` method which replaces `Body` with `State` in the type signature.
     let app = Router::new();
+    #[allow(deprecated)]
     let app = app
         .merge(SwaggerUi::new("/docs").url("/docs/openapi.json", ApiDoc::openapi()))
         .route("/", get(rest::index))
@@ -146,7 +136,7 @@ pub async fn run(opts: RunOptions, state: ApiState) -> Result<()> {
         .route("/api/price_feed_ids", get(rest::price_feed_ids))
         .route(
             "/v2/updates/price/stream",
-            get(sse::price_stream_sse_handler),
+            get(rest::price_stream_sse_handler),
         )
         .route("/v2/updates/price/latest", get(rest::latest_price_updates))
         .route(

apps/hermes/src/api/doc_examples.rs

@@ -1,4 +1,4 @@
-use crate::aggregate::UnixTimestamp;
+use crate::state::aggregate::UnixTimestamp;
 
 // Example values for the utoipa API docs.
 // Note that each of these expressions is only evaluated once when the documentation is created,

apps/hermes/src/api/rest.rs

@@ -1,5 +1,6 @@
 use {
     super::ApiState,
+    crate::state::aggregate::Aggregates,
     axum::{
         http::StatusCode,
         response::{
@@ -35,6 +36,7 @@ pub use {
     v2::{
         latest_price_updates::*,
         price_feeds_metadata::*,
+        sse::*,
         timestamp_price_updates::*,
     },
 };
@@ -92,11 +94,15 @@ impl IntoResponse for RestError {
 }
 
 /// Verify that the price ids exist in the aggregate state.
-pub async fn verify_price_ids_exist(
-    state: &ApiState,
+pub async fn verify_price_ids_exist<S>(
+    state: &ApiState<S>,
     price_ids: &[PriceIdentifier],
-) -> Result<(), RestError> {
-    let all_ids = crate::aggregate::get_price_feed_ids(&*state.state).await;
+) -> Result<(), RestError>
+where
+    S: Aggregates,
+{
+    let state = &*state.state;
+    let all_ids = Aggregates::get_price_feed_ids(state).await;
     let missing_ids = price_ids
         .iter()
         .filter(|id| !all_ids.contains(id))

apps/hermes/src/api/rest/get_price_feed.rs

@@ -1,10 +1,6 @@
 use {
     super::verify_price_ids_exist,
     crate::{
-        aggregate::{
-            RequestTime,
-            UnixTimestamp,
-        },
         api::{
             doc_examples,
             rest::RestError,
@@ -12,6 +8,12 @@ use {
                 PriceIdInput,
                 RpcPriceFeed,
             },
+            ApiState,
+        },
+        state::aggregate::{
+            Aggregates,
+            RequestTime,
+            UnixTimestamp,
         },
     },
     anyhow::Result,
@@ -47,6 +49,8 @@ pub struct GetPriceFeedQueryParams {
     binary: bool,
 }
 
+/// **Deprecated: use /v2/updates/price/{publish_time} instead**
+///
 /// Get a price update for a price feed with a specific timestamp
 ///
 /// Given a price feed id and timestamp, retrieve the Pyth price update closest to that timestamp.
@@ -60,16 +64,20 @@ pub struct GetPriceFeedQueryParams {
         GetPriceFeedQueryParams
     )
 )]
-pub async fn get_price_feed(
-    State(state): State<crate::api::ApiState>,
+#[deprecated]
+pub async fn get_price_feed<S>(
+    State(state): State<ApiState<S>>,
     QsQuery(params): QsQuery<GetPriceFeedQueryParams>,
-) -> Result<Json<RpcPriceFeed>, RestError> {
+) -> Result<Json<RpcPriceFeed>, RestError>
+where
+    S: Aggregates,
+{
     let price_id: PriceIdentifier = params.id.into();
-
     verify_price_ids_exist(&state, &[price_id]).await?;
 
-    let price_feeds_with_update_data = crate::aggregate::get_price_feeds_with_update_data(
-        &*state.state,
+    let state = &*state.state;
+    let price_feeds_with_update_data = Aggregates::get_price_feeds_with_update_data(
+        state,
         &[price_id],
         RequestTime::FirstAfter(params.publish_time),
     )

apps/hermes/src/api/rest/get_vaa.rs

@@ -1,15 +1,16 @@
 use {
     super::verify_price_ids_exist,
     crate::{
-        aggregate::{
-            get_price_feeds_with_update_data,
-            RequestTime,
-            UnixTimestamp,
-        },
         api::{
             doc_examples,
             rest::RestError,
             types::PriceIdInput,
+            ApiState,
+        },
+        state::aggregate::{
+            Aggregates,
+            RequestTime,
+            UnixTimestamp,
         },
     },
     anyhow::Result,
@@ -54,6 +55,8 @@ pub struct GetVaaResponse {
     publish_time: UnixTimestamp,
 }
 
+/// **Deprecated: use /v2/updates/price/{publish_time} instead**
+///
 /// Get a VAA for a price feed with a specific timestamp
 ///
 /// Given a price feed id and timestamp, retrieve the Pyth price update closest to that timestamp.
@@ -68,16 +71,20 @@ pub struct GetVaaResponse {
         GetVaaQueryParams
     )
 )]
-pub async fn get_vaa(
-    State(state): State<crate::api::ApiState>,
+#[deprecated]
+pub async fn get_vaa<S>(
+    State(state): State<ApiState<S>>,
     QsQuery(params): QsQuery<GetVaaQueryParams>,
-) -> Result<Json<GetVaaResponse>, RestError> {
+) -> Result<Json<GetVaaResponse>, RestError>
+where
+    S: Aggregates,
+{
     let price_id: PriceIdentifier = params.id.into();
-
     verify_price_ids_exist(&state, &[price_id]).await?;
 
-    let price_feeds_with_update_data = get_price_feeds_with_update_data(
-        &*state.state,
+    let state = &*state.state;
+    let price_feeds_with_update_data = Aggregates::get_price_feeds_with_update_data(
+        state,
         &[price_id],
         RequestTime::FirstAfter(params.publish_time),
     )

apps/hermes/src/api/rest/get_vaa_ccip.rs

@@ -1,11 +1,15 @@
 use {
     super::verify_price_ids_exist,
     crate::{
-        aggregate::{
+        api::{
+            rest::RestError,
+            ApiState,
+        },
+        state::aggregate::{
+            Aggregates,
             RequestTime,
             UnixTimestamp,
         },
-        api::rest::RestError,
     },
     anyhow::Result,
     axum::{
@@ -42,6 +46,8 @@ pub struct GetVaaCcipResponse {
     data: String, // TODO: Use a typed wrapper for the hex output with leading 0x.
 }
 
+/// **Deprecated: use /v2/updates/price/{publish_time} instead**
+///
 /// Get a VAA for a price feed using CCIP
 ///
 /// This endpoint accepts a single argument which is a hex-encoded byte string of the following form:
@@ -56,25 +62,30 @@ pub struct GetVaaCcipResponse {
         GetVaaCcipQueryParams
     )
 )]
-pub async fn get_vaa_ccip(
-    State(state): State<crate::api::ApiState>,
+#[deprecated]
+pub async fn get_vaa_ccip<S>(
+    State(state): State<ApiState<S>>,
     QsQuery(params): QsQuery<GetVaaCcipQueryParams>,
-) -> Result<Json<GetVaaCcipResponse>, RestError> {
+) -> Result<Json<GetVaaCcipResponse>, RestError>
+where
+    S: Aggregates,
+{
     let price_id: PriceIdentifier = PriceIdentifier::new(
         params.data[0..32]
             .try_into()
             .map_err(|_| RestError::InvalidCCIPInput)?,
     );
+    verify_price_ids_exist(&state, &[price_id]).await?;
+
     let publish_time = UnixTimestamp::from_be_bytes(
         params.data[32..40]
             .try_into()
             .map_err(|_| RestError::InvalidCCIPInput)?,
     );
 
-    verify_price_ids_exist(&state, &[price_id]).await?;
-
-    let price_feeds_with_update_data = crate::aggregate::get_price_feeds_with_update_data(
-        &*state.state,
+    let state = &*state.state;
+    let price_feeds_with_update_data = Aggregates::get_price_feeds_with_update_data(
+        state,
         &[price_id],
         RequestTime::FirstAfter(publish_time),
     )

apps/hermes/src/api/rest/index.rs

@@ -17,6 +17,7 @@ pub async fn index() -> impl IntoResponse {
         "/api/get_vaa?id=<price_feed_id>&publish_time=<publish_time_in_unix_timestamp>",
         "/api/get_vaa_ccip?data=<0x<price_feed_id_32_bytes>+<publish_time_unix_timestamp_be_8_bytes>>",
         "/v2/updates/price/latest?ids[]=<price_feed_id>&ids[]=<price_feed_id_2>&..(&encoding=hex|base64)(&parsed=false)",
+        "/v2/updates/price/stream?ids[]=<price_feed_id>&ids[]=<price_feed_id_2>&..(&encoding=hex|base64)(&parsed=false)(&allow_unordered=false)(&benchmarks_only=false)",
         "/v2/updates/price/<timestamp>?ids[]=<price_feed_id>&ids[]=<price_feed_id_2>&..(&encoding=hex|base64)(&parsed=false)",
         "/v2/price_feeds?(query=btc)(&asset_type=crypto|equity|fx|metal|rates)",
     ])

apps/hermes/src/api/rest/latest_price_feeds.rs

@@ -1,13 +1,17 @@
 use {
     super::verify_price_ids_exist,
     crate::{
-        aggregate::RequestTime,
         api::{
             rest::RestError,
             types::{
                 PriceIdInput,
                 RpcPriceFeed,
             },
+            ApiState,
+        },
+        state::aggregate::{
+            Aggregates,
+            RequestTime,
         },
     },
     anyhow::Result,
@@ -46,6 +50,8 @@ pub struct LatestPriceFeedsQueryParams {
     binary: bool,
 }
 
+/// **Deprecated: use /v2/updates/price/latest instead**
+///
 /// Get the latest price updates by price feed id.
 ///
 /// Given a collection of price feed ids, retrieve the latest Pyth price for each price feed.
@@ -59,28 +65,29 @@ pub struct LatestPriceFeedsQueryParams {
         LatestPriceFeedsQueryParams
     )
 )]
-pub async fn latest_price_feeds(
-    State(state): State<crate::api::ApiState>,
+#[deprecated]
+pub async fn latest_price_feeds<S>(
+    State(state): State<ApiState<S>>,
     QsQuery(params): QsQuery<LatestPriceFeedsQueryParams>,
-) -> Result<Json<Vec<RpcPriceFeed>>, RestError> {
+) -> Result<Json<Vec<RpcPriceFeed>>, RestError>
+where
+    S: Aggregates,
+{
     let price_ids: Vec<PriceIdentifier> = params.ids.into_iter().map(|id| id.into()).collect();
-
     verify_price_ids_exist(&state, &price_ids).await?;
 
-    let price_feeds_with_update_data = crate::aggregate::get_price_feeds_with_update_data(
-        &*state.state,
-        &price_ids,
-        RequestTime::Latest,
-    )
-    .await
-    .map_err(|e| {
-        tracing::warn!(
-            "Error getting price feeds {:?} with update data: {:?}",
-            price_ids,
-            e
-        );
-        RestError::UpdateDataNotFound
-    })?;
+    let state = &*state.state;
+    let price_feeds_with_update_data =
+        Aggregates::get_price_feeds_with_update_data(state, &price_ids, RequestTime::Latest)
+            .await
+            .map_err(|e| {
+                tracing::warn!(
+                    "Error getting price feeds {:?} with update data: {:?}",
+                    price_ids,
+                    e
+                );
+                RestError::UpdateDataNotFound
+            })?;
 
     Ok(Json(
         price_feeds_with_update_data

apps/hermes/src/api/rest/latest_vaas.rs

@@ -1,11 +1,15 @@
 use {
     super::verify_price_ids_exist,
     crate::{
-        aggregate::RequestTime,
         api::{
             doc_examples,
             rest::RestError,
             types::PriceIdInput,
+            ApiState,
+        },
+        state::aggregate::{
+            Aggregates,
+            RequestTime,
         },
     },
     anyhow::Result,
@@ -39,6 +43,8 @@ pub struct LatestVaasQueryParams {
 }
 
 
+/// **Deprecated: use /v2/updates/price/latest instead**
+///
 /// Get VAAs for a set of price feed ids.
 ///
 /// Given a collection of price feed ids, retrieve the latest VAA for each. The returned VAA(s) can
@@ -54,28 +60,29 @@ pub struct LatestVaasQueryParams {
         (status = 200, description = "VAAs retrieved successfully", body = Vec<String>, example=json!([doc_examples::vaa_example()]))
     ),
 )]
-pub async fn latest_vaas(
-    State(state): State<crate::api::ApiState>,
+#[deprecated]
+pub async fn latest_vaas<S>(
+    State(state): State<ApiState<S>>,
     QsQuery(params): QsQuery<LatestVaasQueryParams>,
-) -> Result<Json<Vec<String>>, RestError> {
+) -> Result<Json<Vec<String>>, RestError>
+where
+    S: Aggregates,
+{
     let price_ids: Vec<PriceIdentifier> = params.ids.into_iter().map(|id| id.into()).collect();
-
     verify_price_ids_exist(&state, &price_ids).await?;
 
-    let price_feeds_with_update_data = crate::aggregate::get_price_feeds_with_update_data(
-        &*state.state,
-        &price_ids,
-        RequestTime::Latest,
-    )
-    .await
-    .map_err(|e| {
-        tracing::warn!(
-            "Error getting price feeds {:?} with update data: {:?}",
-            price_ids,
-            e
-        );
-        RestError::UpdateDataNotFound
-    })?;
+    let state = &*state.state;
+    let price_feeds_with_update_data =
+        Aggregates::get_price_feeds_with_update_data(state, &price_ids, RequestTime::Latest)
+            .await
+            .map_err(|e| {
+                tracing::warn!(
+                    "Error getting price feeds {:?} with update data: {:?}",
+                    price_ids,
+                    e
+                );
+                RestError::UpdateDataNotFound
+            })?;
 
     Ok(Json(
         price_feeds_with_update_data

apps/hermes/src/api/rest/price_feed_ids.rs

@@ -1,7 +1,11 @@
 use {
-    crate::api::{
-        rest::RestError,
-        types::RpcPriceIdentifier,
+    crate::{
+        api::{
+            rest::RestError,
+            types::RpcPriceIdentifier,
+            ApiState,
+        },
+        state::aggregate::Aggregates,
     },
     anyhow::Result,
     axum::{
@@ -10,6 +14,8 @@ use {
     },
 };
 
+/// **Deprecated: use /v2/price_feeds instead**
+///
 /// Get the set of price feed IDs.
 ///
 /// This endpoint fetches all of the price feed IDs for which price updates can be retrieved.
@@ -21,10 +27,15 @@ use {
         (status = 200, description = "Price feed ids retrieved successfully", body = Vec<RpcPriceIdentifier>)
     ),
 )]
-pub async fn price_feed_ids(
-    State(state): State<crate::api::ApiState>,
-) -> Result<Json<Vec<RpcPriceIdentifier>>, RestError> {
-    let price_feed_ids = crate::aggregate::get_price_feed_ids(&*state.state)
+#[deprecated]
+pub async fn price_feed_ids<S>(
+    State(state): State<ApiState<S>>,
+) -> Result<Json<Vec<RpcPriceIdentifier>>, RestError>
+where
+    S: Aggregates,
+{
+    let state = &*state.state;
+    let price_feed_ids = Aggregates::get_price_feed_ids(state)
         .await
         .into_iter()
         .map(RpcPriceIdentifier::from)

apps/hermes/src/api/rest/ready.rs

@@ -0,0 +1,25 @@
+use {
+    crate::{
+        api::ApiState,
+        state::aggregate::Aggregates,
+    },
+    axum::{
+        extract::State,
+        http::StatusCode,
+        response::{
+            IntoResponse,
+            Response,
+        },
+    },
+};
+
+pub async fn ready<S>(State(state): State<ApiState<S>>) -> Response
+where
+    S: Aggregates,
+{
+    let state = &*state.state;
+    match Aggregates::is_ready(state).await {
+        true => (StatusCode::OK, "OK").into_response(),
+        false => (StatusCode::SERVICE_UNAVAILABLE, "Service Unavailable").into_response(),
+    }
+}

apps/hermes/src/api/rest/v2/latest_price_updates.rs

@@ -1,6 +1,5 @@
 use {
     crate::{
-        aggregate::RequestTime,
         api::{
             rest::{
                 verify_price_ids_exist,
@@ -13,6 +12,11 @@ use {
                 PriceIdInput,
                 PriceUpdate,
             },
+            ApiState,
+        },
+        state::aggregate::{
+            Aggregates,
+            RequestTime,
         },
     },
     anyhow::Result,
@@ -46,11 +50,11 @@ pub struct LatestPriceUpdatesQueryParams {
     #[param(example = "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43")]
     ids: Vec<PriceIdInput>,
 
-    /// If true, include the parsed price update in the `parsed` field of each returned feed.
+    /// If true, include the parsed price update in the `parsed` field of each returned feed. Default is `hex`.
     #[serde(default)]
     encoding: EncodingType,
 
-    /// If true, include the parsed price update in the `parsed` field of each returned feed.
+    /// If true, include the parsed price update in the `parsed` field of each returned feed. Default is `true`.
     #[serde(default = "default_true")]
     parsed: bool,
 }
@@ -73,28 +77,28 @@ fn default_true() -> bool {
         LatestPriceUpdatesQueryParams
     )
 )]
-pub async fn latest_price_updates(
-    State(state): State<crate::api::ApiState>,
+pub async fn latest_price_updates<S>(
+    State(state): State<ApiState<S>>,
     QsQuery(params): QsQuery<LatestPriceUpdatesQueryParams>,
-) -> Result<Json<PriceUpdate>, RestError> {
+) -> Result<Json<PriceUpdate>, RestError>
+where
+    S: Aggregates,
+{
     let price_ids: Vec<PriceIdentifier> = params.ids.into_iter().map(|id| id.into()).collect();
-
     verify_price_ids_exist(&state, &price_ids).await?;
 
-    let price_feeds_with_update_data = crate::aggregate::get_price_feeds_with_update_data(
-        &*state.state,
-        &price_ids,
-        RequestTime::Latest,
-    )
-    .await
-    .map_err(|e| {
-        tracing::warn!(
-            "Error getting price feeds {:?} with update data: {:?}",
-            price_ids,
-            e
-        );
-        RestError::UpdateDataNotFound
-    })?;
+    let state = &*state.state;
+    let price_feeds_with_update_data =
+        Aggregates::get_price_feeds_with_update_data(state, &price_ids, RequestTime::Latest)
+            .await
+            .map_err(|e| {
+                tracing::warn!(
+                    "Error getting price feeds {:?} with update data: {:?}",
+                    price_ids,
+                    e
+                );
+                RestError::UpdateDataNotFound
+            })?;
 
     let price_update_data = price_feeds_with_update_data.update_data;
     let encoded_data: Vec<String> = price_update_data

apps/hermes/src/api/rest/v2/mod.rs

@@ -1,3 +1,4 @@
 pub mod latest_price_updates;
 pub mod price_feeds_metadata;
+pub mod sse;
 pub mod timestamp_price_updates;

apps/hermes/src/api/rest/v2/sse.rs

@@ -1,9 +1,5 @@
 use {
     crate::{
-        aggregate::{
-            AggregationEvent,
-            RequestTime,
-        },
         api::{
             rest::{
                 verify_price_ids_exist,
@@ -15,9 +11,15 @@ use {
                 ParsedPriceUpdate,
                 PriceIdInput,
                 PriceUpdate,
+                RpcPriceIdentifier,
             },
             ApiState,
         },
+        state::aggregate::{
+            Aggregates,
+            AggregationEvent,
+            RequestTime,
+        },
     },
     anyhow::Result,
     axum::{
@@ -56,13 +58,21 @@ pub struct StreamPriceUpdatesQueryParams {
     #[param(example = "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43")]
     ids: Vec<PriceIdInput>,
 
-    /// If true, include the parsed price update in the `parsed` field of each returned feed.
+    /// If true, include the parsed price update in the `parsed` field of each returned feed. Default is `hex`.
     #[serde(default)]
     encoding: EncodingType,
 
-    /// If true, include the parsed price update in the `parsed` field of each returned feed.
+    /// If true, include the parsed price update in the `parsed` field of each returned feed. Default is `true`.
     #[serde(default = "default_true")]
     parsed: bool,
+
+    /// If true, allows unordered price updates to be included in the stream.
+    #[serde(default)]
+    allow_unordered: bool,
+
+    /// If true, only include benchmark prices that are the initial price updates at a given timestamp (i.e., prevPubTime != pubTime).
+    #[serde(default)]
+    benchmarks_only: bool,
 }
 
 fn default_true() -> bool {
@@ -79,16 +89,22 @@ fn default_true() -> bool {
     params(StreamPriceUpdatesQueryParams)
 )]
 /// SSE route handler for streaming price updates.
-pub async fn price_stream_sse_handler(
-    State(state): State<ApiState>,
+pub async fn price_stream_sse_handler<S>(
+    State(state): State<ApiState<S>>,
     QsQuery(params): QsQuery<StreamPriceUpdatesQueryParams>,
-) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, RestError> {
+) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, RestError>
+where
+    S: Aggregates,
+    S: Sync,
+    S: Send,
+    S: 'static,
+{
     let price_ids: Vec<PriceIdentifier> = params.ids.into_iter().map(Into::into).collect();
 
     verify_price_ids_exist(&state, &price_ids).await?;
 
     // Clone the update_tx receiver to listen for new price updates
-    let update_rx: broadcast::Receiver<AggregationEvent> = state.update_tx.subscribe();
+    let update_rx: broadcast::Receiver<AggregationEvent> = Aggregates::subscribe(&*state.state);
 
     // Convert the broadcast receiver into a Stream
     let stream = BroadcastStream::new(update_rx);
@@ -105,10 +121,15 @@ pub async fn price_stream_sse_handler(
                         price_ids_clone,
                         params.encoding,
                         params.parsed,
+                        params.benchmarks_only,
+                        params.allow_unordered,
                     )
                     .await
                     {
-                        Ok(price_update) => Ok(Event::default().json_data(price_update).unwrap()),
+                        Ok(Some(update)) => Ok(Event::default()
+                            .json_data(update)
+                            .unwrap_or_else(|e| error_event(e))),
+                        Ok(None) => Ok(Event::default().comment("No update available")),
                         Err(e) => Ok(error_event(e)),
                     }
                 }
@@ -120,24 +141,73 @@ pub async fn price_stream_sse_handler(
     Ok(Sse::new(sse_stream).keep_alive(KeepAlive::default()))
 }
 
-async fn handle_aggregation_event(
+async fn handle_aggregation_event<S>(
     event: AggregationEvent,
-    state: ApiState,
+    state: ApiState<S>,
     mut price_ids: Vec<PriceIdentifier>,
     encoding: EncodingType,
     parsed: bool,
-) -> Result<PriceUpdate> {
+    benchmarks_only: bool,
+    allow_unordered: bool,
+) -> Result<Option<PriceUpdate>>
+where
+    S: Aggregates,
+{
+    // Handle out-of-order events
+    if let AggregationEvent::OutOfOrder { .. } = event {
+        if !allow_unordered {
+            return Ok(None);
+        }
+    }
+
     // We check for available price feed ids to ensure that the price feed ids provided exists since price feeds can be removed.
-    let available_price_feed_ids = crate::aggregate::get_price_feed_ids(&*state.state).await;
+    let available_price_feed_ids = Aggregates::get_price_feed_ids(&*state.state).await;
 
     price_ids.retain(|price_feed_id| available_price_feed_ids.contains(price_feed_id));
 
-    let price_feeds_with_update_data = crate::aggregate::get_price_feeds_with_update_data(
+    let mut price_feeds_with_update_data = Aggregates::get_price_feeds_with_update_data(
         &*state.state,
         &price_ids,
         RequestTime::AtSlot(event.slot()),
     )
     .await?;
+
+    let mut parsed_price_updates: Vec<ParsedPriceUpdate> = price_feeds_with_update_data
+        .price_feeds
+        .into_iter()
+        .map(|price_feed| price_feed.into())
+        .collect();
+
+
+    if benchmarks_only {
+        // Remove those with metadata.prev_publish_time != price.publish_time from parsed_price_updates
+        parsed_price_updates.retain(|price_feed| {
+            price_feed
+                .metadata
+                .prev_publish_time
+                .map_or(false, |prev_time| {
+                    prev_time != price_feed.price.publish_time
+                })
+        });
+        // Retain price id in price_ids that are in parsed_price_updates
+        price_ids.retain(|price_id| {
+            parsed_price_updates
+                .iter()
+                .any(|price_feed| price_feed.id == RpcPriceIdentifier::from(*price_id))
+        });
+        price_feeds_with_update_data = Aggregates::get_price_feeds_with_update_data(
+            &*state.state,
+            &price_ids,
+            RequestTime::AtSlot(event.slot()),
+        )
+        .await?;
+    }
+
+    // Check if price_ids is empty after filtering and return None if it is
+    if price_ids.is_empty() {
+        return Ok(None);
+    }
+
     let price_update_data = price_feeds_with_update_data.update_data;
     let encoded_data: Vec<String> = price_update_data
         .into_iter()
@@ -147,23 +217,15 @@ async fn handle_aggregation_event(
         encoding,
         data: encoded_data,
     };
-    let parsed_price_updates: Option<Vec<ParsedPriceUpdate>> = if parsed {
-        Some(
-            price_feeds_with_update_data
-                .price_feeds
-                .into_iter()
-                .map(|price_feed| price_feed.into())
-                .collect(),
-        )
-    } else {
-        None
-    };
 
-
-    Ok(PriceUpdate {
+    Ok(Some(PriceUpdate {
         binary: binary_price_update,
-        parsed: parsed_price_updates,
-    })
+        parsed: if parsed {
+            Some(parsed_price_updates)
+        } else {
+            None
+        },
+    }))
 }
 
 fn error_event<E: std::fmt::Debug>(e: E) -> Event {

apps/hermes/src/api/rest/v2/timestamp_price_updates.rs

@@ -1,9 +1,5 @@
 use {
     crate::{
-        aggregate::{
-            RequestTime,
-            UnixTimestamp,
-        },
         api::{
             doc_examples,
             rest::{
@@ -17,6 +13,12 @@ use {
                 PriceIdInput,
                 PriceUpdate,
             },
+            ApiState,
+        },
+        state::aggregate::{
+            Aggregates,
+            RequestTime,
+            UnixTimestamp,
         },
     },
     anyhow::Result,
@@ -58,11 +60,11 @@ pub struct TimestampPriceUpdatesQueryParams {
     #[param(example = "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43")]
     ids: Vec<PriceIdInput>,
 
-    /// If true, include the parsed price update in the `parsed` field of each returned feed.
+    /// If true, include the parsed price update in the `parsed` field of each returned feed. Default is `hex`.
     #[serde(default)]
     encoding: EncodingType,
 
-    /// If true, include the parsed price update in the `parsed` field of each returned feed.
+    /// If true, include the parsed price update in the `parsed` field of each returned feed. Default is `true`.
     #[serde(default = "default_true")]
     parsed: bool,
 }
@@ -87,18 +89,22 @@ fn default_true() -> bool {
         TimestampPriceUpdatesQueryParams
     )
 )]
-pub async fn timestamp_price_updates(
-    State(state): State<crate::api::ApiState>,
+pub async fn timestamp_price_updates<S>(
+    State(state): State<ApiState<S>>,
     Path(path_params): Path<TimestampPriceUpdatesPathParams>,
     QsQuery(query_params): QsQuery<TimestampPriceUpdatesQueryParams>,
-) -> Result<Json<PriceUpdate>, RestError> {
+) -> Result<Json<PriceUpdate>, RestError>
+where
+    S: Aggregates,
+{
     let price_ids: Vec<PriceIdentifier> =
         query_params.ids.into_iter().map(|id| id.into()).collect();
 
     verify_price_ids_exist(&state, &price_ids).await?;
 
-    let price_feeds_with_update_data = crate::aggregate::get_price_feeds_with_update_data(
-        &*state.state,
+    let state = &*state.state;
+    let price_feeds_with_update_data = Aggregates::get_price_feeds_with_update_data(
+        state,
         &price_ids,
         RequestTime::FirstAfter(path_params.publish_time),
     )

apps/hermes/src/api/types.rs

@@ -1,6 +1,6 @@
 use {
     super::doc_examples,
-    crate::aggregate::{
+    crate::state::aggregate::{
         PriceFeedUpdate,
         PriceFeedsWithUpdateData,
         Slot,

apps/hermes/src/api/ws.rs

@@ -1,14 +1,18 @@
 use {
-    super::types::{
-        PriceIdInput,
-        RpcPriceFeed,
+    super::{
+        types::{
+            PriceIdInput,
+            RpcPriceFeed,
+        },
+        ApiState,
     },
-    crate::{
+    crate::state::{
         aggregate::{
+            Aggregates,
             AggregationEvent,
             RequestTime,
         },
-        state::State,
+        State,
     },
     anyhow::{
         anyhow,
@@ -212,11 +216,10 @@ pub async fn ws_route_handler(
 }
 
 #[tracing::instrument(skip(stream, state, subscriber_ip))]
-async fn websocket_handler(
-    stream: WebSocket,
-    state: super::ApiState,
-    subscriber_ip: Option<IpAddr>,
-) {
+async fn websocket_handler<S>(stream: WebSocket, state: ApiState<S>, subscriber_ip: Option<IpAddr>)
+where
+    S: Aggregates,
+{
     let ws_state = state.ws.clone();
 
     // Retain the recent rate limit data for the IP addresses to
@@ -235,7 +238,7 @@ async fn websocket_handler(
         })
         .inc();
 
-    let notify_receiver = state.update_tx.subscribe();
+    let notify_receiver = Aggregates::subscribe(&*state.state);
     let (sender, receiver) = stream.split();
     let mut subscriber = Subscriber::new(
         id,
@@ -254,11 +257,11 @@ pub type SubscriberId = usize;
 
 /// Subscriber is an actor that handles a single websocket connection.
 /// It listens to the store for updates and sends them to the client.
-pub struct Subscriber {
+pub struct Subscriber<S> {
     id:                      SubscriberId,
     ip_addr:                 Option<IpAddr>,
     closed:                  bool,
-    store:                   Arc<State>,
+    state:                   Arc<S>,
     ws_state:                Arc<WsState>,
     notify_receiver:         Receiver<AggregationEvent>,
     receiver:                SplitStream<WebSocket>,
@@ -269,11 +272,14 @@ pub struct Subscriber {
     responded_to_ping:       bool,
 }
 
-impl Subscriber {
+impl<S> Subscriber<S>
+where
+    S: Aggregates,
+{
     pub fn new(
         id: SubscriberId,
         ip_addr: Option<IpAddr>,
-        store: Arc<State>,
+        state: Arc<S>,
         ws_state: Arc<WsState>,
         notify_receiver: Receiver<AggregationEvent>,
         receiver: SplitStream<WebSocket>,
@@ -283,7 +289,7 @@ impl Subscriber {
             id,
             ip_addr,
             closed: false,
-            store,
+            state,
             ws_state,
             notify_receiver,
             receiver,
@@ -350,8 +356,9 @@ impl Subscriber {
             .cloned()
             .collect::<Vec<_>>();
 
-        let updates = match crate::aggregate::get_price_feeds_with_update_data(
-            &*self.store,
+        let state = &*self.state;
+        let updates = match Aggregates::get_price_feeds_with_update_data(
+            state,
             &price_feed_ids,
             RequestTime::AtSlot(event.slot()),
         )
@@ -364,8 +371,7 @@ impl Subscriber {
                 // subscription. In this case we just remove the non-existing
                 // price feed from the list and will keep sending updates for
                 // the rest.
-                let available_price_feed_ids =
-                    crate::aggregate::get_price_feed_ids(&*self.store).await;
+                let available_price_feed_ids = Aggregates::get_price_feed_ids(state).await;
 
                 self.price_feeds_with_config
                     .retain(|price_feed_id, _| available_price_feed_ids.contains(price_feed_id));
@@ -376,8 +382,8 @@ impl Subscriber {
                     .cloned()
                     .collect::<Vec<_>>();
 
-                crate::aggregate::get_price_feeds_with_update_data(
-                    &*self.store,
+                Aggregates::get_price_feeds_with_update_data(
+                    state,
                     &price_feed_ids,
                     RequestTime::AtSlot(event.slot()),
                 )
@@ -545,7 +551,7 @@ impl Subscriber {
                 allow_out_of_order,
             }) => {
                 let price_ids: Vec<PriceIdentifier> = ids.into_iter().map(|id| id.into()).collect();
-                let available_price_ids = crate::aggregate::get_price_feed_ids(&*self.store).await;
+                let available_price_ids = Aggregates::get_price_feed_ids(&*self.state).await;
 
                 let not_found_price_ids: Vec<&PriceIdentifier> = price_ids
                     .iter()

apps/hermes/src/config/pythnet.rs

@@ -19,9 +19,9 @@ pub struct Options {
     #[arg(env = "PYTHNET_HTTP_ADDR")]
     pub http_addr: String,
 
-    /// Pyth mapping account address.
-    #[arg(long = "mapping-address")]
+    /// Pyth mapping account address on Pythnet.
+    #[arg(long = "pythnet-mapping-addr")]
     #[arg(default_value = DEFAULT_PYTHNET_MAPPING_ADDR)]
-    #[arg(env = "MAPPING_ADDRESS")]
+    #[arg(env = "PYTHNET_MAPPING_ADDR")]
     pub mapping_addr: Pubkey,
 }

apps/hermes/src/main.rs

@@ -17,7 +17,6 @@ use {
     },
 };
 
-mod aggregate;
 mod api;
 mod config;
 mod metrics_server;
@@ -54,7 +53,7 @@ async fn init() -> Result<()> {
             let (update_tx, _) = tokio::sync::broadcast::channel(1000);
 
             // Initialize a cache store with a 1000 element circular buffer.
-            let store = State::new(update_tx.clone(), 1000, opts.benchmarks.endpoint.clone());
+            let state = State::new(update_tx.clone(), 1000, opts.benchmarks.endpoint.clone());
 
             // Listen for Ctrl+C so we can set the exit flag and wait for a graceful shutdown.
             spawn(async move {
@@ -67,10 +66,10 @@ async fn init() -> Result<()> {
             // Spawn all worker tasks, and wait for all to complete (which will happen if a shutdown
             // signal has been observed).
             let tasks = join_all(vec![
-                spawn(network::wormhole::spawn(opts.clone(), store.clone())),
-                spawn(network::pythnet::spawn(opts.clone(), store.clone())),
-                spawn(metrics_server::run(opts.clone(), store.clone())),
-                spawn(api::spawn(opts.clone(), store.clone(), update_tx)),
+                spawn(network::wormhole::spawn(opts.clone(), state.clone())),
+                spawn(network::pythnet::spawn(opts.clone(), state.clone())),
+                spawn(metrics_server::run(opts.clone(), state.clone())),
+                spawn(api::spawn(opts.clone(), state.clone())),
             ])
             .await;
 

apps/hermes/src/network/pythnet.rs

@@ -4,10 +4,6 @@
 
 use {
     crate::{
-        aggregate::{
-            AccumulatorMessages,
-            Update,
-        },
         api::types::PriceFeedMetadata,
         config::RunOptions,
         network::wormhole::{
@@ -20,7 +16,14 @@ use {
             PriceFeedMeta,
             DEFAULT_PRICE_FEEDS_CACHE_UPDATE_INTERVAL,
         },
-        state::State,
+        state::{
+            aggregate::{
+                AccumulatorMessages,
+                Aggregates,
+                Update,
+            },
+            State,
+        },
     },
     anyhow::{
         anyhow,
@@ -136,7 +139,7 @@ async fn fetch_bridge_data(
     }
 }
 
-pub async fn run(store: Arc<State>, pythnet_ws_endpoint: String) -> Result<()> {
+pub async fn run(store: Arc<State>, pythnet_ws_endpoint: String) -> Result<!> {
     let client = PubsubClient::new(pythnet_ws_endpoint.as_ref()).await?;
 
     let config = RpcProgramAccountsConfig {
@@ -157,59 +160,54 @@ pub async fn run(store: Arc<State>, pythnet_ws_endpoint: String) -> Result<()> {
         .program_subscribe(&system_program::id(), Some(config))
         .await?;
 
-    loop {
-        match notif.next().await {
-            Some(update) => {
-                let account: Account = match update.value.account.decode() {
-                    Some(account) => account,
-                    None => {
-                        tracing::error!(?update, "Failed to decode account from update.");
-                        continue;
-                    }
-                };
-
-                let accumulator_messages = AccumulatorMessages::try_from_slice(&account.data);
-                match accumulator_messages {
-                    Ok(accumulator_messages) => {
-                        let (candidate, _) = Pubkey::find_program_address(
-                            &[
-                                b"AccumulatorState",
-                                &accumulator_messages.ring_index().to_be_bytes(),
-                            ],
-                            &system_program::id(),
-                        );
-
-                        if candidate.to_string() == update.value.pubkey {
-                            let store = store.clone();
-                            tokio::spawn(async move {
-                                if let Err(err) = crate::aggregate::store_update(
-                                    &store,
-                                    Update::AccumulatorMessages(accumulator_messages),
-                                )
-                                .await
-                                {
-                                    tracing::error!(error = ?err, "Failed to store accumulator messages.");
-                                }
-                            });
-                        } else {
-                            tracing::error!(
-                                ?candidate,
-                                ?update.value.pubkey,
-                                "Failed to verify message public keys.",
-                            );
-                        }
-                    }
-
-                    Err(err) => {
-                        tracing::error!(error = ?err, "Failed to parse AccumulatorMessages.");
-                    }
-                };
-            }
+    while let Some(update) = notif.next().await {
+        let account: Account = match update.value.account.decode() {
+            Some(account) => account,
             None => {
-                return Err(anyhow!("Pythnet network listener terminated"));
+                tracing::error!(?update, "Failed to decode account from update.");
+                continue;
             }
-        }
+        };
+
+        let accumulator_messages = AccumulatorMessages::try_from_slice(&account.data);
+        match accumulator_messages {
+            Ok(accumulator_messages) => {
+                let (candidate, _) = Pubkey::find_program_address(
+                    &[
+                        b"AccumulatorState",
+                        &accumulator_messages.ring_index().to_be_bytes(),
+                    ],
+                    &system_program::id(),
+                );
+
+                if candidate.to_string() == update.value.pubkey {
+                    let store = store.clone();
+                    tokio::spawn(async move {
+                        if let Err(err) = Aggregates::store_update(
+                            &*store,
+                            Update::AccumulatorMessages(accumulator_messages),
+                        )
+                        .await
+                        {
+                            tracing::error!(error = ?err, "Failed to store accumulator messages.");
+                        }
+                    });
+                } else {
+                    tracing::error!(
+                        ?candidate,
+                        ?update.value.pubkey,
+                        "Failed to verify message public keys.",
+                    );
+                }
+            }
+
+            Err(err) => {
+                tracing::error!(error = ?err, "Failed to parse AccumulatorMessages.");
+            }
+        };
     }
+
+    Err(anyhow!("Pythnet network listener connection terminated"))
 }
 
 /// Fetch existing GuardianSet accounts from Wormhole.

apps/hermes/src/network/wormhole.rs

@@ -7,7 +7,13 @@
 use {
     crate::{
         config::RunOptions,
-        state::State,
+        state::{
+            aggregate::{
+                Aggregates,
+                Update,
+            },
+            State,
+        },
     },
     anyhow::{
         anyhow,
@@ -43,7 +49,11 @@ use {
         Digest,
         Keccak256,
     },
-    std::sync::Arc,
+    std::{
+        sync::Arc,
+        time::Duration,
+    },
+    tokio::time::Instant,
     tonic::Request,
     wormhole_sdk::{
         vaa::{
@@ -100,10 +110,10 @@ pub struct BridgeConfig {
 /// GuardianSetData extracted from wormhole bridge account, due to no API.
 #[derive(borsh::BorshDeserialize)]
 pub struct GuardianSetData {
-    pub index:           u32,
-    pub keys:            Vec<[u8; 20]>,
-    pub creation_time:   u32,
-    pub expiration_time: u32,
+    pub _index:           u32,
+    pub keys:             Vec<[u8; 20]>,
+    pub _creation_time:   u32,
+    pub _expiration_time: u32,
 }
 
 /// Update the guardian set with the given ID in the state.
@@ -152,10 +162,16 @@ mod proto {
 pub async fn spawn(opts: RunOptions, state: Arc<State>) -> Result<()> {
     let mut exit = crate::EXIT.subscribe();
     loop {
+        let current_time = Instant::now();
         tokio::select! {
             _ = exit.changed() => break,
             Err(err) = run(opts.clone(), state.clone()) => {
                 tracing::error!(error = ?err, "Wormhole gRPC service failed.");
+
+                if current_time.elapsed() < Duration::from_secs(30) {
+                    tracing::error!("Wormhole listener restarting too quickly. Sleep 1s.");
+                    tokio::time::sleep(Duration::from_secs(1)).await;
+                }
             }
         }
     }
@@ -164,7 +180,7 @@ pub async fn spawn(opts: RunOptions, state: Arc<State>) -> Result<()> {
 }
 
 #[tracing::instrument(skip(opts, state))]
-async fn run(opts: RunOptions, state: Arc<State>) -> Result<()> {
+async fn run(opts: RunOptions, state: Arc<State>) -> Result<!> {
     let mut client = SpyRpcServiceClient::connect(opts.wormhole.spy_rpc_addr).await?;
     let mut stream = client
         .subscribe_signed_vaa(Request::new(SubscribeSignedVaaRequest {
@@ -184,7 +200,7 @@ async fn run(opts: RunOptions, state: Arc<State>) -> Result<()> {
         }
     }
 
-    Ok(())
+    Err(anyhow!("Wormhole gRPC stream terminated."))
 }
 
 /// Process a message received via a Wormhole gRPC connection.
@@ -225,7 +241,10 @@ pub async fn process_message(state: Arc<State>, vaa_bytes: Vec<u8>) -> Result<()
     )?;
 
     // Finally, store the resulting VAA in Hermes.
-    store_vaa(state.clone(), vaa.sequence, vaa_bytes).await?;
+    let sequence = vaa.sequence;
+    tokio::spawn(async move {
+        store_vaa(state.clone(), sequence, vaa_bytes).await;
+    });
 
     Ok(())
 }
@@ -334,16 +353,14 @@ pub fn verify_vaa<'a>(
 }
 
 #[tracing::instrument(skip(state, vaa_bytes))]
-pub async fn store_vaa(state: Arc<State>, sequence: u64, vaa_bytes: Vec<u8>) -> Result<()> {
+pub async fn store_vaa(state: Arc<State>, sequence: u64, vaa_bytes: Vec<u8>) {
     // Check VAA hasn't already been seen, this may have been checked previously
-    // but due to async nature It's possible other threads have mutated the state
+    // but due to async nature it's possible other threads have mutated the state
     // since this VAA started processing.
     let mut observed_vaa_seqs = state.observed_vaa_seqs.write().await;
-    ensure!(
-        !observed_vaa_seqs.contains(&sequence),
-        "Previously observed VAA: {}",
-        sequence,
-    );
+    if observed_vaa_seqs.contains(&sequence) {
+        return;
+    }
 
     // Clear old cached VAA sequences.
     while observed_vaa_seqs.len() > OBSERVED_CACHE_SIZE {
@@ -351,5 +368,7 @@ pub async fn store_vaa(state: Arc<State>, sequence: u64, vaa_bytes: Vec<u8>) ->
     }
 
     // Hand the VAA to the aggregate store.
-    crate::aggregate::store_update(&state, crate::aggregate::Update::Vaa(vaa_bytes)).await
+    if let Err(e) = Aggregates::store_update(&*state, Update::Vaa(vaa_bytes)).await {
+        tracing::error!(error = ?e, "Failed to store VAA in aggregate store.");
+    }
 }

apps/hermes/src/price_feeds_metadata.rs

@@ -31,6 +31,7 @@ impl<'a> From<&'a State> for &'a PriceFeedMetaState {
     }
 }
 
+#[async_trait::async_trait]
 pub trait PriceFeedMeta {
     async fn retrieve_price_feeds_metadata(&self) -> Result<Vec<PriceFeedMetadata>>;
     async fn store_price_feeds_metadata(
@@ -44,6 +45,7 @@ pub trait PriceFeedMeta {
     ) -> Result<Vec<PriceFeedMetadata>>;
 }
 
+#[async_trait::async_trait]
 impl<T> PriceFeedMeta for T
 where
     for<'a> &'a T: Into<&'a PriceFeedMetaState>,

apps/hermes/src/state.rs

@@ -2,14 +2,14 @@
 
 use {
     self::{
-        benchmarks::BenchmarksState,
-        cache::CacheState,
-    },
-    crate::{
         aggregate::{
             AggregateState,
             AggregationEvent,
         },
+        benchmarks::BenchmarksState,
+        cache::CacheState,
+    },
+    crate::{
         network::wormhole::GuardianSet,
         price_feeds_metadata::PriceFeedMetaState,
     },
@@ -28,6 +28,7 @@ use {
     },
 };
 
+pub mod aggregate;
 pub mod benchmarks;
 pub mod cache;
 
@@ -41,6 +42,9 @@ pub struct State {
     /// State for the `PriceFeedMeta` service for looking up metadata related to Pyth price feeds.
     pub price_feed_meta: PriceFeedMetaState,
 
+    /// State for accessing/storing Pyth price aggregates.
+    pub aggregates: AggregateState,
+
     /// Sequence numbers of lately observed Vaas. Store uses this set
     /// to ignore the previously observed Vaas as a performance boost.
     pub observed_vaa_seqs: RwLock<BTreeSet<u64>>,
@@ -48,12 +52,6 @@ pub struct State {
     /// Wormhole guardian sets. It is used to verify Vaas before using them.
     pub guardian_set: RwLock<BTreeMap<u32, GuardianSet>>,
 
-    /// The sender to the channel between Store and Api to notify completed updates.
-    pub api_update_tx: Sender<AggregationEvent>,
-
-    /// The aggregate module state.
-    pub aggregate_state: RwLock<AggregateState>,
-
     /// Metrics registry
     pub metrics_registry: RwLock<Registry>,
 }
@@ -69,10 +67,9 @@ impl State {
             cache:             CacheState::new(cache_size),
             benchmarks:        BenchmarksState::new(benchmarks_endpoint),
             price_feed_meta:   PriceFeedMetaState::new(),
+            aggregates:        AggregateState::new(update_tx, &mut metrics_registry),
             observed_vaa_seqs: RwLock::new(Default::default()),
             guardian_set:      RwLock::new(Default::default()),
-            api_update_tx:     update_tx,
-            aggregate_state:   RwLock::new(AggregateState::new(&mut metrics_registry)),
             metrics_registry:  RwLock::new(metrics_registry),
         })
     }

apps/hermes/src/state/aggregate.rs

@@ -20,6 +20,7 @@ use {
     },
     crate::{
         network::wormhole::VaaBytes,
+        price_feeds_metadata::PriceFeedMeta,
         state::{
             benchmarks::Benchmarks,
             cache::{
@@ -59,6 +60,13 @@ use {
         collections::HashSet,
         time::Duration,
     },
+    tokio::sync::{
+        broadcast::{
+            Receiver,
+            Sender,
+        },
+        RwLock,
+    },
     wormhole_sdk::Vaa,
 };
 
@@ -102,8 +110,7 @@ impl AggregationEvent {
     }
 }
 
-#[derive(Clone, Debug)]
-pub struct AggregateState {
+pub struct AggregateStateData {
     /// The latest completed slot. This is used to check whether a completed state is new or out of
     /// order.
     pub latest_completed_slot: Option<Slot>,
@@ -119,7 +126,7 @@ pub struct AggregateState {
     pub metrics: metrics::Metrics,
 }
 
-impl AggregateState {
+impl AggregateStateData {
     pub fn new(metrics_registry: &mut Registry) -> Self {
         Self {
             latest_completed_slot:      None,
@@ -130,6 +137,20 @@ impl AggregateState {
     }
 }
 
+pub struct AggregateState {
+    pub data:          RwLock<AggregateStateData>,
+    pub api_update_tx: Sender<AggregationEvent>,
+}
+
+impl AggregateState {
+    pub fn new(update_tx: Sender<AggregationEvent>, metrics_registry: &mut Registry) -> Self {
+        Self {
+            data:          RwLock::new(AggregateStateData::new(metrics_registry)),
+            api_update_tx: update_tx,
+        }
+    }
+}
+
 /// Accumulator messages coming from Pythnet validators.
 ///
 /// The validators writes the accumulator messages using Borsh with
@@ -177,125 +198,220 @@ const READINESS_STALENESS_THRESHOLD: Duration = Duration::from_secs(30);
 /// 10 slots is almost 5 seconds.
 const READINESS_MAX_ALLOWED_SLOT_LAG: Slot = 10;
 
-/// Stores the update data in the store
-#[tracing::instrument(skip(state, update))]
-pub async fn store_update(state: &State, update: Update) -> Result<()> {
-    // The slot that the update is originating from. It should be available
-    // in all the updates.
-    let slot = match update {
-        Update::Vaa(update_vaa) => {
-            let vaa = serde_wormhole::from_slice::<Vaa<&serde_wormhole::RawMessage>>(
-                update_vaa.as_ref(),
-            )?;
-            match WormholeMessage::try_from_bytes(vaa.payload)?.payload {
-                WormholePayload::Merkle(proof) => {
-                    tracing::info!(slot = proof.slot, "Storing VAA Merkle Proof.");
+#[async_trait::async_trait]
+pub trait Aggregates
+where
+    Self: Cache,
+    Self: Benchmarks,
+    Self: PriceFeedMeta,
+{
+    fn subscribe(&self) -> Receiver<AggregationEvent>;
+    async fn is_ready(&self) -> bool;
+    async fn store_update(&self, update: Update) -> Result<()>;
+    async fn get_price_feed_ids(&self) -> HashSet<PriceIdentifier>;
+    async fn get_price_feeds_with_update_data(
+        &self,
+        price_ids: &[PriceIdentifier],
+        request_time: RequestTime,
+    ) -> Result<PriceFeedsWithUpdateData>;
+}
 
-                    store_wormhole_merkle_verified_message(
-                        state,
-                        proof.clone(),
-                        update_vaa.to_owned(),
-                    )
-                    .await?;
+/// Allow downcasting State into CacheState for functions that depend on the `Cache` service.
+impl<'a> From<&'a State> for &'a AggregateState {
+    fn from(state: &'a State) -> &'a AggregateState {
+        &state.aggregates
+    }
+}
 
-                    state
-                        .aggregate_state
-                        .write()
-                        .await
-                        .metrics
-                        .observe(proof.slot, metrics::Event::Vaa);
+#[async_trait::async_trait]
+impl<T> Aggregates for T
+where
+    for<'a> &'a T: Into<&'a AggregateState>,
+    T: Sync,
+    T: Send,
+    T: Cache,
+    T: Benchmarks,
+    T: PriceFeedMeta,
+{
+    fn subscribe(&self) -> Receiver<AggregationEvent> {
+        self.into().api_update_tx.subscribe()
+    }
 
-                    proof.slot
+    /// Stores the update data in the store
+    #[tracing::instrument(skip(self, update))]
+    async fn store_update(&self, update: Update) -> Result<()> {
+        // The slot that the update is originating from. It should be available
+        // in all the updates.
+        let slot = match update {
+            Update::Vaa(update_vaa) => {
+                let vaa = serde_wormhole::from_slice::<Vaa<&serde_wormhole::RawMessage>>(
+                    update_vaa.as_ref(),
+                )?;
+                match WormholeMessage::try_from_bytes(vaa.payload)?.payload {
+                    WormholePayload::Merkle(proof) => {
+                        tracing::info!(slot = proof.slot, "Storing VAA Merkle Proof.");
+
+                        store_wormhole_merkle_verified_message(
+                            self,
+                            proof.clone(),
+                            update_vaa.to_owned(),
+                        )
+                        .await?;
+
+                        self.into()
+                            .data
+                            .write()
+                            .await
+                            .metrics
+                            .observe(proof.slot, metrics::Event::Vaa);
+
+                        proof.slot
+                    }
                 }
             }
-        }
-        Update::AccumulatorMessages(accumulator_messages) => {
-            let slot = accumulator_messages.slot;
-            tracing::info!(slot = slot, "Storing Accumulator Messages.");
+            Update::AccumulatorMessages(accumulator_messages) => {
+                let slot = accumulator_messages.slot;
+                tracing::info!(slot = slot, "Storing Accumulator Messages.");
 
-            state
-                .store_accumulator_messages(accumulator_messages)
-                .await?;
+                self.store_accumulator_messages(accumulator_messages)
+                    .await?;
 
-            state
-                .aggregate_state
-                .write()
-                .await
-                .metrics
-                .observe(slot, metrics::Event::AccumulatorMessages);
-            slot
-        }
-    };
-
-    // Update the aggregate state with the latest observed slot
-    {
-        let mut aggregate_state = state.aggregate_state.write().await;
-        aggregate_state.latest_observed_slot = aggregate_state
-            .latest_observed_slot
-            .map(|latest| latest.max(slot))
-            .or(Some(slot));
-    }
-
-    let accumulator_messages = state.fetch_accumulator_messages(slot).await?;
-    let wormhole_merkle_state = state.fetch_wormhole_merkle_state(slot).await?;
-
-    let (accumulator_messages, wormhole_merkle_state) =
-        match (accumulator_messages, wormhole_merkle_state) {
-            (Some(accumulator_messages), Some(wormhole_merkle_state)) => {
-                (accumulator_messages, wormhole_merkle_state)
+                self.into()
+                    .data
+                    .write()
+                    .await
+                    .metrics
+                    .observe(slot, metrics::Event::AccumulatorMessages);
+                slot
             }
-            _ => return Ok(()),
         };
 
-    tracing::info!(slot = wormhole_merkle_state.root.slot, "Completed Update.");
-
-    // Once the accumulator reaches a complete state for a specific slot
-    // we can build the message states
-    let message_states = build_message_states(accumulator_messages, wormhole_merkle_state)?;
-
-    let message_state_keys = message_states
-        .iter()
-        .map(|message_state| message_state.key())
-        .collect::<HashSet<_>>();
-
-    tracing::info!(len = message_states.len(), "Storing Message States.");
-    state.store_message_states(message_states).await?;
-
-    // Update the aggregate state
-    let mut aggregate_state = state.aggregate_state.write().await;
-
-    // Check if the update is new or out of order
-    match aggregate_state.latest_completed_slot {
-        None => {
-            aggregate_state.latest_completed_slot.replace(slot);
-            state.api_update_tx.send(AggregationEvent::New { slot })?;
+        // Update the aggregate state with the latest observed slot
+        {
+            let mut aggregate_state = self.into().data.write().await;
+            aggregate_state.latest_observed_slot = aggregate_state
+                .latest_observed_slot
+                .map(|latest| latest.max(slot))
+                .or(Some(slot));
         }
-        Some(latest) if slot > latest => {
-            state.prune_removed_keys(message_state_keys).await;
-            aggregate_state.latest_completed_slot.replace(slot);
-            state.api_update_tx.send(AggregationEvent::New { slot })?;
-        }
-        _ => {
-            state
+
+        let accumulator_messages = self.fetch_accumulator_messages(slot).await?;
+        let wormhole_merkle_state = self.fetch_wormhole_merkle_state(slot).await?;
+
+        let (accumulator_messages, wormhole_merkle_state) =
+            match (accumulator_messages, wormhole_merkle_state) {
+                (Some(accumulator_messages), Some(wormhole_merkle_state)) => {
+                    (accumulator_messages, wormhole_merkle_state)
+                }
+                _ => return Ok(()),
+            };
+
+        tracing::info!(slot = wormhole_merkle_state.root.slot, "Completed Update.");
+
+        // Once the accumulator reaches a complete state for a specific slot
+        // we can build the message states
+        let message_states = build_message_states(accumulator_messages, wormhole_merkle_state)?;
+
+        let message_state_keys = message_states
+            .iter()
+            .map(|message_state| message_state.key())
+            .collect::<HashSet<_>>();
+
+        tracing::info!(len = message_states.len(), "Storing Message States.");
+        self.store_message_states(message_states).await?;
+
+        // Update the aggregate state
+        let mut aggregate_state = self.into().data.write().await;
+
+        // Send update event to subscribers. We are purposefully ignoring the result
+        // because there might be no subscribers.
+        let _ = match aggregate_state.latest_completed_slot {
+            None => {
+                aggregate_state.latest_completed_slot.replace(slot);
+                self.into()
+                    .api_update_tx
+                    .send(AggregationEvent::New { slot })
+            }
+            Some(latest) if slot > latest => {
+                self.prune_removed_keys(message_state_keys).await;
+                aggregate_state.latest_completed_slot.replace(slot);
+                self.into()
+                    .api_update_tx
+                    .send(AggregationEvent::New { slot })
+            }
+            _ => self
+                .into()
                 .api_update_tx
-                .send(AggregationEvent::OutOfOrder { slot })?;
+                .send(AggregationEvent::OutOfOrder { slot }),
+        };
+
+        aggregate_state.latest_completed_slot = aggregate_state
+            .latest_completed_slot
+            .map(|latest| latest.max(slot))
+            .or(Some(slot));
+
+        aggregate_state
+            .latest_completed_update_at
+            .replace(Instant::now());
+
+        aggregate_state
+            .metrics
+            .observe(slot, metrics::Event::CompletedUpdate);
+
+        Ok(())
+    }
+
+    async fn get_price_feeds_with_update_data(
+        &self,
+        price_ids: &[PriceIdentifier],
+        request_time: RequestTime,
+    ) -> Result<PriceFeedsWithUpdateData> {
+        match get_verified_price_feeds(self, price_ids, request_time.clone()).await {
+            Ok(price_feeds_with_update_data) => Ok(price_feeds_with_update_data),
+            Err(e) => {
+                if let RequestTime::FirstAfter(publish_time) = request_time {
+                    return Benchmarks::get_verified_price_feeds(self, price_ids, publish_time)
+                        .await;
+                }
+                Err(e)
+            }
         }
     }
 
-    aggregate_state.latest_completed_slot = aggregate_state
-        .latest_completed_slot
-        .map(|latest| latest.max(slot))
-        .or(Some(slot));
+    async fn get_price_feed_ids(&self) -> HashSet<PriceIdentifier> {
+        Cache::message_state_keys(self)
+            .await
+            .iter()
+            .map(|key| PriceIdentifier::new(key.feed_id))
+            .collect()
+    }
 
-    aggregate_state
-        .latest_completed_update_at
-        .replace(Instant::now());
+    async fn is_ready(&self) -> bool {
+        let metadata = self.into().data.read().await;
+        let price_feeds_metadata = PriceFeedMeta::retrieve_price_feeds_metadata(self)
+            .await
+            .unwrap();
 
-    aggregate_state
-        .metrics
-        .observe(slot, metrics::Event::CompletedUpdate);
+        let has_completed_recently = match metadata.latest_completed_update_at.as_ref() {
+            Some(latest_completed_update_time) => {
+                latest_completed_update_time.elapsed() < READINESS_STALENESS_THRESHOLD
+            }
+            None => false,
+        };
 
-    Ok(())
+        let is_not_behind = match (
+            metadata.latest_completed_slot,
+            metadata.latest_observed_slot,
+        ) {
+            (Some(latest_completed_slot), Some(latest_observed_slot)) => {
+                latest_observed_slot - latest_completed_slot <= READINESS_MAX_ALLOWED_SLOT_LAG
+            }
+            _ => false,
+        };
+
+        let is_metadata_loaded = !price_feeds_metadata.is_empty();
+        has_completed_recently && is_not_behind && is_metadata_loaded
+    }
 }
 
 #[tracing::instrument(skip(accumulator_messages, wormhole_merkle_state))]
@@ -390,73 +506,12 @@ where
     })
 }
 
-pub async fn get_price_feeds_with_update_data<S>(
-    state: &S,
-    price_ids: &[PriceIdentifier],
-    request_time: RequestTime,
-) -> Result<PriceFeedsWithUpdateData>
-where
-    S: Cache,
-    S: Benchmarks,
-{
-    match get_verified_price_feeds(state, price_ids, request_time.clone()).await {
-        Ok(price_feeds_with_update_data) => Ok(price_feeds_with_update_data),
-        Err(e) => {
-            if let RequestTime::FirstAfter(publish_time) = request_time {
-                return state
-                    .get_verified_price_feeds(price_ids, publish_time)
-                    .await;
-            }
-            Err(e)
-        }
-    }
-}
-
-pub async fn get_price_feed_ids<S>(state: &S) -> HashSet<PriceIdentifier>
-where
-    S: Cache,
-{
-    state
-        .message_state_keys()
-        .await
-        .iter()
-        .map(|key| PriceIdentifier::new(key.feed_id))
-        .collect()
-}
-
-pub async fn is_ready(state: &State) -> bool {
-    let metadata = state.aggregate_state.read().await;
-    let price_feeds_metadata = state.price_feed_meta.data.read().await;
-
-    let has_completed_recently = match metadata.latest_completed_update_at.as_ref() {
-        Some(latest_completed_update_time) => {
-            latest_completed_update_time.elapsed() < READINESS_STALENESS_THRESHOLD
-        }
-        None => false,
-    };
-
-    let is_not_behind = match (
-        metadata.latest_completed_slot,
-        metadata.latest_observed_slot,
-    ) {
-        (Some(latest_completed_slot), Some(latest_observed_slot)) => {
-            latest_observed_slot - latest_completed_slot <= READINESS_MAX_ALLOWED_SLOT_LAG
-        }
-        _ => false,
-    };
-
-    let is_metadata_loaded = !price_feeds_metadata.is_empty();
-
-    has_completed_recently && is_not_behind && is_metadata_loaded
-}
-
 #[cfg(test)]
 mod test {
     use {
         super::*,
         crate::{
             api::types::PriceFeedMetadata,
-            price_feeds_metadata::PriceFeedMeta,
             state::test::setup_state,
         },
         futures::future::join_all,
@@ -558,7 +613,7 @@ mod test {
     }
 
     pub async fn store_multiple_concurrent_valid_updates(state: Arc<State>, updates: Vec<Update>) {
-        let res = join_all(updates.into_iter().map(|u| store_update(&state, u))).await;
+        let res = join_all(updates.into_iter().map(|u| (&state).store_update(u))).await;
         // Check that all store_update calls succeeded
         assert!(res.into_iter().all(|r| r.is_ok()));
     }
@@ -584,19 +639,19 @@ mod test {
 
         // Check the price ids are stored correctly
         assert_eq!(
-            get_price_feed_ids(&*state).await,
+            (&*state).get_price_feed_ids().await,
             vec![PriceIdentifier::new([100; 32])].into_iter().collect()
         );
 
         // Check get_price_feeds_with_update_data retrieves the correct
         // price feed with correct update data.
-        let price_feeds_with_update_data = get_price_feeds_with_update_data(
-            &*state,
-            &[PriceIdentifier::new([100; 32])],
-            RequestTime::Latest,
-        )
-        .await
-        .unwrap();
+        let price_feeds_with_update_data = (&*state)
+            .get_price_feeds_with_update_data(
+                &[PriceIdentifier::new([100; 32])],
+                RequestTime::Latest,
+            )
+            .await
+            .unwrap();
 
         assert_eq!(
             price_feeds_with_update_data.price_feeds,
@@ -709,7 +764,7 @@ mod test {
 
         // Check the price ids are stored correctly
         assert_eq!(
-            get_price_feed_ids(&*state).await,
+            (&*state).get_price_feed_ids().await,
             vec![
                 PriceIdentifier::new([100; 32]),
                 PriceIdentifier::new([200; 32])
@@ -719,13 +774,13 @@ mod test {
         );
 
         // Check that price feed 2 exists
-        assert!(get_price_feeds_with_update_data(
-            &*state,
-            &[PriceIdentifier::new([200; 32])],
-            RequestTime::Latest,
-        )
-        .await
-        .is_ok());
+        assert!((&*state)
+            .get_price_feeds_with_update_data(
+                &[PriceIdentifier::new([200; 32])],
+                RequestTime::Latest,
+            )
+            .await
+            .is_ok());
 
         // Now send an update with only price feed 1 (without price feed 2)
         // and make sure that price feed 2 is not stored anymore.
@@ -746,17 +801,17 @@ mod test {
 
         // Check that price feed 2 does not exist anymore
         assert_eq!(
-            get_price_feed_ids(&*state).await,
+            (&*state).get_price_feed_ids().await,
             vec![PriceIdentifier::new([100; 32]),].into_iter().collect()
         );
 
-        assert!(get_price_feeds_with_update_data(
-            &*state,
-            &[PriceIdentifier::new([200; 32])],
-            RequestTime::Latest,
-        )
-        .await
-        .is_err());
+        assert!((&*state)
+            .get_price_feeds_with_update_data(
+                &[PriceIdentifier::new([200; 32])],
+                RequestTime::Latest,
+            )
+            .await
+            .is_err());
     }
 
     #[tokio::test]
@@ -792,13 +847,13 @@ mod test {
         MockClock::advance(Duration::from_secs(1));
 
         // Get the price feeds with update data
-        let price_feeds_with_update_data = get_price_feeds_with_update_data(
-            &*state,
-            &[PriceIdentifier::new([100; 32])],
-            RequestTime::Latest,
-        )
-        .await
-        .unwrap();
+        let price_feeds_with_update_data = (&*state)
+            .get_price_feeds_with_update_data(
+                &[PriceIdentifier::new([100; 32])],
+                RequestTime::Latest,
+            )
+            .await
+            .unwrap();
 
         // check received_at is correct
         assert_eq!(price_feeds_with_update_data.price_feeds.len(), 1);
@@ -818,13 +873,13 @@ mod test {
             .unwrap();
 
         // Check the state is ready
-        assert!(is_ready(&state).await);
+        assert!((&state).is_ready().await);
 
         // Advance the clock to make the prices stale
         MockClock::advance_system_time(READINESS_STALENESS_THRESHOLD);
         MockClock::advance(READINESS_STALENESS_THRESHOLD);
         // Check the state is not ready
-        assert!(!is_ready(&state).await);
+        assert!(!(&state).is_ready().await);
     }
 
     /// Test that the state retains the latest slots upon cache eviction.
@@ -867,16 +922,16 @@ mod test {
 
         // Check the last 100 slots are retained
         for slot in 900..1000 {
-            let price_feeds_with_update_data = get_price_feeds_with_update_data(
-                &*state,
-                &[
-                    PriceIdentifier::new([100; 32]),
-                    PriceIdentifier::new([200; 32]),
-                ],
-                RequestTime::FirstAfter(slot as i64),
-            )
-            .await
-            .unwrap();
+            let price_feeds_with_update_data = (&*state)
+                .get_price_feeds_with_update_data(
+                    &[
+                        PriceIdentifier::new([100; 32]),
+                        PriceIdentifier::new([200; 32]),
+                    ],
+                    RequestTime::FirstAfter(slot as i64),
+                )
+                .await
+                .unwrap();
             assert_eq!(price_feeds_with_update_data.price_feeds.len(), 2);
             assert_eq!(price_feeds_with_update_data.price_feeds[0].slot, Some(slot));
             assert_eq!(price_feeds_with_update_data.price_feeds[1].slot, Some(slot));
@@ -884,16 +939,16 @@ mod test {
 
         // Check nothing else is retained
         for slot in 0..900 {
-            assert!(get_price_feeds_with_update_data(
-                &*state,
-                &[
-                    PriceIdentifier::new([100; 32]),
-                    PriceIdentifier::new([200; 32])
-                ],
-                RequestTime::FirstAfter(slot as i64),
-            )
-            .await
-            .is_err());
+            assert!((&*state)
+                .get_price_feeds_with_update_data(
+                    &[
+                        PriceIdentifier::new([100; 32]),
+                        PriceIdentifier::new([200; 32])
+                    ],
+                    RequestTime::FirstAfter(slot as i64),
+                )
+                .await
+                .is_err());
         }
     }
 }

apps/hermes/src/state/benchmarks.rs

@@ -1,14 +1,14 @@
 //! This module communicates with Pyth Benchmarks, an API for historical price feeds and their updates.
 
 use {
-    super::State,
-    crate::{
+    super::{
         aggregate::{
             PriceFeedsWithUpdateData,
             UnixTimestamp,
         },
-        api::types::PriceUpdate,
+        State,
     },
+    crate::api::types::PriceUpdate,
     anyhow::Result,
     base64::{
         engine::general_purpose::STANDARD as base64_standard_engine,
@@ -69,6 +69,7 @@ impl<'a> From<&'a State> for &'a BenchmarksState {
     }
 }
 
+#[async_trait::async_trait]
 pub trait Benchmarks {
     async fn get_verified_price_feeds(
         &self,
@@ -77,6 +78,7 @@ pub trait Benchmarks {
     ) -> Result<PriceFeedsWithUpdateData>;
 }
 
+#[async_trait::async_trait]
 impl<T> Benchmarks for T
 where
     for<'a> &'a T: Into<&'a BenchmarksState>,

apps/hermes/src/state/cache.rs

@@ -1,6 +1,6 @@
 use {
     super::State,
-    crate::aggregate::{
+    crate::state::aggregate::{
         wormhole_merkle::WormholeMerkleState,
         AccumulatorMessages,
         ProofSet,
@@ -132,16 +132,10 @@ impl<'a> From<&'a State> for &'a CacheState {
     }
 }
 
+#[async_trait::async_trait]
 pub trait Cache {
-    async fn message_state_keys(&self) -> Vec<MessageStateKey>;
     async fn store_message_states(&self, message_states: Vec<MessageState>) -> Result<()>;
     async fn prune_removed_keys(&self, current_keys: HashSet<MessageStateKey>);
-    async fn fetch_message_states(
-        &self,
-        ids: Vec<FeedId>,
-        request_time: RequestTime,
-        filter: MessageStateFilter,
-    ) -> Result<Vec<MessageState>>;
     async fn store_accumulator_messages(
         &self,
         accumulator_messages: AccumulatorMessages,
@@ -152,8 +146,16 @@ pub trait Cache {
         wormhole_merkle_state: WormholeMerkleState,
     ) -> Result<()>;
     async fn fetch_wormhole_merkle_state(&self, slot: Slot) -> Result<Option<WormholeMerkleState>>;
+    async fn message_state_keys(&self) -> Vec<MessageStateKey>;
+    async fn fetch_message_states(
+        &self,
+        ids: Vec<FeedId>,
+        request_time: RequestTime,
+        filter: MessageStateFilter,
+    ) -> Result<Vec<MessageState>>;
 }
 
+#[async_trait::async_trait]
 impl<T> Cache for T
 where
     for<'a> &'a T: Into<&'a CacheState>,
@@ -322,9 +324,9 @@ async fn retrieve_message_state(
 mod test {
     use {
         super::*,
-        crate::{
+        crate::state::{
             aggregate::wormhole_merkle::WormholeMerkleMessageProof,
-            state::test::setup_state,
+            test::setup_state,
         },
         pyth_sdk::UnixTimestamp,
         pythnet_sdk::{