jsonrpsee: fix bad params in subscriptions.

Cargo.toml

@@ -297,3 +297,6 @@ inherits = "release"
 lto = "fat"
 # https://doc.rust-lang.org/rustc/codegen-options/index.html#codegen-units
 codegen-units = 1
+
+[patch.crates-io]
+jsonrpsee = { git = "https://github.com/paritytech/jsonrpsee", branch = "master" }

client/beefy/rpc/src/lib.rs

@@ -30,7 +30,8 @@ use futures::{task::SpawnError, FutureExt, StreamExt};
 use jsonrpsee::{
 	core::{async_trait, Error as JsonRpseeError, RpcResult},
 	proc_macros::rpc,
-	SubscriptionSink,
+	types::{error::CallError, ErrorObject},
+	PendingSubscription,
 };
 use log::warn;
 
@@ -49,7 +50,36 @@ pub enum Error {
 	RpcTaskFailure(#[from] SpawnError),
 }
 
-/// Provides RPC methods for interacting with BEEFY.
+/// The error codes returned by jsonrpc.
+pub enum ErrorCode {
+	/// Returned when BEEFY RPC endpoint is not ready.
+	NotReady = 1,
+	/// Returned on BEEFY RPC background task failure.
+	TaskFailure = 2,
+}
+
+impl From<Error> for ErrorCode {
+	fn from(error: Error) -> Self {
+		match error {
+			Error::EndpointNotReady => ErrorCode::NotReady,
+			Error::RpcTaskFailure(_) => ErrorCode::TaskFailure,
+		}
+	}
+}
+
+impl From<Error> for JsonRpseeError {
+	fn from(error: Error) -> Self {
+		let message = error.to_string();
+		let code = ErrorCode::from(error);
+		JsonRpseeError::Call(CallError::Custom(ErrorObject::owned(
+			code as i32,
+			message,
+			None::<()>,
+		)))
+	}
+}
+
+// Provides RPC methods for interacting with BEEFY.
 #[rpc(client, server)]
 pub trait BeefyApi<Notification, Hash> {
 	/// Returns the block most recently finalized by BEEFY, alongside side its justification.
@@ -58,7 +88,7 @@ pub trait BeefyApi<Notification, Hash> {
 		unsubscribe = "beefy_unsubscribeJustifications",
 		item = Notification,
 	)]
-	fn subscribe_justifications(&self) -> RpcResult<()>;
+	fn subscribe_justifications(&self);
 
 	/// Returns hash of the latest BEEFY finalized block as seen by this client.
 	///
@@ -106,17 +136,21 @@ impl<Block> BeefyApiServer<notification::EncodedSignedCommitment, Block::Hash>
 where
 	Block: BlockT,
 {
-	fn subscribe_justifications(&self, sink: SubscriptionSink) -> RpcResult<()> {
+	fn subscribe_justifications(&self, pending: PendingSubscription) {
 		let stream = self
 			.signed_commitment_stream
 			.subscribe()
 			.map(|sc| notification::EncodedSignedCommitment::new::<Block>(sc));
 
-		let fut = sink.pipe_from_stream(stream).map(|_| ()).boxed();
+		let fut = async move {
+			if let Some(mut sink) = pending.accept() {
+				sink.pipe_from_stream(stream).await;
+			}
+		}
+		.boxed();
 
 		self.executor
 			.spawn("substrate-rpc-subscription", Some("rpc"), fut.map(drop).boxed());
-		Ok(())
 	}
 
 	async fn latest_finalized(&self) -> RpcResult<Block::Hash> {
@@ -125,7 +159,7 @@ where
 			.as_ref()
 			.cloned()
 			.ok_or(Error::EndpointNotReady)
-			.map_err(|e| JsonRpseeError::to_call_error(e))
+			.map_err(Into::into)
 	}
 }
 
@@ -168,7 +202,7 @@ mod tests {
 	async fn uninitialized_rpc_handler() {
 		let (rpc, _) = setup_io_handler();
 		let request = r#"{"jsonrpc":"2.0","method":"beefy_getFinalizedHead","params":[],"id":1}"#;
-		let expected_response = r#"{"jsonrpc":"2.0","error":{"code":-32000,"message":"BEEFY RPC endpoint not ready"},"id":1}"#.to_string();
+		let expected_response = r#"{"jsonrpc":"2.0","error":{"code":1,"message":"BEEFY RPC endpoint not ready"},"id":1}"#.to_string();
 		let (result, _) = rpc.raw_json_request(&request).await.unwrap();
 
 		assert_eq!(expected_response, result,);
@@ -193,7 +227,7 @@ mod tests {
 		.to_string();
 		let not_ready = "{\
 			\"jsonrpc\":\"2.0\",\
-			\"error\":{\"code\":-32000,\"message\":\"BEEFY RPC endpoint not ready\"},\
+			\"error\":{\"code\":1,\"message\":\"BEEFY RPC endpoint not ready\"},\
 			\"id\":1\
 		}"
 		.to_string();

client/consensus/babe/rpc/src/lib.rs

@@ -22,6 +22,7 @@ use futures::TryFutureExt;
 use jsonrpsee::{
 	core::{async_trait, Error as JsonRpseeError, RpcResult},
 	proc_macros::rpc,
+	types::{error::CallError, ErrorObject},
 };
 
 use sc_consensus_babe::{authorship, Config, Epoch};
@@ -172,7 +173,11 @@ pub enum Error {
 
 impl From<Error> for JsonRpseeError {
 	fn from(error: Error) -> Self {
-		JsonRpseeError::to_call_error(error)
+		JsonRpseeError::Call(CallError::Custom(ErrorObject::owned(
+			1234,
+			error.to_string(),
+			None::<()>,
+		)))
 	}
 }
 
@@ -267,7 +272,7 @@ mod tests {
 
 		let request = r#"{"jsonrpc":"2.0","method":"babe_epochAuthorship","params":[],"id":1}"#;
 		let (response, _) = api.raw_json_request(request).await.unwrap();
-		let expected = r#"{"jsonrpc":"2.0","error":{"code":-32000,"message":"RPC call is unsafe to be called externally"},"id":1}"#;
+		let expected = r#"{"jsonrpc":"2.0","error":{"code":-32601,"message":"RPC call is unsafe to be called externally"},"id":1}"#;
 
 		assert_eq!(&response, expected);
 	}

client/consensus/manual-seal/src/error.rs

@@ -20,7 +20,10 @@
 //! This is suitable for a testing environment.
 
 use futures::channel::{mpsc::SendError, oneshot};
-use jsonrpsee::{core::Error as JsonRpseeError, types::error::CallError};
+use jsonrpsee::{
+	core::Error as JsonRpseeError,
+	types::error::{CallError, ErrorObject},
+};
 use sc_consensus::ImportResult;
 use sp_blockchain::Error as BlockchainError;
 use sp_consensus::Error as ConsensusError;
@@ -105,6 +108,6 @@ impl Error {
 
 impl From<Error> for JsonRpseeError {
 	fn from(err: Error) -> Self {
-		CallError::Custom { code: err.to_code(), message: err.to_string(), data: None }.into()
+		CallError::Custom(ErrorObject::owned(err.to_code(), err.to_string(), None::<()>)).into()
 	}
 }

client/finality-grandpa/rpc/src/error.rs

@@ -16,7 +16,10 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-use jsonrpsee::types::error::CallError;
+use jsonrpsee::{
+	core::Error as JsonRpseeError,
+	types::error::{CallError, ErrorObject},
+};
 
 #[derive(Debug, thiserror::Error)]
 /// Top-level error type for the RPC handler
@@ -58,11 +61,15 @@ impl From<Error> for ErrorCode {
 	}
 }
 
-impl From<Error> for CallError {
+impl From<Error> for JsonRpseeError {
 	fn from(error: Error) -> Self {
 		let message = error.to_string();
 		let code = ErrorCode::from(error);
-		Self::Custom { code: code as i32, message, data: None }
+		JsonRpseeError::Call(CallError::Custom(ErrorObject::owned(
+			code as i32,
+			message,
+			None::<()>,
+		)))
 	}
 }
 

client/finality-grandpa/rpc/src/lib.rs

@@ -24,9 +24,9 @@ use log::warn;
 use std::sync::Arc;
 
 use jsonrpsee::{
-	core::{async_trait, Error as JsonRpseeError, RpcResult},
+	core::{async_trait, RpcResult},
 	proc_macros::rpc,
-	SubscriptionSink,
+	PendingSubscription,
 };
 
 mod error;
@@ -57,7 +57,7 @@ pub trait GrandpaApi<Notification, Hash, Number> {
 		unsubscribe = "grandpa_unsubscribeJustifications",
 		item = Notification
 	)]
-	fn subscribe_justifications(&self) -> RpcResult<()>;
+	fn subscribe_justifications(&self);
 
 	/// Prove finality for the given block number by returning the Justification for the last block
 	/// in the set and all the intermediary headers to link them together.
@@ -99,21 +99,25 @@ where
 	ProofProvider: RpcFinalityProofProvider<Block> + Send + Sync + 'static,
 {
 	async fn round_state(&self) -> RpcResult<ReportedRoundStates> {
-		ReportedRoundStates::from(&self.authority_set, &self.voter_state)
-			.map_err(|e| JsonRpseeError::to_call_error(e))
+		ReportedRoundStates::from(&self.authority_set, &self.voter_state).map_err(Into::into)
 	}
 
-	fn subscribe_justifications(&self, sink: SubscriptionSink) -> RpcResult<()> {
+	fn subscribe_justifications(&self, pending: PendingSubscription) {
 		let stream = self.justification_stream.subscribe().map(
 			|x: sc_finality_grandpa::GrandpaJustification<Block>| {
 				JustificationNotification::from(x)
 			},
 		);
 
-		let fut = sink.pipe_from_stream(stream).map(|_| ()).boxed();
+		let fut = async move {
+			if let Some(mut sink) = pending.accept() {
+				sink.pipe_from_stream(stream).await;
+			}
+		}
+		.boxed();
+
 		self.executor
 			.spawn("substrate-rpc-subscription", Some("rpc"), fut.map(drop).boxed());
-		Ok(())
 	}
 
 	async fn prove_finality(
@@ -122,8 +126,11 @@ where
 	) -> RpcResult<Option<EncodedFinalityProof>> {
 		self.finality_proof_provider
 			.rpc_prove_finality(block)
-			.map_err(|finality_err| error::Error::ProveFinalityFailed(finality_err))
-			.map_err(|e| JsonRpseeError::to_call_error(e))
+			.map_err(|e| {
+				warn!("Error proving finality: {}", e);
+				error::Error::ProveFinalityFailed(e)
+			})
+			.map_err(Into::into)
 	}
 }
 
@@ -276,7 +283,7 @@ mod tests {
 	#[tokio::test]
 	async fn uninitialized_rpc_handler() {
 		let (rpc, _) = setup_io_handler(EmptyVoterState);
-		let expected_response = r#"{"jsonrpc":"2.0","error":{"code":-32000,"message":"GRANDPA RPC endpoint not ready"},"id":0}"#.to_string();
+		let expected_response = r#"{"jsonrpc":"2.0","error":{"code":1,"message":"GRANDPA RPC endpoint not ready"},"id":0}"#.to_string();
 		let request = r#"{"jsonrpc":"2.0","method":"grandpa_roundState","params":[],"id":0}"#;
 		let (result, _) = rpc.raw_json_request(&request).await.unwrap();