chore(voyager): cleanup

lib/chain-utils/src/keyring.rs

@@ -4,7 +4,7 @@ use crossbeam_queue::ArrayQueue;
 use futures::Future;
 use rand::prelude::SliceRandom;
 use serde::{Deserialize, Serialize};
-use tracing::{info_span, warn, Instrument};
+use tracing::{debug, info_span, Instrument};
 
 pub trait ChainKeyring {
     type Address: Hash + Eq + Clone + Display + Send + Sync;
@@ -88,7 +88,7 @@ impl<A: Hash + Eq + Clone + Display, S: 'static> ConcurrentKeyring<A, S> {
         f: F,
     ) -> Option<Fut::Output> {
         let Some(address) = self.addresses_buffer.pop() else {
-            warn!(keyring = %self.name, "high traffic in keyring");
+            debug!(keyring = %self.name, "high traffic in keyring");
             return None;
         };
 

lib/pg-queue/src/lib.rs

@@ -323,7 +323,7 @@ impl<T: QueueMessage> voyager_vm::Queue<T> for PgQueue<T> {
 
         match row {
             Some(row) => {
-                let span = info_span!("processing item", id = row.id);
+                let span = info_span!("processing item", item_id = row.id);
 
                 trace!(%row.item);
 
@@ -384,20 +384,22 @@ impl<T: QueueMessage> voyager_vm::Queue<T> for PgQueue<T> {
 
                             sqlx::query(
                                 "
-                                INSERT INTO queue (item)
-                                SELECT * FROM UNNEST($1::JSONB[])
+                                INSERT INTO queue (item, parents)
+                                SELECT *, $1 as parents FROM UNNEST($2::JSONB[])
                                 ",
                             )
+                            .bind(vec![row.id])
                             .bind(ready.into_iter().map(Json).collect::<Vec<_>>())
                             .execute(tx.as_mut())
                             .await?;
 
                             sqlx::query(
                                 "
-                                INSERT INTO optimize (item, tag)
-                                SELECT * FROM UNNEST($1::JSONB[], $2::TEXT[])
+                                INSERT INTO optimize (item, tag, parents)
+                                SELECT *, $1 as parents FROM UNNEST($2::JSONB[], $3::TEXT[])
                                 ",
                             )
+                            .bind(vec![row.id])
                             .bind(optimize.iter().map(|(op, _)| Json(op)).collect::<Vec<_>>())
                             .bind(optimize.iter().map(|(_, tag)| *tag).collect::<Vec<_>>())
                             .execute(tx.as_mut())

lib/reconnecting-jsonrpc-ws-client/src/lib.rs

@@ -181,7 +181,7 @@ impl ClientT for Client {
             .as_deref()
             .ok_or_else(|| {
                 jsonrpsee::core::client::Error::Custom(format!(
-                    "not yet connected (request: {method})",
+                    "not yet connected (notification: {method})",
                 ))
             })?
             .notification(method, params)

lib/state-lens-ics23-mpt-light-client-types/src/lib.rs

@@ -1,5 +1,4 @@
 pub mod client_state;
 pub mod consensus_state;
-pub mod header;
 
-pub use crate::{client_state::ClientState, consensus_state::ConsensusState, header::Header};
+pub use crate::{client_state::ClientState, consensus_state::ConsensusState};

lib/state-lens-light-client-types/src/header.rs

@@ -14,18 +14,11 @@ pub struct Header {
 #[cfg(feature = "ethabi")]
 pub mod ethabi {
     use alloy::sol_types::SolValue;
-    use unionlabs::encoding::{Encode, EthAbi};
+    use unionlabs::{ibc::core::client::height::Height, impl_ethabi_via_try_from_into};
 
     use crate::Header;
 
-    impl Encode<EthAbi> for Header {
-        fn encode(self) -> Vec<u8> {
-            Into::<SolHeader>::into(self).abi_encode_params()
-        }
-    }
-
-    #[derive(Debug, Clone, PartialEq, thiserror::Error)]
-    pub enum Error {}
+    impl_ethabi_via_try_from_into!(Header => SolHeader);
 
     alloy::sol! {
         struct SolHeader {
@@ -46,4 +39,15 @@ pub mod ethabi {
             }
         }
     }
+
+    impl From<SolHeader> for Header {
+        fn from(value: SolHeader) -> Self {
+            Self {
+                l1_height: Height::new(value.l1Height),
+                l2_height: Height::new(value.l2Height),
+                l2_consensus_state_proof: value.l2InclusionProof.into(),
+                l2_consensus_state: value.l2ConsensusState.into(),
+            }
+        }
+    }
 }

lib/voyager-message/src/context.rs

@@ -562,39 +562,39 @@ impl Context {
         )
         .await?;
 
-        main_rpc_server.start(Arc::new(modules));
-
         info!("checking for plugin health...");
 
-        {
-            let mut futures = plugins
-                .iter()
-                .map(|(name, client)| async move {
-                    match client
-                        .client
-                        .wait_until_connected(Duration::from_secs(10))
-                        .instrument(debug_span!("health check", %name))
-                        .await
-                    {
-                        Ok(_) => {
-                            info!("plugin {name} connected")
-                        }
-                        Err(_) => {
-                            warn!("plugin {name} failed to connect after 10 seconds")
-                        }
+        let futures = plugins
+            .iter()
+            .map(|(name, client)| async move {
+                match client
+                    .client
+                    .wait_until_connected(Duration::from_secs(10))
+                    .instrument(debug_span!("health check", %name))
+                    .await
+                {
+                    Ok(()) => {
+                        info!("plugin {name} connected")
                     }
-                })
-                .collect::<FuturesUnordered<_>>();
-
-            match cancellation_token
-                .run_until_cancelled(async { while let Some(()) = futures.next().await {} })
-                .await
-            {
-                Some(()) => {}
-                None => return Err(anyhow!("startup error")),
-            }
+                    Err(_) => {
+                        warn!("plugin {name} failed to connect after 10 seconds")
+                    }
+                }
+            })
+            .collect::<FuturesUnordered<_>>();
+
+        match cancellation_token
+            .run_until_cancelled(futures.collect::<Vec<_>>())
+            .await
+        {
+            Some(_) => {}
+            None => return Err(anyhow!("startup error")),
         }
 
+        main_rpc_server.start(Arc::new(modules));
+
+        info!("started");
+
         Ok(Self {
             rpc_server: main_rpc_server,
             plugins,

lib/voyager-message/src/module.rs

@@ -13,6 +13,7 @@ use crate::{
         ChainId, ClientInfo, ClientStateMeta, ClientType, ConsensusStateMeta, IbcInterface, IbcSpec,
     },
     data::Data,
+    rpc::ProofType,
     RawClientId, VoyagerMessage,
 };
 
@@ -350,14 +351,18 @@ pub trait ProofModule<V: IbcSpec> {
     /// Query a proof of IBC state on this chain, at the specified [`Height`],
     /// returning the state as a JSON [`Value`].
     #[method(name = "queryIbcProof", with_extensions)]
-    async fn query_ibc_proof(&self, at: Height, path: V::StorePath) -> RpcResult<Value>;
+    async fn query_ibc_proof(
+        &self,
+        at: Height,
+        path: V::StorePath,
+    ) -> RpcResult<(Value, ProofType)>;
 }
 
 /// Type-erased version of [`ProofModuleClient`].
 #[rpc(client, namespace = "proof")]
 pub trait RawProofModule {
     #[method(name = "queryIbcProof")]
-    async fn query_ibc_proof_raw(&self, at: Height, path: Value) -> RpcResult<Value>;
+    async fn query_ibc_proof_raw(&self, at: Height, path: Value) -> RpcResult<(Value, ProofType)>;
 }
 
 /// Client modules provide functionality to interact with a single light client

lib/voyager-message/src/rpc.rs

@@ -5,7 +5,7 @@ use jsonrpsee::{
     types::{ErrorObject, ErrorObjectOwned},
 };
 use macros::model;
-use serde::de::DeserializeOwned;
+use serde::{de::DeserializeOwned, Deserialize, Serialize};
 use serde_json::{json, Value};
 use unionlabs::{ibc::core::client::height::Height, primitives::Bytes, ErrorReporter};
 use voyager_core::{IbcSpecId, Timestamp};
@@ -167,16 +167,18 @@ impl IbcState<Value> {
 
 #[model]
 pub struct IbcProof {
-    // pub proof_type: ProofType,
+    pub proof_type: ProofType,
     /// The height that the proof was read at.
     pub height: Height,
     pub proof: Value,
 }
 
-// enum ProofType {
-//     Membership,
-//     NonMembership,
-// }
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ProofType {
+    Membership,
+    NonMembership,
+}
 
 #[model]
 pub struct SelfClientState {