jam_std_common/

rpc.rs

use super::{
	crypto::hashing::hash_raw,
	extrinsic::WorkReport,
	node::{
		BlockDesc, ChainSub, ChainSubUpdate, Error as NodeError, Node, NodeResult, SyncState,
		SyncStatus, VersionedParameters, WorkPackageStatus,
	},
};
use bytes::Bytes;
use codec::{DecodeAll, Encode};
use futures::{FutureExt, Stream, StreamExt, stream::BoxStream};
use jam_types::{
	AnyBytes, AnyHash, AnyVec, CoreIndex, ExtrinsicHash, Hash, HeaderHash, MmrPeakHash,
	SegmentBytes, SegmentTreeRoot, ServiceId, Slot, StateRootHash, WorkPackage, WorkPackageHash,
	WorkReportHash,
};
use jsonrpsee::{
	PendingSubscriptionSink, SubscriptionMessage,
	core::{ClientError, RpcResult, SubscriptionResult},
	proc_macros::rpc,
	types::{ErrorObject, ErrorObjectOwned},
};

pub mod error_codes {
	pub const OTHER: i32 = 0;
	pub const BLOCK_UNAVAILABLE: i32 = 1;
	pub const WORK_REPORT_UNAVAILABLE: i32 = 2;
	pub const SEGMENT_UNAVAILABLE: i32 = 3;
}

/// JAM node RPC interface.
///
/// `RpcServer` is not intended to be implemented directly. Instead of implementing `RpcServer`,
/// implement `Node` and use the blanket implementation of `RpcServer` for `Node` types.
///
/// Similarly, instead of using the `RpcClient` trait directly, use the `Node` trait; this is
/// implemented for all `RpcClient` types and it is also directly implemented by eg PolkaJam.
#[rpc(client, server)]
pub trait Rpc {
	/// Returns the chain parameters.
	#[method(name = "parameters")]
	async fn parameters(&self) -> RpcResult<VersionedParameters>;

	/// Returns the header hash and slot of the head of the "best" chain.
	#[method(name = "bestBlock")]
	async fn best_block(&self) -> RpcResult<BlockDesc>;

	/// Subscribe to updates of the head of the "best" chain, as returned by `bestBlock`.
	#[subscription(name = "subscribeBestBlock", item = BlockDesc)]
	async fn subscribe_best_block(&self) -> SubscriptionResult;

	/// Returns the header hash and slot of the latest finalized block.
	#[method(name = "finalizedBlock")]
	async fn finalized_block(&self) -> RpcResult<BlockDesc>;

	/// Subscribe to updates of the latest finalized block, as returned by `finalizedBlock`.
	#[subscription(name = "subscribeFinalizedBlock", item = BlockDesc)]
	async fn subscribe_finalized_block(&self) -> SubscriptionResult;

	/// Returns the header hash and slot of the parent of the block with the given header hash.
	#[method(name = "parent")]
	async fn parent(&self, header_hash: HeaderHash) -> RpcResult<BlockDesc>;

	/// Returns the posterior state root of the block with the given header hash.
	#[method(name = "stateRoot")]
	async fn state_root(&self, header_hash: HeaderHash) -> RpcResult<StateRootHash>;

	/// Returns the BEEFY root of the block with the given header hash.
	#[method(name = "beefyRoot")]
	async fn beefy_root(&self, header_hash: HeaderHash) -> RpcResult<MmrPeakHash>;

	/// Returns the activity statistics stored in the posterior state of the block with the given
	/// header hash.
	///
	/// The statistics are encoded as per the GP.
	#[method(name = "statistics")]
	async fn statistics(&self, header_hash: HeaderHash) -> RpcResult<AnyBytes>;

	/// Subscribe to updates of the activity statistics stored in chain state.
	///
	/// The statistics are encoded as per the GP.
	#[subscription(name = "subscribeStatistics", item = ChainSubUpdate<AnyBytes>)]
	async fn subscribe_statistics(&self, finalized: bool) -> SubscriptionResult;

	/// Returns the storage data for the service with the given ID.
	///
	/// The data are encoded as per the GP. `None` is returned if there is no service with the
	/// given ID.
	#[method(name = "serviceData")]
	async fn service_data(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
	) -> RpcResult<Option<AnyBytes>>;

