rpc module: report error on invalid subscription

test-utils/src/helpers.rs

@@ -27,6 +27,7 @@
 use crate::mocks::{Body, HttpResponse, Id, Uri};
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{Request, Response, Server};
+use serde::Serialize;
 use serde_json::Value;
 use std::convert::Infallible;
 use std::net::SocketAddr;
@@ -95,6 +96,15 @@ pub fn invalid_params(id: Id) -> String {
 	)
 }
 
+pub fn call<T: Serialize>(method: &str, params: Vec<T>, id: Id) -> String {
+	format!(
+		r#"{{"jsonrpc":"2.0","method":{},"params":{},"id":{}}}"#,
+		serde_json::to_string(method).unwrap(),
+		serde_json::to_string(&params).unwrap(),
+		serde_json::to_string(&id).unwrap()
+	)
+}
+
 pub fn call_execution_failed(msg: &str, id: Id) -> String {
 	format!(
 		r#"{{"jsonrpc":"2.0","error":{{"code":-32000,"message":"{}"}},"id":{}}}"#,

types/src/v2/error.rs

@@ -73,7 +73,7 @@ impl<'a> From<ErrorCode> for ErrorObject<'a> {
 impl<'a> PartialEq for ErrorObject<'a> {
 	fn eq(&self, other: &Self) -> bool {
 		let this_raw = self.data.map(|r| r.get());
-		let other_raw = self.data.map(|r| r.get());
+		let other_raw = other.data.map(|r| r.get());
 		self.code == other.code && self.message == other.message && this_raw == other_raw
 	}
 }
@@ -84,6 +84,8 @@ pub const PARSE_ERROR_CODE: i32 = -32700;
 pub const OVERSIZED_REQUEST_CODE: i32 = -32701;
 /// Oversized response error code.
 pub const OVERSIZED_RESPONSE_CODE: i32 = -32702;
+/// Oversized response error code.
+pub const INVALID_SUBSCRIPTION_CODE: i32 = -32703;
 /// Internal error code.
 pub const INTERNAL_ERROR_CODE: i32 = -32603;
 /// Invalid params error code.
@@ -105,6 +107,8 @@ pub const PARSE_ERROR_MSG: &str = "Parse error";
 pub const OVERSIZED_REQUEST_MSG: &str = "Request is too big";
 /// Oversized response message
 pub const OVERSIZED_RESPONSE_MSG: &str = "Response is too big";
+/// Invalid subscription ID.
+pub const INVALID_SUBSCRIPTION_MSG: &str = "Invalid subscription ID";
 /// Internal error message.
 pub const INTERNAL_ERROR_MSG: &str = "Internal error";
 /// Invalid params error message.

utils/src/server/rpc_module.rs

@@ -29,6 +29,9 @@ use crate::server::resource_limiting::{ResourceGuard, ResourceTable, ResourceVec
 use beef::Cow;
 use futures_channel::{mpsc, oneshot};
 use futures_util::{future::BoxFuture, FutureExt, StreamExt};
+use jsonrpsee_types::to_json_raw_value;
+use jsonrpsee_types::v2::error::{INVALID_SUBSCRIPTION_CODE, INVALID_SUBSCRIPTION_MSG};
+use jsonrpsee_types::v2::ErrorObject;
 use jsonrpsee_types::{
 	error::{Error, SubscriptionClosedError},
 	traits::ToRpcParams,
@@ -605,12 +608,32 @@ impl<Context: Send + Sync + 'static> RpcModule<Context> {
 								unsubscribe_method_name,
 								id
 							);
-							send_error(id, tx, ErrorCode::ServerError(-1).into());
+							send_error(
+								id,
+								tx,
+								ErrorObject {
+									code: ErrorCode::ServerError(INVALID_SUBSCRIPTION_CODE),
+									message: INVALID_SUBSCRIPTION_MSG,
+									data: to_json_raw_value(&"Subscription ID must be u64").ok().as_deref(),
+								},
+							);
 							return;
 						}
 					};
-					subscribers.lock().remove(&SubscriptionKey { conn_id, sub_id });
-					send_response(id, tx, "Unsubscribed", max_response_size);
+
+					if subscribers.lock().remove(&SubscriptionKey { conn_id, sub_id }).is_some() {
+						send_response(id, tx, "Unsubscribed", max_response_size);
+					} else {
+						send_error(
+							id,
+							tx,
+							ErrorObject {
+								code: ErrorCode::ServerError(INVALID_SUBSCRIPTION_CODE),
+								message: INVALID_SUBSCRIPTION_MSG,
+								data: to_json_raw_value(&format!("ID={} is not active", sub_id)).ok().as_deref(),
+							},
+						)
+					}
 				})),
 			);
 		}
