make the ws buffer size configurable (#10013)

* make the ws buffer size configurable

* Update client/cli/src/commands/run_cmd.rs

Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

* Update client/cli/src/commands/run_cmd.rs

Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

* Update client/cli/src/commands/run_cmd.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Final touches

* Apply suggestions from code review

* fix bench

* remove in buffer

* Apply suggestions from code review

Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com>
Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

substrate/bin/node/cli/benches/transaction_pool.rs

@@ -85,6 +85,7 @@ fn new_node(tokio_handle: Handle) -> node_cli::service::NewFullBase {
 		rpc_cors: None,
 		rpc_methods: Default::default(),
 		rpc_max_payload: None,
+		ws_max_out_buffer_capacity: None,
 		prometheus_config: None,
 		telemetry_endpoints: None,
 		default_heap_pages: None,

substrate/client/cli/src/commands/run_cmd.rs

@@ -127,6 +127,10 @@ pub struct RunCmd {
 	#[structopt(long = "ws-max-connections", value_name = "COUNT")]
 	pub ws_max_connections: Option<usize>,
 
+	/// Set the the maximum WebSocket output buffer size in MiB. Default is 16.
+	#[structopt(long = "ws-max-out-buffer-capacity")]
+	pub ws_max_out_buffer_capacity: Option<usize>,
+
 	/// Specify browser Origins allowed to access the HTTP & WS RPC servers.
 	///
 	/// A comma-separated list of origins (protocol://domain or special `null`
@@ -434,6 +438,10 @@ impl CliConfiguration for RunCmd {
 		Ok(self.rpc_max_payload)
 	}
 
+	fn ws_max_out_buffer_capacity(&self) -> Result<Option<usize>> {
+		Ok(self.ws_max_out_buffer_capacity)
+	}
+
 	fn transaction_pool(&self) -> Result<TransactionPoolOptions> {
 		Ok(self.pool_config.transaction_pool())
 	}

substrate/client/cli/src/config.rs

@@ -360,6 +360,11 @@ pub trait CliConfiguration<DCV: DefaultConfigurationValues = ()>: Sized {
 		Ok(None)
 	}
 
+	/// Get maximum WS output buffer capacity.
+	fn ws_max_out_buffer_capacity(&self) -> Result<Option<usize>> {
+		Ok(None)
+	}
+
 	/// Get the prometheus configuration (`None` if disabled)
 	///
 	/// By default this is `None`.
@@ -513,6 +518,7 @@ pub trait CliConfiguration<DCV: DefaultConfigurationValues = ()>: Sized {
 			rpc_ws_max_connections: self.rpc_ws_max_connections()?,
 			rpc_cors: self.rpc_cors(is_dev)?,
 			rpc_max_payload: self.rpc_max_payload()?,
+			ws_max_out_buffer_capacity: self.ws_max_out_buffer_capacity()?,
 			prometheus_config: self.prometheus_config(DCV::prometheus_listen_port())?,
 			telemetry_endpoints,
 			default_heap_pages: self.default_heap_pages()?,

substrate/client/rpc-servers/src/lib.rs

@@ -33,6 +33,9 @@ const MEGABYTE: usize = 1024 * 1024;
 /// Maximal payload accepted by RPC servers.
 pub const RPC_MAX_PAYLOAD_DEFAULT: usize = 15 * MEGABYTE;
 
+/// Maximal buffer size in WS server.
+pub const WS_MAX_BUFFER_CAPACITY_DEFAULT: usize = 16 * MEGABYTE;
+
 /// Default maximum number of connections for WS RPC servers.
 const WS_MAX_CONNECTIONS: usize = 100;
 
@@ -172,18 +175,32 @@ pub fn start_ws<
 	cors: Option<&Vec<String>>,
 	io: RpcHandler<M>,
 	maybe_max_payload_mb: Option<usize>,
+	maybe_max_out_buffer_capacity_mb: Option<usize>,
 	server_metrics: ServerMetrics,
 	tokio_handle: tokio::runtime::Handle,
 ) -> io::Result<ws::Server> {
-	let rpc_max_payload = maybe_max_payload_mb
+	let max_payload = maybe_max_payload_mb
 		.map(|mb| mb.saturating_mul(MEGABYTE))
 		.unwrap_or(RPC_MAX_PAYLOAD_DEFAULT);
+	let max_out_buffer_capacity = maybe_max_out_buffer_capacity_mb
+		.map(|mb| mb.saturating_mul(MEGABYTE))
+		.unwrap_or(WS_MAX_BUFFER_CAPACITY_DEFAULT);
+
+	if max_payload > max_out_buffer_capacity {
+		log::warn!(
+			"maximum payload ({}) is more than maximum output buffer ({}) size in ws server, the payload will actually be limited by the buffer size",
+			max_payload,
+			max_out_buffer_capacity,
+		)
+	}
+
 	ws::ServerBuilder::with_meta_extractor(io, |context: &ws::RequestContext| {
 		context.sender().into()
 	})
 	.event_loop_executor(tokio_handle)
-	.max_payload(rpc_max_payload)
+	.max_payload(max_payload)
 	.max_connections(max_connections.unwrap_or(WS_MAX_CONNECTIONS))
+	.max_out_buffer_capacity(max_out_buffer_capacity)
 	.allowed_origins(map_cors(cors))
 	.allowed_hosts(hosts_filtering(cors.is_some()))
 	.session_stats(server_metrics)

substrate/client/service/src/config.rs

@@ -97,6 +97,8 @@ pub struct Configuration {
 	pub rpc_methods: RpcMethods,
 	/// Maximum payload of rpc request/responses.
 	pub rpc_max_payload: Option<usize>,
+	/// Maximum size of the output buffer capacity for websocket connections.
+	pub ws_max_out_buffer_capacity: Option<usize>,
 	/// Prometheus endpoint configuration. `None` if disabled.
 	pub prometheus_config: Option<PrometheusConfig>,
 	/// Telemetry service URL. `None` if disabled.

substrate/client/service/src/lib.rs

@@ -424,6 +424,7 @@ fn start_rpc_servers<
 					),
 				)?,
 				config.rpc_max_payload,
+				config.ws_max_out_buffer_capacity,
 				server_metrics.clone(),
 				config.tokio_handle.clone(),
 			)

substrate/client/service/test/src/lib.rs

@@ -261,6 +261,7 @@ fn node_config<
 		rpc_cors: None,
 		rpc_methods: Default::default(),
 		rpc_max_payload: None,
+		ws_max_out_buffer_capacity: None,
 		prometheus_config: None,
 		telemetry_endpoints: None,
 		default_heap_pages: None,

substrate/test-utils/test-runner/src/utils.rs

@@ -94,6 +94,7 @@ pub fn default_config(tokio_handle: Handle, mut chain_spec: Box<dyn ChainSpec>)
 		rpc_cors: None,
 		rpc_methods: Default::default(),
 		rpc_max_payload: None,
+		ws_max_out_buffer_capacity: None,
 		prometheus_config: None,
 		telemetry_endpoints: None,
 		default_heap_pages: None,