	/// Subscribe to updates of the storage data for the service with the given ID.
	///
	/// The `value` field of updates will contain service data encoded as per the GP. It will be
	/// `None` when there is no service with the given ID.
	#[subscription(name = "subscribeServiceData", item = ChainSubUpdate<Option<AnyBytes>>)]
	async fn subscribe_service_data(&self, id: ServiceId, finalized: bool) -> SubscriptionResult;

	/// Returns the value associated with the given service ID and key in the posterior state of
	/// the block with the given header hash.
	///
	/// `None` is returned if there is no value associated with the given service ID and key.
	///
	/// This method can be used to query arbitrary key-value pairs set by service accumulation
	/// logic.
	#[method(name = "serviceValue")]
	async fn service_value(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
		key: AnyVec,
	) -> RpcResult<Option<AnyBytes>>;

	/// Subscribe to updates of the value associated with the given service ID and key.
	///
	/// The `value` field of updates will be `None` when there is no value associated with the
	/// given service ID and key.
	#[subscription(name = "subscribeServiceValue", item = ChainSubUpdate<Option<AnyBytes>>)]
	async fn subscribe_service_value(
		&self,
		id: ServiceId,
		key: AnyVec,
		finalized: bool,
	) -> SubscriptionResult;

	/// Returns the preimage of the given hash, if it has been provided to the given service in the
	/// posterior state of the block with the given header hash.
	///
	/// `None` is returned if the preimage has not been provided to the given service.
	#[method(name = "servicePreimage")]
	async fn service_preimage(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
		hash: AnyHash,
	) -> RpcResult<Option<AnyBytes>>;

	/// Subscribe to updates of the preimage associated with the given service ID and hash.
	///
	/// The `value` field of updates will be `None` if the preimage has not been provided to the
	/// service. Otherwise, it will be the preimage of the given hash.
	#[subscription(name = "subscribeServicePreimage", item = ChainSubUpdate<Option<AnyBytes>>)]
	async fn subscribe_service_preimage(
		&self,
		id: ServiceId,
		hash: AnyHash,
		finalized: bool,
	) -> SubscriptionResult;

	/// Same as [`service_preimage`](Self::service_preimage) but only returns the length of the
	/// preimage.
	#[method(name = "servicePreimageLen")]
	async fn service_preimage_len(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
		hash: AnyHash,
	) -> RpcResult<Option<u32>>;

	/// Returns the preimage request associated with the given service ID and hash/length in the
	/// posterior state of the block with the given header hash.
	///
	/// `None` is returned if the preimage with the given hash/length has neither been requested by
	/// nor provided to the given service. An empty list is returned if the preimage has been
	/// requested, but not yet provided. A non-empty list means that the preimage has been
	/// provided; the meaning of the slots in the list is as follows:
	///
	/// - The first slot is the slot in which the preimage was provided.
	/// - The second slot, if present, is the slot in which the preimage was "forgotten".
	/// - The third slot, if present, is the slot in which the preimage was requested again.
	#[method(name = "serviceRequest")]
	async fn service_request(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
		hash: AnyHash,
		len: u32,
	) -> RpcResult<Option<Vec<Slot>>>;

	/// Subscribe to updates of the preimage request associated with the given service ID and
	/// hash/length.
	///
	/// The `value` field of updates will either be `None` or a list of slots, with the same
	/// semantics as `serviceRequest` return values.
	#[subscription(name = "subscribeServiceRequest", item = ChainSubUpdate<Option<Vec<Slot>>>)]
	async fn subscribe_service_request(
		&self,
		id: ServiceId,
		hash: AnyHash,
		len: u32,
		finalized: bool,
	) -> SubscriptionResult;

	/// Returns the work-report with the given hash.
	///
	/// The work-report is encoded as per the GP.
	#[method(name = "workReport")]
	async fn work_report(&self, hash: WorkReportHash) -> RpcResult<AnyBytes>;

	/// Submit a work-package to the guarantors currently assigned to the given core.
	#[method(name = "submitWorkPackage")]
	async fn submit_work_package(
		&self,
		core: CoreIndex,
		package: AnyBytes,
		extrinsics: Vec<AnyBytes>,
	) -> RpcResult<()>;

	/// Submit a work-package bundle to the guarantors currently assigned to the given core.
	#[method(name = "submitWorkPackageBundle")]
	async fn submit_work_package_bundle(&self, core: CoreIndex, bundle: AnyBytes) -> RpcResult<()>;

