From a79ca5e8fcc211b6183bac3d2c3d4f52b106d07f Mon Sep 17 00:00:00 2001
From: cloudhead <cloudhead@radicle.xyz>
Date: Wed, 26 Jun 2024 13:36:00 +0200
Subject: [PATCH] node: Constrain worker channel size

To avoid buffering large amounts of data in the process, we set the
worker channel size to `64`. Keep in mind that this is one
`ChannelEvent`, not one byte. `ChannelEvent::Data` can already contain
an arbitrary amount of data via its `Vec<u8>`, but in practice the
maximum is 8192, due to the use of `io::copy`.

We also remove an unused function, and move the flushing to the
`ChannelFlushWriter`.
---
 radicle-node/src/worker/channels.rs    | 24 +++++-------------------
 radicle-node/src/worker/upload_pack.rs |  1 -
 2 files changed, 5 insertions(+), 20 deletions(-)

diff --git a/radicle-node/src/worker/channels.rs b/radicle-node/src/worker/channels.rs
index 1b189abf..298408ec 100644
--- a/radicle-node/src/worker/channels.rs
+++ b/radicle-node/src/worker/channels.rs
@@ -10,7 +10,9 @@ use crate::runtime::Handle;
 use crate::wire::StreamId;
 
 /// Maximum size of channel used to communicate with a worker.
-pub const MAX_WORKER_CHANNEL_SIZE: usize = 4096;
+/// Note that as long as we're using [`std::io::copy`] to copy data from the
+/// upload-pack's stdout, the data chunks are of a maximum size of 8192 bytes.
+pub const MAX_WORKER_CHANNEL_SIZE: usize = 64;
 
 /// A reader and writer pair that can be used in the fetch protocol.
 ///
@@ -184,7 +186,7 @@ struct ChannelWriter<T = Vec<u8>> {
 /// Wraps a [`ChannelWriter`] alongside the associated [`Handle`] and [`NodeId`].
 ///
 /// This allows the channel to [`Write::flush`] when calling
-/// [`Write::write_all`], which is necessary to signal to the
+/// [`Write::write`], which is necessary to signal to the
 /// controller to send the wire data.
 pub struct ChannelFlushWriter<T = Vec<u8>> {
     writer: ChannelWriter<T>,
@@ -206,29 +208,13 @@ impl Write for ChannelFlushWriter<Vec<u8>> {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         let n = buf.len();
         self.writer.send(buf.to_vec())?;
+        self.flush()?;
         Ok(n)
     }
 
     fn flush(&mut self) -> io::Result<()> {
         self.handle.flush(self.remote, self.stream)
     }
-
-    fn write_all(&mut self, mut buf: &[u8]) -> io::Result<()> {
-        while !buf.is_empty() {
-            match self.write(buf) {
-                Ok(0) => {
-                    return Err(io::Error::new(
-                        io::ErrorKind::WriteZero,
-                        "failed to write whole buffer",
-                    ));
-                }
-                Ok(n) => buf = &buf[n..],
-                Err(ref e) if e.kind() == io::ErrorKind::Interrupted => {}
-                Err(e) => return Err(e),
-            }
-        }
-        self.flush()
-    }
 }
 
 impl<T: AsRef<[u8]>> ChannelWriter<T> {
diff --git a/radicle-node/src/worker/upload_pack.rs b/radicle-node/src/worker/upload_pack.rs
index 5a4ce3ca..a73ab93d 100644
--- a/radicle-node/src/worker/upload_pack.rs
+++ b/radicle-node/src/worker/upload_pack.rs
@@ -201,7 +201,6 @@ where
 {
     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
         let n = self.send.write(buf)?;
-        self.send.flush()?;
         self.emit(buf);
         Ok(n)
     }