fix(voyager): thread item ids back to the voyager rpc server properly

unionlabs · Jan 19, 2025 · e5e26fe · e5e26fe
1 parent 1e801ac
commit e5e26fe
Show file tree

Hide file tree

Showing 4 changed files with 158 additions and 66 deletions.
diff --git a/lib/voyager-message/src/context.rs b/lib/voyager-message/src/context.rs
@@ -144,6 +144,8 @@ where
     for<'a> &'a Self: ClientT,
 {
     fn with_id(&self, item_id: Option<ItemId>) -> IdThreadClient<&Self> {
+        trace!(?item_id, "threading id");
+
         IdThreadClient {
             client: self,
             item_id,

diff --git a/lib/voyager-message/src/lib.rs b/lib/voyager-message/src/lib.rs
@@ -552,6 +552,8 @@ impl<Inner: ClientT + Send + Sync> ClientT for IdThreadClient<Inner> {
         R: DeserializeOwned,
         Params: ToRpcParams + Send,
     {
+        trace!(item_id = ?self.item_id);
+
         match self.item_id {
             Some(item_id) => {
                 self.client
@@ -919,6 +921,8 @@ impl<'a, S: RpcServiceT<'a> + Send + Sync> RpcServiceT<'a> for InjectClient<S> {
                         ..request
                     };
 
+                    trace!("request is for item_id {}", item_id.raw());
+
                     request.extensions.insert(item_id);
 
                     request.extensions.insert(VoyagerClient(IdThreadClient {
@@ -932,12 +936,19 @@ impl<'a, S: RpcServiceT<'a> + Send + Sync> RpcServiceT<'a> for InjectClient<S> {
                         .instrument(info_span!("item_id", item_id = item_id.raw()))
                         .left_future();
                 }
-                Err(_) => {
+                Err(err) => {
+                    trace!(
+                        "unable to parse item_id from request: {}",
+                        ErrorReporter(err)
+                    );
+
                     request.params = Some(params);
                 }
             }
         };
 
+        trace!("request is not for a queue item");
+
         request.extensions.insert(VoyagerClient(IdThreadClient {
             client: self.client.clone(),
             item_id: None,

diff --git a/lib/voyager-message/src/rpc.rs b/lib/voyager-message/src/rpc.rs
@@ -27,17 +27,17 @@ pub mod server;
 )]
 // TODO: Ensure that height is always the last parameter for consistency
 pub trait VoyagerRpc {
-    #[method(name = "info")]
+    #[method(name = "info", with_extensions)]
     async fn info(&self) -> RpcResult<LoadedModulesInfo>;
 
     // =========
     // consensus
     // =========
 
-    #[method(name = "queryLatestHeight")]
+    #[method(name = "queryLatestHeight", with_extensions)]
     async fn query_latest_height(&self, chain_id: ChainId, finalized: bool) -> RpcResult<Height>;
 
-    #[method(name = "queryLatestTimestamp")]
+    #[method(name = "queryLatestTimestamp", with_extensions)]
     async fn query_latest_timestamp(
         &self,
         chain_id: ChainId,
@@ -48,15 +48,15 @@ pub trait VoyagerRpc {
     // IBC state queries
     // =================
 
-    #[method(name = "clientInfo")]
+    #[method(name = "clientInfo", with_extensions)]
     async fn client_info(
         &self,
         chain_id: ChainId,
         ibc_spec_id: IbcSpecId,
         client_id: RawClientId,
     ) -> RpcResult<ClientInfo>;
 
-    #[method(name = "clientMeta")]
+    #[method(name = "clientMeta", with_extensions)]
     async fn client_meta(
         &self,
         chain_id: ChainId,
@@ -65,7 +65,7 @@ pub trait VoyagerRpc {
         client_id: RawClientId,
     ) -> RpcResult<ClientStateMeta>;
 
-    #[method(name = "queryIbcState")]
+    #[method(name = "queryIbcState", with_extensions)]
     async fn query_ibc_state(
         &self,
         chain_id: ChainId,
@@ -74,7 +74,7 @@ pub trait VoyagerRpc {
         path: Value,
     ) -> RpcResult<IbcState<Value>>;
 
-    #[method(name = "queryIbcProof")]
+    #[method(name = "queryIbcProof", with_extensions)]
     async fn query_ibc_proof(
         &self,
         chain_id: ChainId,
@@ -87,15 +87,15 @@ pub trait VoyagerRpc {
     // self state queries, for creating clients
     // ========================================
 
-    #[method(name = "selfClientState")]
+    #[method(name = "selfClientState", with_extensions)]
     async fn self_client_state(
         &self,
         chain_id: ChainId,
         client_type: ClientType,
         height: QueryHeight,
     ) -> RpcResult<SelfClientState>;
 
-    #[method(name = "selfConsensusState")]
+    #[method(name = "selfConsensusState", with_extensions)]
     async fn self_consensus_state(
         &self,
         chain_id: ChainId,
@@ -107,7 +107,7 @@ pub trait VoyagerRpc {
     // state and proof codecs
     // ======================
 
-    #[method(name = "encodeProof")]
+    #[method(name = "encodeProof", with_extensions)]
     async fn encode_proof(
         &self,
         client_type: ClientType,
@@ -116,7 +116,7 @@ pub trait VoyagerRpc {
         proof: Value,
     ) -> RpcResult<Bytes>;
 
-    #[method(name = "decodeClientStateMeta")]
+    #[method(name = "decodeClientStateMeta", with_extensions)]
     async fn decode_client_state_meta(
         &self,
         client_type: ClientType,
@@ -125,7 +125,7 @@ pub trait VoyagerRpc {
         client_state: Bytes,
     ) -> RpcResult<ClientStateMeta>;
 
-    #[method(name = "decodeClientState")]
+    #[method(name = "decodeClientState", with_extensions)]
     async fn decode_client_state(
         &self,
         client_type: ClientType,
@@ -134,7 +134,7 @@ pub trait VoyagerRpc {
         client_state: Bytes,
     ) -> RpcResult<Value>;
 
-    #[method(name = "decodeConsensusState")]
+    #[method(name = "decodeConsensusState", with_extensions)]
     async fn decode_consensus_state(
         &self,
         client_type: ClientType,