	/// Returns the status of the given work-package following execution of the block with the
	/// given header hash.
	///
	/// If `anchor` does not match the anchor of the work-package then an error or an incorrect
	/// status may be returned. An error may be returned if `anchor` is too old.
	#[method(name = "workPackageStatus")]
	async fn work_package_status(
		&self,
		header_hash: HeaderHash,
		hash: WorkPackageHash,
		anchor: HeaderHash,
	) -> RpcResult<WorkPackageStatus>;

	/// Subscribe to status updates for the given work-package.
	///
	/// If `anchor` does not match the anchor of the work-package then the subscription may fail or
	/// yield incorrect statuses. The subscription may fail if `anchor` is too old.
	#[subscription(name = "subscribeWorkPackageStatus", item = ChainSubUpdate<WorkPackageStatus>)]
	async fn subscribe_work_package_status(
		&self,
		hash: WorkPackageHash,
		anchor: HeaderHash,
		finalized: bool,
	) -> SubscriptionResult;

	/// Submit a preimage which is being requested by the given service.
	///
	/// Note that this method does not wait for the preimage to be distributed or integrated
	/// on-chain; it returns immediately.
	#[method(name = "submitPreimage")]
	async fn submit_preimage(&self, requester: ServiceId, preimage: AnyBytes) -> RpcResult<()>;

	/// Returns a list of all services currently known to be on JAM.
	///
	/// This is a best-effort list and may not reflect the true state. Nodes could e.g. reasonably
	/// hide services which are not recently active from this list.
	#[method(name = "listServices")]
	async fn list_services(&self, header_hash: HeaderHash) -> RpcResult<Vec<ServiceId>>;

	/// Fetches a list of segments from the DA layer, exported by the work-package with the given
	/// hash.
	#[method(name = "fetchWorkPackageSegments")]
	async fn fetch_work_package_segments(
		&self,
		wp_hash: WorkPackageHash,
		indices: Vec<u16>,
	) -> RpcResult<Vec<SegmentBytes>>;

	/// Fetches a list of segments from the DA layer, exported by a work-package with the given
	/// segment root hash.
	#[method(name = "fetchSegments")]
	async fn fetch_segments(
		&self,
		segment_root: SegmentTreeRoot,
		indices: Vec<u16>,
	) -> RpcResult<Vec<SegmentBytes>>;

	/// Returns the sync status of the node.
	#[method(name = "syncState")]
	async fn sync_state(&self) -> RpcResult<SyncState>;

	/// Subscribe to changes in sync status.
	#[subscription(name = "subscribeSyncStatus", item = SyncStatus)]
	async fn subscribe_sync_status(&self) -> SubscriptionResult;
}

impl From<ClientError> for NodeError {
	fn from(err: ClientError) -> Self {
		match err {
			ClientError::Call(err) => {
				match err.code() {
					error_codes::BLOCK_UNAVAILABLE =>
						if let Some(data) = err.data() &&
							let Ok(hash) = serde_json::from_str(data.get())
						{
							return NodeError::BlockUnavailable(hash)
						},
					error_codes::WORK_REPORT_UNAVAILABLE =>
						if let Some(data) = err.data() &&
							let Ok(hash) = serde_json::from_str(data.get())
						{
							return NodeError::WorkReportUnavailable(hash)
						},
					error_codes::SEGMENT_UNAVAILABLE => return NodeError::SegmentUnavailable,
					_ => (),
				}
				NodeError::Other(err.to_string())
			},
			ClientError::Transport(..) | ClientError::RequestTimeout => NodeError::NetworkFailure,
			_ => NodeError::Other(err.to_string()),
		}
	}
}

impl From<serde_json::Error> for NodeError {
	fn from(err: serde_json::Error) -> Self {
		NodeError::Other(err.to_string())
	}
}

#[async_trait::async_trait]
impl<T: RpcClient + Send + Sync> Node for T {
	async fn parameters(&self) -> NodeResult<VersionedParameters> {
		Ok(<T as RpcClient>::parameters(self).await?)
	}

	async fn best_block(&self) -> NodeResult<BlockDesc> {
		Ok(<T as RpcClient>::best_block(self).await?)
	}

