persist/pubsub: set unlimited decoding message size everywhere (#35934)

This PR sets the `max_decoding_message_size` for all pubsub servers and
clients to `usize::MAX`, to avoid pubsub errors and resulting reconnects
caused by participants not being able to decode received messages.

There was already code in place to do this for a subset of the pubsub
servers created (introduced in
#21034), but it was
(likely due to oversight) not applied to all of them, nor to clients.

Note that we don't need to set `max_encoding_message_size`, as it
already defaults to `usize::MAX`.

### Motivation

We observed the decoding size limit to cause pubsub reconnects, which in
turn tickles a connection leak. This change doesn't fix the leak yet but
makes it less likely that we hit it.

Author: teskje

src/persist-client/src/rpc.rs

@@ -135,6 +135,12 @@ pub(crate) const PUBSUB_RECONNECT_BACKOFF: Config<Duration> = Config::new(
     "Backoff after an established connection to Persist PubSub service fails.",
 );
 
+/// Max message size, used to configure gRPC servers and clients.
+///
+/// While `max_encoding_message_size` defaults to `usize::MAX`, `max_decoding_message_size` only
+/// defaults to 4MB, so we bump it to avoid protocol errors.
+const MAX_GRPC_MESSAGE_SIZE: usize = usize::MAX;
+
 /// Top-level Trait to create a PubSubClient.
 ///
 /// Returns a [PubSubClientConnection] with a [PubSubSender] for issuing RPCs to the PubSub
@@ -324,7 +330,7 @@ impl GrpcPubSubClient {
                 .await;
 
             let mut client = match client {
-                Ok(client) => client,
+                Ok(client) => client.max_decoding_message_size(MAX_GRPC_MESSAGE_SIZE),
                 Err(err) => {
                     error!("fatal error connecting to persist pubsub: {:?}", err);
                     return;
@@ -1064,7 +1070,10 @@ impl PersistGrpcPubSubServer {
     pub async fn serve(self, listen_addr: SocketAddr) -> Result<(), anyhow::Error> {
         // Increase the default message decoding limit to avoid unnecessary panics
         tonic::transport::Server::builder()
-            .add_service(ProtoPersistPubSubServer::new(self).max_decoding_message_size(usize::MAX))
+            .add_service(
+                ProtoPersistPubSubServer::new(self)
+                    .max_decoding_message_size(MAX_GRPC_MESSAGE_SIZE),
+            )
             .serve(listen_addr)
             .await?;
         Ok(())
@@ -1077,7 +1086,10 @@ impl PersistGrpcPubSubServer {
         listener: tokio_stream::wrappers::TcpListenerStream,
     ) -> Result<(), anyhow::Error> {
         tonic::transport::Server::builder()
-            .add_service(ProtoPersistPubSubServer::new(self))
+            .add_service(
+                ProtoPersistPubSubServer::new(self)
+                    .max_decoding_message_size(MAX_GRPC_MESSAGE_SIZE),
+            )
             .serve_with_incoming(listener)
             .await?;
         Ok(())