ctp: make channels unbounded

This commit makes the channels used by the CTP implementation to connect
the send and recv tasks with the `Connection` into unbounded channels.

It was nice to have them bounded for backpressure, but unfortunately
this leads to a deadlock on the server side. Specifically, if the
connection is configured without a timeout and the send task isn't able
to send messages for whatever reason, its channel can fill up. This in
turn can indefinitely block the server's connection task. Reconnecting
doesn't help in this case, as canceling the connection task requires
that task's cooperation (for cancel safety reasons).

Having unbounded channels introduces an opportunity for unbounded memory
growth, but that seems to be the least of all evils. We use unbounded in
a bunch of places already and so far we've not had many issues because
of that, which is reassuring.

The obvious alternative would be to always configure a send timeout.
This is tricky because of self-managed though: We can't rely on users
always choosing reasonable timeouts and cluster connection issues are
notoriously hard to debug, so we want to make cluster connections as
foolproof as possible. We could set a maximum timeout of say, one
minute, but even then a misconfigured timeout would cause cluster
reconnects to take very long, degrading UX for non-obvious reasons.

Author: teskje

src/service/src/transport.rs

@@ -243,9 +243,9 @@ where
 #[derive(Debug)]
 struct Connection<Out, In> {
     /// Message sender connected to the send task.
-    msg_tx: mpsc::Sender<Out>,
+    msg_tx: mpsc::UnboundedSender<Out>,
     /// Message receiver connected to the receive task.
-    msg_rx: mpsc::Receiver<In>,
+    msg_rx: mpsc::UnboundedReceiver<In>,
     /// Receiver for errors encountered by connection tasks.
     error_rx: watch::Receiver<String>,
 
@@ -289,8 +289,8 @@ impl<Out: Message, In: Message> Connection<Out, In> {
 
         handshake(&mut reader, &mut writer, version, server_fqdn).await?;
 
-        let (out_tx, out_rx) = mpsc::channel(1024);
-        let (in_tx, in_rx) = mpsc::channel(1024);
+        let (out_tx, out_rx) = mpsc::unbounded_channel();
+        let (in_tx, in_rx) = mpsc::unbounded_channel();
         // Initialize the error channel with a default error to return if none of the tasks
         // produced an error.
         let (error_tx, error_rx) = watch::channel("connection closed".into());
@@ -314,7 +314,7 @@ impl<Out: Message, In: Message> Connection<Out, In> {
 
     /// Enqueue a message for sending.
     async fn send(&mut self, msg: Out) -> anyhow::Result<()> {
-        match self.msg_tx.send(msg).await {
+        match self.msg_tx.send(msg) {
             Ok(()) => Ok(()),
             Err(_) => bail!(self.collect_error().await),
         }
@@ -347,7 +347,7 @@ impl<Out: Message, In: Message> Connection<Out, In> {
     /// Run a connection's send task.
     async fn run_send_task<W: AsyncWrite + Unpin>(
         mut writer: W,
-        mut msg_rx: mpsc::Receiver<Out>,
+        mut msg_rx: mpsc::UnboundedReceiver<Out>,
         error_tx: watch::Sender<String>,
         mut metrics: impl Metrics<Out, In>,
     ) {
@@ -383,7 +383,7 @@ impl<Out: Message, In: Message> Connection<Out, In> {
     /// Run a connection's recv task.
     async fn run_recv_task<R: AsyncRead + Unpin>(
         mut reader: R,
-        msg_tx: mpsc::Sender<In>,
+        msg_tx: mpsc::UnboundedSender<In>,
         error_tx: watch::Sender<String>,
         mut metrics: impl Metrics<Out, In>,
     ) {
@@ -393,7 +393,7 @@ impl<Out: Message, In: Message> Connection<Out, In> {
                     trace!(?msg, "ctp: received message");
                     metrics.message_received(&msg);
 
-                    if msg_tx.send(msg).await.is_err() {
+                    if msg_tx.send(msg).is_err() {
                         break;
                     }
                 }