	async fn subscribe_best_block(&self) -> NodeResult<BoxStream<NodeResult<BlockDesc>>> {
		let sub = <T as RpcClient>::subscribe_best_block(self).await?;
		Ok(sub.map(|res| Ok(res?)).boxed())
	}

	async fn finalized_block(&self) -> NodeResult<BlockDesc> {
		Ok(<T as RpcClient>::finalized_block(self).await?)
	}

	async fn subscribe_finalized_block(&self) -> NodeResult<BoxStream<NodeResult<BlockDesc>>> {
		let sub = <T as RpcClient>::subscribe_finalized_block(self).await?;
		Ok(sub.map(|res| Ok(res?)).boxed())
	}

	async fn parent(&self, header_hash: HeaderHash) -> NodeResult<BlockDesc> {
		Ok(<T as RpcClient>::parent(self, header_hash).await?)
	}

	async fn state_root(&self, header_hash: HeaderHash) -> NodeResult<StateRootHash> {
		Ok(<T as RpcClient>::state_root(self, header_hash).await?)
	}

	async fn beefy_root(&self, header_hash: HeaderHash) -> NodeResult<MmrPeakHash> {
		Ok(<T as RpcClient>::beefy_root(self, header_hash).await?)
	}

	async fn encoded_statistics(&self, header_hash: HeaderHash) -> NodeResult<Bytes> {
		Ok(<T as RpcClient>::statistics(self, header_hash).await?.into())
	}

	async fn subscribe_encoded_statistics(&self, finalized: bool) -> NodeResult<ChainSub<Bytes>> {
		let sub = <T as RpcClient>::subscribe_statistics(self, finalized).await?;
		Ok(sub.map(|res| Ok(res?.map(Into::into))).boxed())
	}

	async fn encoded_service_data(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
	) -> NodeResult<Option<Bytes>> {
		Ok(<T as RpcClient>::service_data(self, header_hash, id).await?.map(Into::into))
	}

	async fn subscribe_encoded_service_data(
		&self,
		id: ServiceId,
		finalized: bool,
	) -> NodeResult<ChainSub<Option<Bytes>>> {
		let sub = <T as RpcClient>::subscribe_service_data(self, id, finalized).await?;
		Ok(sub.map(|res| Ok(res?.map_some(Into::into))).boxed())
	}

	async fn service_value(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
		key: &[u8],
	) -> NodeResult<Option<Bytes>> {
		Ok(<T as RpcClient>::service_value(self, header_hash, id, key.to_vec().into())
			.await?
			.map(Into::into))
	}

	async fn subscribe_service_value(
		&self,
		id: ServiceId,
		key: &[u8],
		finalized: bool,
	) -> NodeResult<ChainSub<Option<Bytes>>> {
		let sub =
			<T as RpcClient>::subscribe_service_value(self, id, key.to_vec().into(), finalized)
				.await?;
		Ok(sub.map(|res| Ok(res?.map_some(Into::into))).boxed())
	}

	async fn service_preimage(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
		hash: Hash,
	) -> NodeResult<Option<Bytes>> {
		Ok(<T as RpcClient>::service_preimage(self, header_hash, id, hash.into())
			.await?
			.map(Into::into))
	}

	async fn service_preimage_len(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
		hash: Hash,
	) -> NodeResult<Option<u32>> {
		Ok(<T as RpcClient>::service_preimage_len(self, header_hash, id, hash.into()).await?)
	}

	async fn subscribe_service_preimage(
		&self,
		id: ServiceId,
		hash: Hash,
		finalized: bool,
	) -> NodeResult<ChainSub<Option<Bytes>>> {
		let sub =
			<T as RpcClient>::subscribe_service_preimage(self, id, hash.into(), finalized).await?;
		Ok(sub.map(|res| Ok(res?.map_some(Into::into))).boxed())
	}

	async fn service_request(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
		hash: Hash,
		len: u32,
	) -> NodeResult<Option<Vec<Slot>>> {
		Ok(<T as RpcClient>::service_request(self, header_hash, id, hash.into(), len).await?)
	}

	async fn subscribe_service_request(
		&self,
		id: ServiceId,
		hash: Hash,
		len: u32,
		finalized: bool,
	) -> NodeResult<ChainSub<Option<Vec<Slot>>>> {
		let sub =
			<T as RpcClient>::subscribe_service_request(self, id, hash.into(), len, finalized)
				.await?;
		Ok(sub.map(|res| Ok(res?)).boxed())
	}

