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]
@@ -16,7 +16,7 @@ ethers      = { version = "2.0.14", features = ["ws"] }
 futures            = { version = "0.3.28" }
 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(
+        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,
-            &metadata.seed,
-            metadata.chain_length,
+                &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.3"
+version = "0.5.9"
 dependencies = [
  "anyhow",
  "async-trait",
@@ -1839,6 +1839,7 @@ dependencies = [
  "solana-sdk",
  "strum",
  "tokio",
+ "tokio-stream",
  "tonic",
  "tonic-build",
  "tower-http",
@@ -3137,7 +3138,7 @@ dependencies = [
 
 [[package]]
 name = "pythnet-sdk"
-version = "2.0.0"
+version = "2.1.0"
 dependencies = [
  "bincode",
  "borsh 0.10.3",
@@ -5188,9 +5189,9 @@ dependencies = [
 
 [[package]]
 name = "termcolor"
-version = "1.4.1"
+version = "1.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "06794f8f6c5c898b3275aebefa6b8a1cb24cd2c6c79397ab15774837a0bc5755"
+checksum = "bab24d30b911b2376f3a13cc2cd443142f0c81dda04c118693e35b3835757755"
 dependencies = [
  "winapi-util",
 ]
@@ -5385,6 +5386,7 @@ dependencies = [
  "futures-core",
  "pin-project-lite",
  "tokio",
+ "tokio-util",
 ]
 
 [[package]]

apps/hermes/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name        = "hermes"
-version     = "0.5.3"
+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"] }
@@ -42,6 +42,7 @@ serde_wormhole     = { git     = "https://github.com/wormhole-foundation/wormhol
 sha3               = { version = "0.10.4" }
 strum              = { version = "0.24.1", features = ["derive"] }
 tokio              = { version = "1.26.0", features = ["full"] }
+tokio-stream       = { version = "0.1.15", features = ["full"] }
 tonic              = { version = "0.10.1", features = ["tls"] }
 tower-http         = { version = "0.4.0", features = ["cors"] }
 tracing            = { version = "0.1.37", features = ["log"] }

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,
@@ -26,20 +24,29 @@ mod rest;
 pub mod types;
 mod ws;
 
-#[derive(Clone)]
-pub struct ApiState {
-    pub state:     Arc<State>,
+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>,
 }
 
-impl ApiState {
+/// Manually implement `Clone` as the derive macro will try and slap `Clone` on
+/// `State` which should not be Clone.
+impl<S> Clone for ApiState<S> {
+    fn clone(&self) -> Self {
+        Self {
+            state:   self.state.clone(),
+            ws:      self.ws.clone(),
+            metrics: self.metrics.clone(),
+        }
+    }
+}
+
+impl ApiState<State> {
     pub fn new(
         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())),
@@ -49,24 +56,18 @@ impl ApiState {
                 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,
         )
     };
 
@@ -93,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(
@@ -122,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))
@@ -131,6 +134,10 @@ pub async fn run(opts: RunOptions, state: ApiState) -> Result<()> {
         .route("/api/latest_price_feeds", get(rest::latest_price_feeds))
         .route("/api/latest_vaas", get(rest::latest_vaas))
         .route("/api/price_feed_ids", get(rest::price_feed_ids))
+        .route(
+            "/v2/updates/price/stream",
+            get(rest::price_stream_sse_handler),
+        )
         .route("/v2/updates/price/latest", get(rest::latest_price_updates))
         .route(
             "/v2/updates/price/:publish_time",

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::{
@@ -21,6 +22,7 @@ mod price_feed_ids;
 mod ready;
 mod v2;
 
+
 pub use {
     get_price_feed::*,
     get_vaa::*,
@@ -34,10 +36,12 @@ pub use {
     v2::{
         latest_price_updates::*,
         price_feeds_metadata::*,
+        sse::*,
         timestamp_price_updates::*,
     },
 };
 
+#[derive(Debug)]
 pub enum RestError {
     BenchmarkPriceNotUnique,
     UpdateDataNotFound,
@@ -90,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,19 +65,20 @@ 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,
-    )
+    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!(

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,19 +60,20 @@ 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,
-    )
+    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!(

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

@@ -1,7 +1,11 @@
 use {
-    crate::api::{
+    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,19 +77,19 @@ 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,
-    )
+    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!(

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/price_feeds_metadata.rs

@@ -6,8 +6,9 @@ use {
                 AssetType,
                 PriceFeedMetadata,
             },
+            ApiState,
         },
-        price_feeds_metadata::get_price_feeds_metadata,
+        price_feeds_metadata::PriceFeedMeta,
     },
     anyhow::Result,
     axum::{
@@ -46,12 +47,16 @@ pub struct PriceFeedsMetadataQueryParams {
         PriceFeedsMetadataQueryParams
     )
 )]
-pub async fn price_feeds_metadata(
-    State(state): State<crate::api::ApiState>,
+pub async fn price_feeds_metadata<S>(
+    State(state): State<ApiState<S>>,
     QsQuery(params): QsQuery<PriceFeedsMetadataQueryParams>,
-) -> Result<Json<Vec<PriceFeedMetadata>>, RestError> {
-    let price_feeds_metadata =
-        get_price_feeds_metadata(&state.state, params.query, params.asset_type)
+) -> Result<Json<Vec<PriceFeedMetadata>>, RestError>
+where
+    S: PriceFeedMeta,
+{
+    let state = &state.state;
+    let price_feeds_metadata = state
+        .get_price_feeds_metadata(params.query, params.asset_type)
         .await
         .map_err(|e| {
             tracing::warn!("RPC connection error: {}", e);

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

@@ -0,0 +1,235 @@
+use {
+    crate::{
+        api::{
+            rest::{
+                verify_price_ids_exist,
+                RestError,
+            },
+            types::{
+                BinaryPriceUpdate,
+                EncodingType,
+                ParsedPriceUpdate,
+                PriceIdInput,
+                PriceUpdate,
+                RpcPriceIdentifier,
+            },
+            ApiState,
+        },
+        state::aggregate::{
+            Aggregates,
+            AggregationEvent,
+            RequestTime,
+        },
+    },
+    anyhow::Result,
+    axum::{
+        extract::State,
+        response::sse::{
+            Event,
+            KeepAlive,
+            Sse,
+        },
+    },
+    futures::Stream,
+    pyth_sdk::PriceIdentifier,
+    serde::Deserialize,
+    serde_qs::axum::QsQuery,
+    std::convert::Infallible,
+    tokio::sync::broadcast,
+    tokio_stream::{
+        wrappers::BroadcastStream,
+        StreamExt as _,
+    },
+    utoipa::IntoParams,
+};
+
+#[derive(Debug, Deserialize, IntoParams)]
+#[into_params(parameter_in = Query)]
+pub struct StreamPriceUpdatesQueryParams {
+    /// Get the most recent price update for this set of price feed ids.
+    ///
+    /// This parameter can be provided multiple times to retrieve multiple price updates,
+    /// for example see the following query string:
+    ///
+    /// ```
+    /// ?ids[]=a12...&ids[]=b4c...
+    /// ```
+    #[param(rename = "ids[]")]
+    #[param(example = "e62df6c8b4a85fe1a67db44dc12de5db330f7ac66b72dc658afedf0f4a415b43")]
+    ids: Vec<PriceIdInput>,
+
+    /// 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. 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 {
+    true
+}
+
+#[utoipa::path(
+    get,
+    path = "/v2/updates/price/stream",
+    responses(
+        (status = 200, description = "Price updates retrieved successfully", body = PriceUpdate),
+        (status = 404, description = "Price ids not found", body = String)
+    ),
+    params(StreamPriceUpdatesQueryParams)
+)]
+/// SSE route handler for streaming price updates.
+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>
+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> = Aggregates::subscribe(&*state.state);
+
+    // Convert the broadcast receiver into a Stream
+    let stream = BroadcastStream::new(update_rx);
+
+    let sse_stream = stream.then(move |message| {
+        let state_clone = state.clone(); // Clone again to use inside the async block
+        let price_ids_clone = price_ids.clone(); // Clone again for use inside the async block
+        async move {
+            match message {
+                Ok(event) => {
+                    match handle_aggregation_event(
+                        event,
+                        state_clone,
+                        price_ids_clone,
+                        params.encoding,
+                        params.parsed,
+                        params.benchmarks_only,
+                        params.allow_unordered,
+                    )
+                    .await
+                    {
+                        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)),
+                    }
+                }
+                Err(e) => Ok(error_event(e)),
+            }
+        }
+    });
+
+    Ok(Sse::new(sse_stream).keep_alive(KeepAlive::default()))
+}
+
+async fn handle_aggregation_event<S>(
+    event: AggregationEvent,
+    state: ApiState<S>,
+    mut price_ids: Vec<PriceIdentifier>,
+    encoding: EncodingType,
+    parsed: bool,
+    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 = Aggregates::get_price_feed_ids(&*state.state).await;
+
+    price_ids.retain(|price_feed_id| available_price_feed_ids.contains(price_feed_id));
+
+    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()
+        .map(|data| encoding.encode_str(&data))
+        .collect();
+    let binary_price_update = BinaryPriceUpdate {
+        encoding,
+        data: encoded_data,
+    };
+
+    Ok(Some(PriceUpdate {
+        binary: binary_price_update,
+        parsed: if parsed {
+            Some(parsed_price_updates)
+        } else {
+            None
+        },
+    }))
+}
+
+fn error_event<E: std::fmt::Debug>(e: E) -> Event {
+    Event::default()
+        .event("error")
+        .data(format!("Error receiving update: {:?}", e))
+}

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::{
+    super::{
+        types::{
             PriceIdInput,
             RpcPriceFeed,
         },
-    crate::{
+        ApiState,
+    },
+    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;
@@ -35,7 +34,7 @@ lazy_static! {
     /// - Exit logic doesn't really require carefully threading this value through the app.
     /// - The `Receiver` side of a watch channel performs the detection based on if the change
     ///   happened after the subscribe, so it means all listeners should always be notified
-    ///   currectly.
+    ///   correctly.
     pub static ref EXIT: watch::Sender<bool> = watch::channel(false).0;
 }
 
@@ -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 {
@@ -66,11 +65,11 @@ 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([
-                Box::pin(spawn(network::wormhole::spawn(opts.clone(), store.clone()))),
-                Box::pin(spawn(network::pythnet::spawn(opts.clone(), store.clone()))),
-                Box::pin(spawn(metrics_server::run(opts.clone(), store.clone()))),
-                Box::pin(spawn(api::spawn(opts.clone(), store.clone(), update_tx))),
+            let tasks = join_all(vec![
+                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::{
@@ -17,10 +13,17 @@ use {
             GuardianSetData,
         },
         price_feeds_metadata::{
-            store_price_feeds_metadata,
+            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,9 +160,7 @@ 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) => {
+    while let Some(update) = notif.next().await {
         let account: Account = match update.value.account.decode() {
             Some(account) => account,
             None => {
@@ -182,8 +183,8 @@ pub async fn run(store: Arc<State>, pythnet_ws_endpoint: String) -> Result<()> {
                 if candidate.to_string() == update.value.pubkey {
                     let store = store.clone();
                     tokio::spawn(async move {
-                                if let Err(err) = crate::aggregate::store_update(
-                                    &store,
+                        if let Err(err) = Aggregates::store_update(
+                            &*store,
                             Update::AccumulatorMessages(accumulator_messages),
                         )
                         .await
@@ -205,11 +206,8 @@ pub async fn run(store: Arc<State>, pythnet_ws_endpoint: String) -> Result<()> {
             }
         };
     }
-            None => {
-                return Err(anyhow!("Pythnet network listener terminated"));
-            }
-        }
-    }
+
+    Err(anyhow!("Pythnet network listener connection terminated"))
 }
 
 /// Fetch existing GuardianSet accounts from Wormhole.
@@ -325,6 +323,19 @@ pub async fn spawn(opts: RunOptions, state: Arc<State>) -> Result<()> {
         let price_feeds_state = state.clone();
         let mut exit = crate::EXIT.subscribe();
         tokio::spawn(async move {
+            // Run fetch and store once before the loop
+            if let Err(e) = fetch_and_store_price_feeds_metadata(
+                price_feeds_state.as_ref(),
+                &opts.pythnet.mapping_addr,
+                &rpc_client,
+            )
+            .await
+            {
+                tracing::error!(
+                    "Error in initial fetching and storing price feeds metadata: {}",
+                    e
+                );
+            }
             loop {
                 tokio::select! {
                     _ = exit.changed() => break,
@@ -353,13 +364,18 @@ pub async fn spawn(opts: RunOptions, state: Arc<State>) -> Result<()> {
 }
 
 
-pub async fn fetch_and_store_price_feeds_metadata(
-    state: &State,
+pub async fn fetch_and_store_price_feeds_metadata<S>(
+    state: &S,
     mapping_address: &Pubkey,
     rpc_client: &RpcClient,
-) -> Result<Vec<PriceFeedMetadata>> {
+) -> Result<Vec<PriceFeedMetadata>>
+where
+    S: PriceFeedMeta,
+{
     let price_feeds_metadata = fetch_price_feeds_metadata(mapping_address, rpc_client).await?;
-    store_price_feeds_metadata(state, &price_feeds_metadata).await?;
+    state
+        .store_price_feeds_metadata(&price_feeds_metadata)
+        .await?;
     Ok(price_feeds_metadata)
 }
 

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 _index:           u32,
     pub keys:             Vec<[u8; 20]>,
-    pub creation_time:   u32,
-    pub expiration_time: u32,
+    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

@@ -0,0 +1,96 @@
+use {
+    crate::{
+        api::types::{
+            AssetType,
+            PriceFeedMetadata,
+        },
+        state::State,
+    },
+    anyhow::Result,
+    tokio::sync::RwLock,
+};
+
+pub const DEFAULT_PRICE_FEEDS_CACHE_UPDATE_INTERVAL: u64 = 600;
+
+pub struct PriceFeedMetaState {
+    pub data: RwLock<Vec<PriceFeedMetadata>>,
+}
+
+impl PriceFeedMetaState {
+    pub fn new() -> Self {
+        Self {
+            data: RwLock::new(Vec::new()),
+        }
+    }
+}
+
+/// Allow downcasting State into CacheState for functions that depend on the `Cache` service.
+impl<'a> From<&'a State> for &'a PriceFeedMetaState {
+    fn from(state: &'a State) -> &'a PriceFeedMetaState {
+        &state.price_feed_meta
+    }
+}
+
+#[async_trait::async_trait]
+pub trait PriceFeedMeta {
+    async fn retrieve_price_feeds_metadata(&self) -> Result<Vec<PriceFeedMetadata>>;
+    async fn store_price_feeds_metadata(
+        &self,
+        price_feeds_metadata: &[PriceFeedMetadata],
+    ) -> Result<()>;
+    async fn get_price_feeds_metadata(
+        &self,
+        query: Option<String>,
+        asset_type: Option<AssetType>,
+    ) -> Result<Vec<PriceFeedMetadata>>;
+}
+
+#[async_trait::async_trait]
+impl<T> PriceFeedMeta for T
+where
+    for<'a> &'a T: Into<&'a PriceFeedMetaState>,
+    T: Sync,
+{
+    async fn retrieve_price_feeds_metadata(&self) -> Result<Vec<PriceFeedMetadata>> {
+        let price_feeds_metadata = self.into().data.read().await;
+        Ok(price_feeds_metadata.clone())
+    }
+
+    async fn store_price_feeds_metadata(
+        &self,
+        price_feeds_metadata: &[PriceFeedMetadata],
+    ) -> Result<()> {
+        let mut price_feeds_metadata_write_guard = self.into().data.write().await;
+        *price_feeds_metadata_write_guard = price_feeds_metadata.to_vec();
+        Ok(())
+    }
+
+
+    async fn get_price_feeds_metadata(
+        &self,
+        query: Option<String>,
+        asset_type: Option<AssetType>,
+    ) -> Result<Vec<PriceFeedMetadata>> {
+        let mut price_feeds_metadata = self.retrieve_price_feeds_metadata().await?;
+
+        // Filter by query if provided
+        if let Some(query_str) = &query {
+            price_feeds_metadata.retain(|feed| {
+                feed.attributes.get("symbol").map_or(false, |symbol| {
+                    symbol.to_lowercase().contains(&query_str.to_lowercase())
+                })
+            });
+        }
+
+        // Filter by asset_type if provided
+        if let Some(asset_type) = &asset_type {
+            price_feeds_metadata.retain(|feed| {
+                feed.attributes.get("asset_type").map_or(false, |type_str| {
+                    type_str.to_lowercase() == asset_type.to_string().to_lowercase()
+                })
+            });
+        }
+
+        Ok(price_feeds_metadata)
+    }
+}

apps/hermes/src/state.rs

@@ -1,14 +1,17 @@
 //! This module contains the global state of the application.
 
 use {
-    self::cache::Cache,
-    crate::{
+    self::{
         aggregate::{
             AggregateState,
             AggregationEvent,
         },
-        api::types::PriceFeedMetadata,
+        benchmarks::BenchmarksState,
+        cache::CacheState,
+    },
+    crate::{
         network::wormhole::GuardianSet,
+        price_feeds_metadata::PriceFeedMetaState,
     },
     prometheus_client::registry::Registry,
     reqwest::Url,
@@ -25,13 +28,22 @@ use {
     },
 };
 
+pub mod aggregate;
 pub mod benchmarks;
 pub mod cache;
 
 pub struct State {
-    /// Storage is a short-lived cache of the state of all the updates that have been passed to the
-    /// store.
-    pub cache: Cache,
+    /// State for the `Cache` service for short-lived storage of updates.
+    pub cache: CacheState,
+
+    /// State for the `Benchmarks` service for looking up historical updates.
+    pub benchmarks: BenchmarksState,
+
+    /// 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.
@@ -40,20 +52,8 @@ 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>,
-
-    /// Benchmarks endpoint
-    pub benchmarks_endpoint: Option<Url>,
-
     /// Metrics registry
     pub metrics_registry: RwLock<Registry>,
-
-    /// Price feeds metadata
-    pub price_feeds_metadata: RwLock<Vec<PriceFeedMetadata>>,
 }
 
 impl State {
@@ -64,14 +64,13 @@ impl State {
     ) -> Arc<Self> {
         let mut metrics_registry = Registry::default();
         Arc::new(Self {
-            cache: Cache::new(cache_size),
+            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)),
-            benchmarks_endpoint,
             metrics_registry:  RwLock::new(metrics_registry),
-            price_feeds_metadata: RwLock::new(Default::default()),
         })
     }
 }

apps/hermes/src/state/aggregate.rs

@@ -20,10 +20,11 @@ use {
     },
     crate::{
         network::wormhole::VaaBytes,
+        price_feeds_metadata::PriceFeedMeta,
         state::{
             benchmarks::Benchmarks,
             cache::{
-                AggregateCache,
+                Cache,
                 MessageState,
                 MessageStateFilter,
             },
@@ -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,9 +198,48 @@ const READINESS_STALENESS_THRESHOLD: Duration = Duration::from_secs(30);
 /// 10 slots is almost 5 seconds.
 const READINESS_MAX_ALLOWED_SLOT_LAG: Slot = 10;
 
+#[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>;
+}
+
+/// 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
+    }
+}
+
+#[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()
+    }
+
     /// Stores the update data in the store
-#[tracing::instrument(skip(state, update))]
-pub async fn store_update(state: &State, update: Update) -> Result<()> {
+    #[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 {
@@ -192,14 +252,14 @@ pub async fn store_update(state: &State, update: Update) -> Result<()> {
                         tracing::info!(slot = proof.slot, "Storing VAA Merkle Proof.");
 
                         store_wormhole_merkle_verified_message(
-                        state,
+                            self,
                             proof.clone(),
                             update_vaa.to_owned(),
                         )
                         .await?;
 
-                    state
-                        .aggregate_state
+                        self.into()
+                            .data
                             .write()
                             .await
                             .metrics
@@ -213,12 +273,11 @@ pub async fn store_update(state: &State, update: Update) -> Result<()> {
                 let slot = accumulator_messages.slot;
                 tracing::info!(slot = slot, "Storing Accumulator Messages.");
 
-            state
-                .store_accumulator_messages(accumulator_messages)
+                self.store_accumulator_messages(accumulator_messages)
                     .await?;
 
-            state
-                .aggregate_state
+                self.into()
+                    .data
                     .write()
                     .await
                     .metrics
@@ -229,15 +288,15 @@ pub async fn store_update(state: &State, update: Update) -> Result<()> {
 
         // Update the aggregate state with the latest observed slot
         {
-        let mut aggregate_state = state.aggregate_state.write().await;
+            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));
         }
 
-    let accumulator_messages = state.fetch_accumulator_messages(slot).await?;
-    let wormhole_merkle_state = state.fetch_wormhole_merkle_state(slot).await?;
+        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) {
@@ -259,28 +318,32 @@ pub async fn store_update(state: &State, update: Update) -> Result<()> {
             .collect::<HashSet<_>>();
 
         tracing::info!(len = message_states.len(), "Storing Message States.");
-    state.store_message_states(message_states).await?;
+        self.store_message_states(message_states).await?;
 
         // Update the aggregate state
-    let mut aggregate_state = state.aggregate_state.write().await;
+        let mut aggregate_state = self.into().data.write().await;
 
-    // Check if the update is new or out of order
-    match aggregate_state.latest_completed_slot {
+        // 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);
-            state.api_update_tx.send(AggregationEvent::New { slot })?;
+                self.into()
+                    .api_update_tx
+                    .send(AggregationEvent::New { slot })
             }
             Some(latest) if slot > latest => {
-            state.prune_removed_keys(message_state_keys).await;
+                self.prune_removed_keys(message_state_keys).await;
                 aggregate_state.latest_completed_slot.replace(slot);
-            state.api_update_tx.send(AggregationEvent::New { slot })?;
-        }
-        _ => {
-            state
+                self.into()
                     .api_update_tx
-                .send(AggregationEvent::OutOfOrder { slot })?;
-        }
+                    .send(AggregationEvent::New { slot })
             }
+            _ => self
+                .into()
+                .api_update_tx
+                .send(AggregationEvent::OutOfOrder { slot }),
+        };
 
         aggregate_state.latest_completed_slot = aggregate_state
             .latest_completed_slot
@@ -298,6 +361,59 @@ pub async fn store_update(state: &State, update: Update) -> Result<()> {
         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)
+            }
+        }
+    }
+
+    async fn get_price_feed_ids(&self) -> HashSet<PriceIdentifier> {
+        Cache::message_state_keys(self)
+            .await
+            .iter()
+            .map(|key| PriceIdentifier::new(key.feed_id))
+            .collect()
+    }
+
+    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();
+
+        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
+    }
+}
+
 #[tracing::instrument(skip(accumulator_messages, wormhole_merkle_state))]
 fn build_message_states(
     accumulator_messages: AccumulatorMessages,
@@ -336,7 +452,7 @@ async fn get_verified_price_feeds<S>(
     request_time: RequestTime,
 ) -> Result<PriceFeedsWithUpdateData>
 where
-    S: AggregateCache,
+    S: Cache,
 {
     let messages = state
         .fetch_message_states(
@@ -390,71 +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: AggregateCache,
-    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 Benchmarks::get_verified_price_feeds(state, price_ids, publish_time).await;
-            }
-            Err(e)
-        }
-    }
-}
-
-pub async fn get_price_feed_ids<S>(state: &S) -> HashSet<PriceIdentifier>
-where
-    S: AggregateCache,
-{
-    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_feeds_metadata.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::store_price_feeds_metadata,
             state::test::setup_state,
         },
         futures::future::join_all,
@@ -468,10 +525,7 @@ mod test {
                 Accumulator,
             },
             hashers::keccak256_160::Keccak160,
-            messages::{
-                Message,
-                PriceFeedMessage,
-            },
+            messages::PriceFeedMessage,
             wire::v1::{
                 AccumulatorUpdateData,
                 Proof,
@@ -559,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()));
     }
@@ -585,14 +639,14 @@ 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,
+        let price_feeds_with_update_data = (&*state)
+            .get_price_feeds_with_update_data(
                 &[PriceIdentifier::new([100; 32])],
                 RequestTime::Latest,
             )
@@ -710,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])
@@ -720,8 +774,8 @@ mod test {
         );
 
         // Check that price feed 2 exists
-        assert!(get_price_feeds_with_update_data(
-            &*state,
+        assert!((&*state)
+            .get_price_feeds_with_update_data(
                 &[PriceIdentifier::new([200; 32])],
                 RequestTime::Latest,
             )
@@ -747,12 +801,12 @@ 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,
+        assert!((&*state)
+            .get_price_feeds_with_update_data(
                 &[PriceIdentifier::new([200; 32])],
                 RequestTime::Latest,
             )
@@ -793,8 +847,8 @@ 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,
+        let price_feeds_with_update_data = (&*state)
+            .get_price_feeds_with_update_data(
                 &[PriceIdentifier::new([100; 32])],
                 RequestTime::Latest,
             )
@@ -810,24 +864,22 @@ mod test {
 
 
         // Add a dummy price feeds metadata
-        store_price_feeds_metadata(
-            &state,
-            &[PriceFeedMetadata {
+        state
+            .store_price_feeds_metadata(&[PriceFeedMetadata {
                 id:         PriceIdentifier::new([100; 32]),
                 attributes: Default::default(),
-            }],
-        )
+            }])
             .await
             .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.
@@ -870,8 +922,8 @@ 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,
+            let price_feeds_with_update_data = (&*state)
+                .get_price_feeds_with_update_data(
                     &[
                         PriceIdentifier::new([100; 32]),
                         PriceIdentifier::new([200; 32]),
@@ -887,8 +939,8 @@ mod test {
 
         // Check nothing else is retained
         for slot in 0..900 {
-            assert!(get_price_feeds_with_update_data(
-                &*state,
+            assert!((&*state)
+                .get_price_feeds_with_update_data(
                     &[
                         PriceIdentifier::new([100; 32]),
                         PriceIdentifier::new([200; 32])

apps/hermes/src/state/aggregate/wormhole_merkle.rs

@@ -7,7 +7,7 @@ use {
     crate::{
         network::wormhole::VaaBytes,
         state::cache::{
-            AggregateCache,
+            Cache,
             MessageState,
         },
     },
@@ -70,14 +70,14 @@ impl From<MessageState> for RawMessageWithMerkleProof {
 }
 
 pub async fn store_wormhole_merkle_verified_message<S>(
-    store: &S,
+    state: &S,
     root: WormholeMerkleRoot,
     vaa: VaaBytes,
 ) -> Result<()>
 where
-    S: AggregateCache,
+    S: Cache,
 {
-    store
+    state
         .store_wormhole_merkle_state(WormholeMerkleState { root, vaa })
         .await?;
     Ok(())

apps/hermes/src/state/benchmarks.rs

@@ -1,19 +1,21 @@
 //! This module communicates with Pyth Benchmarks, an API for historical price feeds and their updates.
 
 use {
-    crate::{
+    super::{
         aggregate::{
             PriceFeedsWithUpdateData,
             UnixTimestamp,
         },
-        api::types::PriceUpdate,
+        State,
     },
+    crate::api::types::PriceUpdate,
     anyhow::Result,
     base64::{
         engine::general_purpose::STANDARD as base64_standard_engine,
         Engine as _,
     },
     pyth_sdk::PriceIdentifier,
+    reqwest::Url,
     serde::Deserialize,
 };
 
@@ -50,6 +52,23 @@ impl TryFrom<BinaryBlob> for Vec<Vec<u8>> {
     }
 }
 
+pub struct BenchmarksState {
+    endpoint: Option<Url>,
+}
+
+impl BenchmarksState {
+    pub fn new(url: Option<Url>) -> Self {
+        Self { endpoint: url }
+    }
+}
+
+/// Allow downcasting State into BenchmarksState for functions that depend on the `Benchmarks` service.
+impl<'a> From<&'a State> for &'a BenchmarksState {
+    fn from(state: &'a State) -> &'a BenchmarksState {
+        &state.benchmarks
+    }
+}
+
 #[async_trait::async_trait]
 pub trait Benchmarks {
     async fn get_verified_price_feeds(
@@ -60,21 +79,25 @@ pub trait Benchmarks {
 }
 
 #[async_trait::async_trait]
-impl Benchmarks for crate::state::State {
+impl<T> Benchmarks for T
+where
+    for<'a> &'a T: Into<&'a BenchmarksState>,
+    T: Sync,
+{
     async fn get_verified_price_feeds(
         &self,
         price_ids: &[PriceIdentifier],
         publish_time: UnixTimestamp,
     ) -> Result<PriceFeedsWithUpdateData> {
         let endpoint = self
-            .benchmarks_endpoint
+            .into()
+            .endpoint
             .as_ref()
             .ok_or_else(|| anyhow::anyhow!("Benchmarks endpoint is not set"))?
             .join(&format!("/v1/updates/price/{}", publish_time))
             .unwrap();
 
-        let client = reqwest::Client::new();
-        let mut request = client
+        let mut request = reqwest::Client::new()
             .get(endpoint)
             .timeout(BENCHMARKS_REQUEST_TIMEOUT)
             .query(&[("encoding", "hex")])

apps/hermes/src/state/cache.rs

@@ -1,5 +1,6 @@
 use {
-    crate::aggregate::{
+    super::State,
+    crate::state::aggregate::{
         wormhole_merkle::WormholeMerkleState,
         AccumulatorMessages,
         ProofSet,
@@ -96,23 +97,186 @@ pub enum MessageStateFilter {
     Only(MessageType),
 }
 
-pub struct Cache {
-    /// Accumulator messages cache
-    ///
-    /// We do not write to this cache much, so we can use a simple RwLock instead of a DashMap.
-    accumulator_messages_cache: Arc<RwLock<BTreeMap<Slot, AccumulatorMessages>>>,
+/// A Cache of AccumulatorMessage by slot. We do not write to this cache much, so we can use a simple RwLock instead of a DashMap.
+type AccumulatorMessagesCache = Arc<RwLock<BTreeMap<Slot, AccumulatorMessages>>>;
 
-    /// Wormhole merkle state cache
-    ///
-    /// We do not write to this cache much, so we can use a simple RwLock instead of a DashMap.
-    wormhole_merkle_state_cache: Arc<RwLock<BTreeMap<Slot, WormholeMerkleState>>>,
+/// A Cache of WormholeMerkleState by slot. We do not write to this cache much, so we can use a simple RwLock instead of a DashMap.
+type WormholeMerkleStateCache = Arc<RwLock<BTreeMap<Slot, WormholeMerkleState>>>;
 
-    message_cache: Arc<RwLock<HashMap<MessageStateKey, BTreeMap<MessageStateTime, MessageState>>>>,
+/// A Cache of `Time<->MessageState` by feed id.
+type MessageCache = Arc<RwLock<HashMap<MessageStateKey, BTreeMap<MessageStateTime, MessageState>>>>;
+
+/// A collection of caches for various program state.
+pub struct CacheState {
+    accumulator_messages_cache:  AccumulatorMessagesCache,
+    wormhole_merkle_state_cache: WormholeMerkleStateCache,
+    message_cache:               MessageCache,
     cache_size:                  u64,
 }
 
+impl CacheState {
+    pub fn new(size: u64) -> Self {
+        Self {
+            accumulator_messages_cache:  Arc::new(RwLock::new(BTreeMap::new())),
+            wormhole_merkle_state_cache: Arc::new(RwLock::new(BTreeMap::new())),
+            message_cache:               Arc::new(RwLock::new(HashMap::new())),
+            cache_size:                  size,
+        }
+    }
+}
+
+/// Allow downcasting State into CacheState for functions that depend on the `Cache` service.
+impl<'a> From<&'a State> for &'a CacheState {
+    fn from(state: &'a State) -> &'a CacheState {
+        &state.cache
+    }
+}
+
+#[async_trait::async_trait]
+pub trait Cache {
+    async fn store_message_states(&self, message_states: Vec<MessageState>) -> Result<()>;
+    async fn prune_removed_keys(&self, current_keys: HashSet<MessageStateKey>);
+    async fn store_accumulator_messages(
+        &self,
+        accumulator_messages: AccumulatorMessages,
+    ) -> Result<()>;
+    async fn fetch_accumulator_messages(&self, slot: Slot) -> Result<Option<AccumulatorMessages>>;
+    async fn store_wormhole_merkle_state(
+        &self,
+        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>,
+    T: Sync,
+{
+    async fn message_state_keys(&self) -> Vec<MessageStateKey> {
+        self.into()
+            .message_cache
+            .read()
+            .await
+            .iter()
+            .map(|entry| entry.0.clone())
+            .collect::<Vec<_>>()
+    }
+
+    async fn store_message_states(&self, message_states: Vec<MessageState>) -> Result<()> {
+        let mut message_cache = self.into().message_cache.write().await;
+
+        for message_state in message_states {
+            let key = message_state.key();
+            let time = message_state.time();
+            let cache = message_cache.entry(key).or_insert_with(BTreeMap::new);
+            cache.insert(time, message_state);
+
+            // Remove the earliest message states if the cache size is exceeded
+            while cache.len() > self.into().cache_size as usize {
+                cache.pop_first();
+            }
+        }
+
+        Ok(())
+    }
+
+    /// This method takes the current feed ids and prunes the cache for the keys
+    /// that are not present in the current feed ids.
+    ///
+    /// There is a side-effect of this: if a key gets removed, we will
+    /// lose the cache for that key and cannot retrieve it for historical
+    /// price queries.
+    async fn prune_removed_keys(&self, current_keys: HashSet<MessageStateKey>) {
+        let mut message_cache = self.into().message_cache.write().await;
+
+        // Sometimes, some keys are removed from the accumulator. We track which keys are not
+        // present in the message states and remove them from the cache.
+        let keys_in_cache = message_cache
+            .iter()
+            .map(|(key, _)| key.clone())
+            .collect::<HashSet<_>>();
+
+        for key in keys_in_cache {
+            if !current_keys.contains(&key) {
+                tracing::info!("Feed {:?} seems to be removed. Removing it from cache", key);
+                message_cache.remove(&key);
+            }
+        }
+    }
+
+    async fn fetch_message_states(
+        &self,
+        ids: Vec<FeedId>,
+        request_time: RequestTime,
+        filter: MessageStateFilter,
+    ) -> Result<Vec<MessageState>> {
+        join_all(ids.into_iter().flat_map(|id| {
+            let request_time = request_time.clone();
+            let message_types: Vec<MessageType> = match filter {
+                MessageStateFilter::All => MessageType::iter().collect(),
+                MessageStateFilter::Only(t) => vec![t],
+            };
+
+            message_types.into_iter().map(move |message_type| {
+                let key = MessageStateKey {
+                    feed_id: id,
+                    type_:   message_type,
+                };
+                retrieve_message_state(self.into(), key, request_time.clone())
+            })
+        }))
+        .await
+        .into_iter()
+        .collect::<Option<Vec<_>>>()
+        .ok_or(anyhow!("Message not found"))
+    }
+
+    async fn store_accumulator_messages(
+        &self,
+        accumulator_messages: AccumulatorMessages,
+    ) -> Result<()> {
+        let mut cache = self.into().accumulator_messages_cache.write().await;
+        cache.insert(accumulator_messages.slot, accumulator_messages);
+        while cache.len() > self.into().cache_size as usize {
+            cache.pop_first();
+        }
+        Ok(())
+    }
+
+    async fn fetch_accumulator_messages(&self, slot: Slot) -> Result<Option<AccumulatorMessages>> {
+        let cache = self.into().accumulator_messages_cache.read().await;
+        Ok(cache.get(&slot).cloned())
+    }
+
+    async fn store_wormhole_merkle_state(
+        &self,
+        wormhole_merkle_state: WormholeMerkleState,
+    ) -> Result<()> {
+        let mut cache = self.into().wormhole_merkle_state_cache.write().await;
+        cache.insert(wormhole_merkle_state.root.slot, wormhole_merkle_state);
+        while cache.len() > self.into().cache_size as usize {
+            cache.pop_first();
+        }
+        Ok(())
+    }
+
+    async fn fetch_wormhole_merkle_state(&self, slot: Slot) -> Result<Option<WormholeMerkleState>> {
+        let cache = self.into().wormhole_merkle_state_cache.read().await;
+        Ok(cache.get(&slot).cloned())
+    }
+}
+
 async fn retrieve_message_state(
-    cache: &Cache,
+    cache: &CacheState,
     key: MessageStateKey,
     request_time: RequestTime,
 ) -> Option<MessageState> {
@@ -156,179 +320,19 @@ async fn retrieve_message_state(
     }
 }
 
-impl Cache {
-    pub fn new(cache_size: u64) -> Self {
-        Self {
-            message_cache: Arc::new(RwLock::new(HashMap::new())),
-            accumulator_messages_cache: Arc::new(RwLock::new(BTreeMap::new())),
-            wormhole_merkle_state_cache: Arc::new(RwLock::new(BTreeMap::new())),
-            cache_size,
-        }
-    }
-}
-
-#[async_trait::async_trait]
-pub trait AggregateCache {
-    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,
-    ) -> Result<()>;
-    async fn fetch_accumulator_messages(&self, slot: Slot) -> Result<Option<AccumulatorMessages>>;
-    async fn store_wormhole_merkle_state(
-        &self,
-        wormhole_merkle_state: WormholeMerkleState,
-    ) -> Result<()>;
-    async fn fetch_wormhole_merkle_state(&self, slot: Slot) -> Result<Option<WormholeMerkleState>>;
-}
-
-#[async_trait::async_trait]
-impl AggregateCache for crate::state::State {
-    async fn message_state_keys(&self) -> Vec<MessageStateKey> {
-        self.cache
-            .message_cache
-            .read()
-            .await
-            .iter()
-            .map(|entry| entry.0.clone())
-            .collect::<Vec<_>>()
-    }
-
-    async fn store_message_states(&self, message_states: Vec<MessageState>) -> Result<()> {
-        let mut message_cache = self.cache.message_cache.write().await;
-
-        for message_state in message_states {
-            let key = message_state.key();
-            let time = message_state.time();
-            let cache = message_cache.entry(key).or_insert_with(BTreeMap::new);
-            cache.insert(time, message_state);
-
-            // Remove the earliest message states if the cache size is exceeded
-            while cache.len() > self.cache.cache_size as usize {
-                cache.pop_first();
-            }
-        }
-
-        Ok(())
-    }
-
-    /// This method takes the current feed ids and prunes the cache for the keys
-    /// that are not present in the current feed ids.
-    ///
-    /// There is a side-effect of this: if a key gets removed, we will
-    /// lose the cache for that key and cannot retrieve it for historical
-    /// price queries.
-    async fn prune_removed_keys(&self, current_keys: HashSet<MessageStateKey>) {
-        let mut message_cache = self.cache.message_cache.write().await;
-
-        // Sometimes, some keys are removed from the accumulator. We track which keys are not
-        // present in the message states and remove them from the cache.
-        let keys_in_cache = message_cache
-            .iter()
-            .map(|(key, _)| key.clone())
-            .collect::<HashSet<_>>();
-
-        for key in keys_in_cache {
-            if !current_keys.contains(&key) {
-                tracing::info!("Feed {:?} seems to be removed. Removing it from cache", key);
-                message_cache.remove(&key);
-            }
-        }
-    }
-
-    async fn fetch_message_states(
-        &self,
-        ids: Vec<FeedId>,
-        request_time: RequestTime,
-        filter: MessageStateFilter,
-    ) -> Result<Vec<MessageState>> {
-        join_all(ids.into_iter().flat_map(|id| {
-            let request_time = request_time.clone();
-            let message_types: Vec<MessageType> = match filter {
-                MessageStateFilter::All => MessageType::iter().collect(),
-                MessageStateFilter::Only(t) => vec![t],
-            };
-
-            message_types.into_iter().map(move |message_type| {
-                let key = MessageStateKey {
-                    feed_id: id,
-                    type_:   message_type,
-                };
-                retrieve_message_state(&self.cache, key, request_time.clone())
-            })
-        }))
-        .await
-        .into_iter()
-        .collect::<Option<Vec<_>>>()
-        .ok_or(anyhow!("Message not found"))
-    }
-
-    async fn store_accumulator_messages(
-        &self,
-        accumulator_messages: AccumulatorMessages,
-    ) -> Result<()> {
-        let mut cache = self.cache.accumulator_messages_cache.write().await;
-        cache.insert(accumulator_messages.slot, accumulator_messages);
-        while cache.len() > self.cache.cache_size as usize {
-            cache.pop_first();
-        }
-        Ok(())
-    }
-
-    async fn fetch_accumulator_messages(&self, slot: Slot) -> Result<Option<AccumulatorMessages>> {
-        let cache = self.cache.accumulator_messages_cache.read().await;
-        Ok(cache.get(&slot).cloned())
-    }
-
-    async fn store_wormhole_merkle_state(
-        &self,
-        wormhole_merkle_state: WormholeMerkleState,
-    ) -> Result<()> {
-        let mut cache = self.cache.wormhole_merkle_state_cache.write().await;
-        cache.insert(wormhole_merkle_state.root.slot, wormhole_merkle_state);
-        while cache.len() > self.cache.cache_size as usize {
-            cache.pop_first();
-        }
-        Ok(())
-    }
-
-    async fn fetch_wormhole_merkle_state(&self, slot: Slot) -> Result<Option<WormholeMerkleState>> {
-        let cache = self.cache.wormhole_merkle_state_cache.read().await;
-        Ok(cache.get(&slot).cloned())
-    }
-}
-
 #[cfg(test)]
 mod test {
     use {
         super::*,
-        crate::{
-            aggregate::{
-                wormhole_merkle::{
-                    WormholeMerkleMessageProof,
-                    WormholeMerkleState,
-                },
-                AccumulatorMessages,
-                ProofSet,
-            },
-            state::test::setup_state,
+        crate::state::{
+            aggregate::wormhole_merkle::WormholeMerkleMessageProof,
+            test::setup_state,
         },
         pyth_sdk::UnixTimestamp,
         pythnet_sdk::{
             accumulators::merkle::MerklePath,
             hashers::keccak256_160::Keccak160,
-            messages::{
-                Message,
-                PriceFeedMessage,
-            },
+            messages::PriceFeedMessage,
             wire::v1::WormholeMerkleRoot,
         },
     };
@@ -369,7 +373,7 @@ mod test {
         slot: Slot,
     ) -> MessageState
     where
-        S: AggregateCache,
+        S: Cache,
     {
         let message_state = create_dummy_price_feed_message_state(feed_id, publish_time, slot);
         state