fix: use server_id when unsubscribing

crates/provider/src/provider/root.rs

@@ -87,14 +87,14 @@ impl<T: Transport + Clone, N: Network> RootProvider<T, N> {
     #[cfg(feature = "pubsub")]
     pub async fn get_subscription<R: alloy_json_rpc::RpcReturn>(
         &self,
-        id: alloy_primitives::U256,
+        id: alloy_primitives::B256,
     ) -> alloy_transport::TransportResult<Subscription<R>> {
         self.pubsub_frontend()?.get_subscription(id).await.map(Subscription::from)
     }
 
     /// Unsubscribes from the subscription corresponding to the given RPC subscription ID.
     #[cfg(feature = "pubsub")]
-    pub fn unsubscribe(&self, id: alloy_primitives::U256) -> alloy_transport::TransportResult<()> {
+    pub fn unsubscribe(&self, id: alloy_primitives::B256) -> alloy_transport::TransportResult<()> {
         self.pubsub_frontend()?.unsubscribe(id)
     }
 

crates/provider/src/provider/trait.rs

@@ -860,7 +860,7 @@ pub trait Provider<T: Transport + Clone = BoxTransport, N: Network = Ethereum>:
 
     /// Cancels a subscription given the subscription ID.
     #[cfg(feature = "pubsub")]
-    async fn unsubscribe(&self, id: U256) -> TransportResult<()> {
+    async fn unsubscribe(&self, id: B256) -> TransportResult<()> {
         self.root().unsubscribe(id)
     }
 

crates/pubsub/src/frontend.rs

@@ -1,6 +1,6 @@
 use crate::{ix::PubSubInstruction, managers::InFlight, RawSubscription};
 use alloy_json_rpc::{RequestPacket, Response, ResponsePacket, SerializedRequest};
-use alloy_primitives::U256;
+use alloy_primitives::B256;
 use alloy_transport::{TransportError, TransportErrorKind, TransportFut, TransportResult};
 use futures::{future::try_join_all, FutureExt, TryFutureExt};
 use std::{
@@ -38,7 +38,7 @@ impl PubSubFrontend {
     /// Get the subscription ID for a local ID.
     pub fn get_subscription(
         &self,
-        id: U256,
+        id: B256,
     ) -> impl Future<Output = TransportResult<RawSubscription>> + Send + 'static {
         let backend_tx = self.tx.clone();
         async move {
@@ -51,7 +51,7 @@ impl PubSubFrontend {
     }
 
     /// Unsubscribe from a subscription.
-    pub fn unsubscribe(&self, id: U256) -> TransportResult<()> {
+    pub fn unsubscribe(&self, id: B256) -> TransportResult<()> {
         self.tx
             .send(PubSubInstruction::Unsubscribe(id))
             .map_err(|_| TransportErrorKind::backend_gone())

crates/pubsub/src/ix.rs

@@ -1,5 +1,5 @@
 use crate::{managers::InFlight, RawSubscription};
-use alloy_primitives::U256;
+use alloy_primitives::B256;
 use std::fmt;
 use tokio::sync::oneshot;
 
@@ -8,9 +8,9 @@ pub(crate) enum PubSubInstruction {
     /// Send a request.
     Request(InFlight),
     /// Get the subscription ID for a local ID.
-    GetSub(U256, oneshot::Sender<RawSubscription>),
+    GetSub(B256, oneshot::Sender<RawSubscription>),
     /// Unsubscribe from a subscription.
-    Unsubscribe(U256),
+    Unsubscribe(B256),
 }
 
 impl fmt::Debug for PubSubInstruction {

crates/pubsub/src/managers/sub.rs

@@ -63,6 +63,11 @@ impl SubscriptionManager {
         self.local_to_server.get_by_right(server_id).copied()
     }
 
+    /// De-alias an alias, getting the original ID.
+    pub(crate) fn server_id_for(&self, local_id: &B256) -> Option<&SubId> {
+        self.local_to_server.get_by_left(local_id)
+    }
+
     /// Drop all server_ids.
     pub(crate) fn drop_server_ids(&mut self) {
         self.local_to_server.clear();

crates/pubsub/src/service.rs

@@ -5,7 +5,7 @@ use crate::{
     PubSubConnect, PubSubFrontend, RawSubscription,
 };
 use alloy_json_rpc::{Id, PubSubItem, Request, Response, ResponsePayload, SubId};
-use alloy_primitives::U256;
+use alloy_primitives::B256;
 use alloy_transport::{
     utils::{to_json_raw_value, Spawnable},
     TransportErrorKind, TransportResult,
@@ -123,19 +123,21 @@ impl<T: PubSubConnect> PubSubService<T> {
     /// the subscription does not exist, the waiter is sent nothing, and the
     /// `tx` is dropped. This notifies the waiter that the subscription does
     /// not exist.
-    fn service_get_sub(&mut self, local_id: U256, tx: oneshot::Sender<RawSubscription>) {
-        if let Some(rx) = self.subs.get_subscription(local_id.into()) {
+    fn service_get_sub(&mut self, local_id: B256, tx: oneshot::Sender<RawSubscription>) {
+        if let Some(rx) = self.subs.get_subscription(local_id) {
             let _ = tx.send(rx);
         }
     }
 
     /// Service an unsubscribe instruction.
-    fn service_unsubscribe(&mut self, local_id: U256) -> TransportResult<()> {
-        let req = Request::new("eth_unsubscribe", Id::None, [local_id]);
-        let brv = req.serialize().expect("no ser error").take_request();
+    fn service_unsubscribe(&mut self, local_id: B256) -> TransportResult<()> {
+        if let Some(server_id) = self.subs.server_id_for(&local_id) {
+            let req = Request::new("eth_unsubscribe", Id::None, [server_id]);
+            let brv = req.serialize().expect("no ser error").take_request();
 
-        self.dispatch_request(brv)?;
-        self.subs.remove_sub(local_id.into());
+            self.dispatch_request(brv)?;
+        }
+        self.subs.remove_sub(local_id);
         Ok(())
     }
 

crates/rpc-client/src/client.rs

@@ -306,14 +306,14 @@ mod pubsub_impl {
 
     impl RpcClientInner<PubSubFrontend> {
         /// Get a [`RawSubscription`] for the given subscription ID.
-        pub async fn get_raw_subscription(&self, id: alloy_primitives::U256) -> RawSubscription {
+        pub async fn get_raw_subscription(&self, id: alloy_primitives::B256) -> RawSubscription {
             self.transport.get_subscription(id).await.unwrap()
         }
 
         /// Get a [`Subscription`] for the given subscription ID.
         pub async fn get_subscription<T: serde::de::DeserializeOwned>(
             &self,
-            id: alloy_primitives::U256,
+            id: alloy_primitives::B256,
         ) -> Subscription<T> {
             Subscription::from(self.get_raw_subscription(id).await)
         }