	async fn work_report(&self, hash: WorkReportHash) -> NodeResult<WorkReport> {
		let encoded = <T as RpcClient>::work_report(self, hash).await?;
		Ok(WorkReport::decode_all(&mut &encoded[..])?)
	}

	async fn submit_encoded_work_package(
		&self,
		core: CoreIndex,
		package: Bytes,
		extrinsics: &[Bytes],
	) -> NodeResult<()> {
		Ok(<T as RpcClient>::submit_work_package(
			self,
			core,
			package.into(),
			extrinsics.iter().cloned().map(Into::into).collect(),
		)
		.await?)
	}

	async fn submit_encoded_work_package_bundle(
		&self,
		core: CoreIndex,
		bundle: Bytes,
	) -> NodeResult<()> {
		Ok(<T as RpcClient>::submit_work_package_bundle(self, core, bundle.into()).await?)
	}

	async fn work_package_status(
		&self,
		header_hash: HeaderHash,
		hash: WorkPackageHash,
		anchor: HeaderHash,
	) -> NodeResult<WorkPackageStatus> {
		Ok(<T as RpcClient>::work_package_status(self, header_hash, hash, anchor).await?)
	}

	async fn subscribe_work_package_status(
		&self,
		hash: WorkPackageHash,
		anchor: HeaderHash,
		finalized: bool,
	) -> NodeResult<ChainSub<WorkPackageStatus>> {
		let sub =
			<T as RpcClient>::subscribe_work_package_status(self, hash, anchor, finalized).await?;
		Ok(sub.map(|res| Ok(res?)).boxed())
	}

	async fn submit_preimage(&self, requester: ServiceId, preimage: Bytes) -> NodeResult<()> {
		Ok(<T as RpcClient>::submit_preimage(self, requester, preimage.into()).await?)
	}

	async fn list_services(&self, header_hash: HeaderHash) -> NodeResult<Vec<ServiceId>> {
		Ok(<T as RpcClient>::list_services(self, header_hash).await?)
	}

	async fn fetch_work_package_segments(
		&self,
		wp_hash: WorkPackageHash,
		indices: Vec<u16>,
	) -> NodeResult<Vec<SegmentBytes>> {
		Ok(<T as RpcClient>::fetch_work_package_segments(self, wp_hash, indices).await?)
	}

	async fn fetch_segments(
		&self,
		segment_root: SegmentTreeRoot,
		indices: Vec<u16>,
	) -> NodeResult<Vec<SegmentBytes>> {
		Ok(<T as RpcClient>::fetch_segments(self, segment_root, indices).await?)
	}

	async fn sync_state(&self) -> NodeResult<SyncState> {
		Ok(<T as RpcClient>::sync_state(self).await?)
	}

	async fn subscribe_sync_status(&self) -> NodeResult<BoxStream<NodeResult<SyncStatus>>> {
		let sub = <T as RpcClient>::subscribe_sync_status(self).await?;
		Ok(sub.map(|res| Ok(res?)).boxed())
	}
}

impl From<NodeError> for ErrorObjectOwned {
	fn from(err: NodeError) -> Self {
		match err {
			NodeError::Other(message) =>
				ErrorObject::owned(error_codes::OTHER, message, None::<()>),
			NodeError::BlockUnavailable(hash) =>
				ErrorObject::owned(error_codes::BLOCK_UNAVAILABLE, err.to_string(), Some(hash)),
			NodeError::WorkReportUnavailable(hash) => ErrorObject::owned(
				error_codes::WORK_REPORT_UNAVAILABLE,
				err.to_string(),
				Some(hash),
			),
			NodeError::SegmentUnavailable =>
				ErrorObject::owned(error_codes::SEGMENT_UNAVAILABLE, err.to_string(), None::<()>),
			NodeError::NetworkFailure =>
				ErrorObject::owned(error_codes::OTHER, "Network failure", None::<()>),
		}
	}
}