@@ -698,6 +721,7 @@ impl SubscriptionSink {
 	fn inner_close(&mut self, err: &SubscriptionClosedError) {
 		self.is_connected.take();
 		if let Some((sink, _)) = self.subscribers.lock().remove(&self.uniq_sub) {
+			tracing::debug!("Closing subscription: {:?}", self.uniq_sub.sub_id);
 			let msg = self.build_message(err).expect("valid json infallible; qed");
 			let _ = sink.unbounded_send(msg);
 		}

ws-server/src/tests.rs

@@ -27,12 +27,15 @@
 #![cfg(test)]
 
 use crate::types::error::{CallError, Error};
+use crate::types::v2::{self, Response, RpcError};
+use crate::types::DeserializeOwned;
 use crate::{future::ServerHandle, RpcModule, WsServerBuilder};
 use anyhow::anyhow;
 use futures_util::future::join;
 use jsonrpsee_test_utils::helpers::*;
 use jsonrpsee_test_utils::mocks::{Id, TestContext, WebSocketTestClient, WebSocketTestError};
 use jsonrpsee_test_utils::TimeoutFutureExt;
+use jsonrpsee_types::to_json_raw_value;
 use serde_json::Value as JsonValue;
 use std::{fmt, net::SocketAddr, time::Duration};
 use tracing_subscriber::{EnvFilter, FmtSubscriber};
@@ -41,6 +44,11 @@ fn init_logger() {
 	let _ = FmtSubscriber::builder().with_env_filter(EnvFilter::from_default_env()).try_init();
 }
 
+fn deser_call<T: DeserializeOwned>(raw: String) -> T {
+	let out: Response<T> = serde_json::from_str(&raw).unwrap();
+	out.result
+}
+
 /// Applications can/should provide their own error.
 #[derive(Debug)]
 struct MyAppError;
@@ -107,6 +115,16 @@ async fn server_with_handles() -> (SocketAddr, ServerHandle) {
 			Ok("Yawn!")
 		})
 		.unwrap();
+	module
+		.register_subscription("subscribe_hello", "unsubscribe_hello", |_, sink, _| {
+			std::thread::spawn(move || {
+				// prevent the sink for being dropped and keep the subscription alive for testing.
+				let _ = sink;
+				loop {}
+			});
+			Ok(())
+		})
+		.unwrap();
 
 	let addr = server.local_addr().unwrap();
 
@@ -569,3 +587,58 @@ async fn run_forever() {
 	// Send the shutdown request from one handle and await the server on the second one.
 	join(server_handle.clone().stop().unwrap(), server_handle).with_timeout(TIMEOUT).await.unwrap();
 }
+
+#[tokio::test]
+async fn unsubscribe_twice_should_indicate_error() {
+	init_logger();
+	let addr = server().await;
+	let mut client = WebSocketTestClient::new(addr).with_default_timeout().await.unwrap().unwrap();
+
+	let sub_call = call("subscribe_hello", Vec::<()>::new(), Id::Num(0));
+	let sub_id: u64 = deser_call(client.send_request_text(sub_call).await.unwrap());
+
+	let unsub_call = call("unsubscribe_hello", vec![sub_id], Id::Num(1));
+	let unsub_1: String = deser_call(client.send_request_text(unsub_call).await.unwrap());
+	assert_eq!(&unsub_1, "Unsubscribed");
+
+	let unsub_call = call("unsubscribe_hello", vec![sub_id], Id::Num(2));
+	let unsub_2 = client.send_request_text(unsub_call).await.unwrap();
+	let unsub_2_err: RpcError = serde_json::from_str(&unsub_2).unwrap();
+	let sub_id = to_json_raw_value(&sub_id).unwrap();
+	assert_eq!(
+		unsub_2_err,
+		RpcError {
+			jsonrpc: v2::TwoPointZero,
+			error: v2::ErrorObject {
+				code: v2::ErrorCode::ServerError(v2::error::INVALID_SUBSCRIPTION_CODE),
+				message: v2::error::INVALID_SUBSCRIPTION_MSG,
+				data: Some(&to_json_raw_value(&format!("ID={} is not active", sub_id)).unwrap()),
+			},
+			id: v2::Id::Number(2)
+		}
+	);
+}
+
+#[tokio::test]
+async fn unsubscribe_wrong_sub_id_type() {
+	init_logger();
+	let addr = server().await;
+	let mut client = WebSocketTestClient::new(addr).with_default_timeout().await.unwrap().unwrap();
+	let data = to_json_raw_value(&"Subscription ID must be u64").unwrap();
+
+	let unsub =
+		client.send_request_text(call("unsubscribe_hello", vec!["string_is_not_supprted"], Id::Num(0))).await.unwrap();
+	let unsub_2_err: RpcError = serde_json::from_str(&unsub).unwrap();
+	assert_eq!(
+		unsub_2_err,
+		RpcError {
+			jsonrpc: v2::TwoPointZero,
+			error: v2::ErrorObject {
+				code: v2::ErrorCode::ServerError(v2::error::INVALID_SUBSCRIPTION_CODE),
+				message: v2::error::INVALID_SUBSCRIPTION_MSG,
+				data: Some(&data),
+			},
+			id: v2::Id::Number(0)
+		}
+	);
+}