persist/pubsub: fix connection leak on gRPC stream error (#35938)

When the client's gRPC response stream errors, the reconnect loop drops
the tonic `Channel` but hyper's background task keeps running: It's
blocked polling the `broadcast_messages` async stream, which holds a
live `BroadcastStream` receiver. This keeps the HTTP2 connection open,
leaking one connection per reconnect.

Fix this by giving the stream a cancellation token. When the reconnect
loop drops `cancel_tx`, the stream observes it via `select!` and
terminates, allowing hyper to close the HTTP2 connection.

### Motivation

Fixes https://github.com/MaterializeInc/database-issues/issues/11276

Author: teskje

src/persist-client/src/rpc.rs

@@ -353,10 +353,18 @@ impl GrpcPubSubClient {
                 .broadcast_recv_lagged_count
                 .clone();
 
+            // `client.pub_sub(...)` starts a hyper background task reading from the
+            // `broadcast_messages` stream, to serve the HTTP2 connection. The broadcast stream
+            // doesn't normally terminate, which means the HTTP2 connection doesn't terminate
+            // either. We set up a cancelation token to force termination and avoid a connection
+            // leak.
+            let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<()>();
+
             // shard subscriptions are tracked by connection on the server, so if our
             // gRPC stream is ever swapped out, we must inform the server which shards
             // our client intended to be subscribed to.
             let broadcast_messages = async_stream::stream! {
+                let mut cancel_rx = std::pin::pin!(cancel_rx);
                 'reconnect: loop {
                     // If we have active subscriptions, resend them.
                     for id in sender.subscriptions() {
@@ -370,18 +378,28 @@ impl GrpcPubSubClient {
                     }
 
                     // Forward on messages from the broadcast channel, reconnecting if necessary.
-                    while let Some(message) = broadcast.next().await {
-                        debug!("sending pubsub message: {:?}", message);
-                        match message {
-                            Ok(message) => yield message,
-                            Err(BroadcastStreamRecvError::Lagged(i)) => {
-                                broadcast_errors.inc_by(i);
-                                continue 'reconnect;
+                    loop {
+                        tokio::select! {
+                            message = broadcast.next() => {
+                                debug!("sending pubsub message: {:?}", message);
+                                match message {
+                                    Some(Ok(message)) => yield message,
+                                    Some(Err(BroadcastStreamRecvError::Lagged(i))) => {
+                                        broadcast_errors.inc_by(i);
+                                        continue 'reconnect;
+                                    }
+                                    None => {
+                                        debug!("exhausted pubsub broadcast stream; shutting down");
+                                        return;
+                                    }
+                                }
+                            }
+                            _ = &mut cancel_rx => {
+                                debug!("pubsub broadcast stream cancelled; shutting down");
+                                return;
                             }
                         }
                     }
-                    debug!("exhausted pubsub broadcast stream; shutting down");
-                    break;
                 }
             };
             let pubsub_request =
@@ -403,6 +421,8 @@ impl GrpcPubSubClient {
             )
             .await;
 
+            drop(cancel_tx);
+
             match stream_completed {
                 // common case: reconnect due to some transient error
                 Ok(_) => continue,