async fn relay_subscription<T: serde::Serialize>(
	sub: NodeResult<impl Stream<Item = NodeResult<T>> + Unpin>,
	pending: PendingSubscriptionSink,
) -> SubscriptionResult {
	let mut sub = match sub {
		Ok(sub) => sub,
		Err(err) => {
			pending.reject(err).await;
			return Ok(())
		},
	};
	let sink = pending.accept().await?;
	// `sink` holds the connection's subscription permit until this task returns, and jsonrpsee
	// never cancels the task, so a never-ending stream leaks a permit on every unsubscribe.
	loop {
		futures::select! {
			res = sub.next().fuse() => match res {
				Some(res) => {
					let item: T = res?;
					let msg = SubscriptionMessage::new(
						sink.method_name(),
						sink.subscription_id(),
						&item,
					)?;
					sink.send(msg).await?;
				},
				None => break,
			},
			_ = sink.closed().fuse() => break,
		}
	}
	Ok(())
}

/// Returns `Ok` iff `extrinsics` matches the extrinsic specs in `package`.
fn check_extrinsics(package: &WorkPackage, extrinsics: &[Bytes]) -> NodeResult<()> {
	let num_specs = package.extrinsic_count();
	if extrinsics.len() != (num_specs as usize) {
		return Err(NodeError::Other(format!(
			"{} extrinsics provided, package specifies {num_specs}",
			extrinsics.len()
		)))
	}

	for (i, (spec, extrinsic)) in package
		.items
		.iter()
		.flat_map(|item| item.extrinsics.iter())
		.zip(extrinsics.iter())
		.enumerate()
	{
		if extrinsic.len() != (spec.len as usize) {
			return Err(NodeError::Other(format!(
				"Extrinsic {i} has length {}, package specifies length {}",
				extrinsic.len(),
				spec.len
			)));
		}

		let hash: ExtrinsicHash = hash_raw(extrinsic).into();
		if hash != spec.hash {
			return Err(NodeError::Other(format!(
				"Extrinsic {i} has hash {hash}, package specifies hash {}",
				spec.hash
			)));
		}
	}

	Ok(())
}

fn check_package(mut package: &[u8], extrinsics: &[Bytes]) -> NodeResult<()> {
	let package = WorkPackage::decode_all(&mut package)?;
	check_extrinsics(&package, extrinsics)
}

#[async_trait::async_trait]
impl<T: Node + 'static> RpcServer for T {
	async fn parameters(&self) -> RpcResult<VersionedParameters> {
		Ok(<T as Node>::parameters(self).await?)
	}

	async fn best_block(&self) -> RpcResult<BlockDesc> {
		Ok(<T as Node>::best_block(self).await?)
	}

	async fn subscribe_best_block(&self, pending: PendingSubscriptionSink) -> SubscriptionResult {
		relay_subscription::<BlockDesc>(<T as Node>::subscribe_best_block(self).await, pending)
			.await
	}

	async fn finalized_block(&self) -> RpcResult<BlockDesc> {
		Ok(<T as Node>::finalized_block(self).await?)
	}

	async fn subscribe_finalized_block(
		&self,
		pending: PendingSubscriptionSink,
	) -> SubscriptionResult {
		relay_subscription::<BlockDesc>(<T as Node>::subscribe_finalized_block(self).await, pending)
			.await
	}

	async fn parent(&self, header_hash: HeaderHash) -> RpcResult<BlockDesc> {
		Ok(<T as Node>::parent(self, header_hash).await?)
	}

	async fn state_root(&self, header_hash: HeaderHash) -> RpcResult<StateRootHash> {
		Ok(<T as Node>::state_root(self, header_hash).await?)
	}

	async fn beefy_root(&self, header_hash: HeaderHash) -> RpcResult<MmrPeakHash> {
		Ok(<T as Node>::beefy_root(self, header_hash).await?)
	}

	async fn statistics(&self, header_hash: HeaderHash) -> RpcResult<AnyBytes> {
		Ok(<T as Node>::encoded_statistics(self, header_hash).await?.into())
	}

	async fn subscribe_statistics(
		&self,
		pending: PendingSubscriptionSink,
		finalized: bool,
	) -> SubscriptionResult {
		relay_subscription::<ChainSubUpdate<AnyBytes>>(
			<T as Node>::subscribe_encoded_statistics(self, finalized)
				.await
				.map(|sub| sub.map(|res| Ok(res?.map(Into::into)))),
			pending,
		)
		.await
	}

	async fn service_data(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
	) -> RpcResult<Option<AnyBytes>> {
		Ok(<T as Node>::encoded_service_data(self, header_hash, id).await?.map(Into::into))
	}

	async fn subscribe_service_data(
		&self,
		pending: PendingSubscriptionSink,
		id: ServiceId,
		finalized: bool,
	) -> SubscriptionResult {
		relay_subscription::<ChainSubUpdate<Option<AnyBytes>>>(
			<T as Node>::subscribe_encoded_service_data(self, id, finalized)
				.await
				.map(|sub| sub.map(|res| Ok(res?.map_some(Into::into)))),
			pending,
		)
		.await
	}

	async fn service_value(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
		key: AnyVec,
	) -> RpcResult<Option<AnyBytes>> {
		Ok(<T as Node>::service_value(self, header_hash, id, &key).await?.map(Into::into))
	}

	async fn subscribe_service_value(
		&self,
		pending: PendingSubscriptionSink,
		id: ServiceId,
		key: AnyVec,
		finalized: bool,
	) -> SubscriptionResult {
		relay_subscription::<ChainSubUpdate<Option<AnyBytes>>>(
			<T as Node>::subscribe_service_value(self, id, &key, finalized)
				.await
				.map(|sub| sub.map(|res| Ok(res?.map_some(Into::into)))),
			pending,
		)
		.await
	}

	async fn service_preimage(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
		hash: AnyHash,
	) -> RpcResult<Option<AnyBytes>> {
		Ok(<T as Node>::service_preimage(self, header_hash, id, hash.into())
			.await?
			.map(Into::into))
	}

	async fn service_preimage_len(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
		hash: AnyHash,
	) -> RpcResult<Option<u32>> {
		Ok(<T as Node>::service_preimage_len(self, header_hash, id, hash.into()).await?)
	}

	async fn subscribe_service_preimage(
		&self,
		pending: PendingSubscriptionSink,
		id: ServiceId,
		hash: AnyHash,
		finalized: bool,
	) -> SubscriptionResult {
		relay_subscription::<ChainSubUpdate<Option<AnyBytes>>>(
			<T as Node>::subscribe_service_preimage(self, id, hash.into(), finalized)
				.await
				.map(|sub| sub.map(|res| Ok(res?.map_some(Into::into)))),
			pending,
		)
		.await
	}

	async fn service_request(
		&self,
		header_hash: HeaderHash,
		id: ServiceId,
		hash: AnyHash,
		len: u32,
	) -> RpcResult<Option<Vec<Slot>>> {
		Ok(<T as Node>::service_request(self, header_hash, id, hash.into(), len).await?)
	}

	async fn subscribe_service_request(
		&self,
		pending: PendingSubscriptionSink,
		id: ServiceId,
		hash: AnyHash,
		len: u32,
		finalized: bool,
	) -> SubscriptionResult {
		relay_subscription::<ChainSubUpdate<Option<Vec<Slot>>>>(
			<T as Node>::subscribe_service_request(self, id, hash.into(), len, finalized).await,
			pending,
		)
		.await
	}

	async fn work_report(&self, hash: WorkReportHash) -> RpcResult<AnyBytes> {
		let report = <T as Node>::work_report(self, hash).await?;
		Ok(Bytes::from(report.encode()).into())
	}

	async fn submit_work_package(
		&self,
		core: CoreIndex,
		package: AnyBytes,
		extrinsics: Vec<AnyBytes>,
	) -> RpcResult<()> {
		// This _should_ be a no-op!
		let extrinsics: Vec<Bytes> = extrinsics.into_iter().map(Into::into).collect();
		check_package(&package, &extrinsics)?;
		Ok(<T as Node>::submit_encoded_work_package(self, core, package.into(), &extrinsics)
			.await?)
	}

	async fn submit_work_package_bundle(&self, core: CoreIndex, bundle: AnyBytes) -> RpcResult<()> {
		Ok(<T as Node>::submit_encoded_work_package_bundle(self, core, bundle.into()).await?)
	}

	async fn work_package_status(
		&self,
		header_hash: HeaderHash,
		hash: WorkPackageHash,
		anchor: HeaderHash,
	) -> RpcResult<WorkPackageStatus> {
		Ok(<T as Node>::work_package_status(self, header_hash, hash, anchor).await?)
	}

	async fn subscribe_work_package_status(
		&self,
		pending: PendingSubscriptionSink,
		hash: WorkPackageHash,
		anchor: HeaderHash,
		finalized: bool,
	) -> SubscriptionResult {
		relay_subscription::<ChainSubUpdate<WorkPackageStatus>>(
			<T as Node>::subscribe_work_package_status(self, hash, anchor, finalized).await,
			pending,
		)
		.await
	}

	async fn submit_preimage(&self, requester: ServiceId, preimage: AnyBytes) -> RpcResult<()> {
		Ok(<T as Node>::submit_preimage(self, requester, preimage.into()).await?)
	}

	async fn list_services(&self, header_hash: HeaderHash) -> RpcResult<Vec<ServiceId>> {
		Ok(<T as Node>::list_services(self, header_hash).await?)
	}

	async fn fetch_work_package_segments(
		&self,
		wp_hash: WorkPackageHash,
		indices: Vec<u16>,
	) -> RpcResult<Vec<SegmentBytes>> {
		Ok(<T as Node>::fetch_work_package_segments(self, wp_hash, indices).await?)
	}

	async fn fetch_segments(
		&self,
		segment_root: SegmentTreeRoot,
		indices: Vec<u16>,
	) -> RpcResult<Vec<SegmentBytes>> {
		Ok(<T as Node>::fetch_segments(self, segment_root, indices).await?)
	}

	async fn sync_state(&self) -> RpcResult<SyncState> {
		Ok(<T as Node>::sync_state(self).await?)
	}

	async fn subscribe_sync_status(&self, pending: PendingSubscriptionSink) -> SubscriptionResult {
		relay_subscription::<SyncStatus>(<T as Node>::subscribe_sync_status(self).await, pending)
			.await
	}
}

#[cfg(test)]
mod tests {
	use super::*;
	use jsonrpsee::{
		RpcModule,
		core::client::{Subscription, SubscriptionClientT},
		rpc_params,
		server::{Server, ServerConfig},
		ws_client::{WsClient, WsClientBuilder},
	};
	use std::time::Duration;

	const MAX_SUBSCRIPTIONS: u32 = 2;
	const SUBSCRIBE_METHOD: &str = "subscribeQuiet";
	const NOTIFICATION_METHOD: &str = "quiet";
	const UNSUBSCRIBE_METHOD: &str = "unsubscribeQuiet";

	/// Subscribes, retrying for as long as the server still considers the connection's
	/// subscription budget exhausted; permits are released asynchronously by the server.
	async fn subscribe(client: &WsClient) -> Subscription<()> {
		const RETRY_INTERVAL: Duration = Duration::from_millis(20);
		const MAX_ATTEMPTS: usize = 100;

		let mut last_error = None;
		for _ in 0..MAX_ATTEMPTS {
			match client
				.subscribe::<(), _>(SUBSCRIBE_METHOD, rpc_params![], UNSUBSCRIBE_METHOD)
				.await
			{
				Ok(subscription) => return subscription,
				Err(err) => {
					last_error = Some(err);
					tokio::time::sleep(RETRY_INTERVAL).await;
				},
			}
		}
		panic!("subscribing kept failing: {}", last_error.expect("at least one attempt was made"));
	}

	/// A subscription whose inner stream never yields anything must still let go of its
	/// per-connection subscription permit once the client unsubscribes. Otherwise a long-lived
	/// connection can subscribe only `max_subscriptions_per_connection` times in total, no matter
	/// how diligently it unsubscribes.
	#[tokio::test]
	async fn unsubscribing_from_a_quiet_subscription_frees_the_connection_budget() {
		let server = Server::builder()
			.set_config(
				ServerConfig::builder()
					.max_subscriptions_per_connection(MAX_SUBSCRIPTIONS)
					.build(),
			)
			.build("127.0.0.1:0")
			.await
			.unwrap();
		let url = format!("ws://{}", server.local_addr().unwrap());

		let mut module = RpcModule::new(());
		module
			.register_subscription(
				SUBSCRIBE_METHOD,
				NOTIFICATION_METHOD,
				UNSUBSCRIBE_METHOD,
				|_params, pending, _context, _extensions| async move {
					relay_subscription::<()>(Ok(futures::stream::pending()), pending).await
				},
			)
			.unwrap();
		let handle = server.start(module);

		let client = WsClientBuilder::default().build(&url).await.unwrap();
		for _ in 0..(MAX_SUBSCRIPTIONS + 1) {
			subscribe(&client).await.unsubscribe().await.unwrap();
		}

		handle.stop().unwrap();
	}
}