diff --git a/Cargo.lock b/Cargo.lock
index dd88c84c..981c3d3b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1,6 +1,6 @@
 # This file is automatically @generated by Cargo.
 # It is not intended for manual editing.
-version = 3
+version = 4
 
 [[package]]
 name = "adler2"
@@ -308,6 +308,15 @@ version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
 
+[[package]]
+name = "bytesize"
+version = "2.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a3c8f83209414aacf0eeae3cf730b18d6981697fba62f200fcfb92b9f082acba"
+dependencies = [
+ "serde",
+]
+
 [[package]]
 name = "cbc"
 version = "0.1.2"
@@ -2173,6 +2182,7 @@ version = "0.14.0"
 dependencies = [
  "amplify",
  "base64 0.21.7",
+ "bytesize",
  "chrono",
  "colored",
  "crossbeam-channel",
diff --git a/radicle-cli/examples/rad-config.md b/radicle-cli/examples/rad-config.md
index e26fd17d..7e4dc39c 100644
--- a/radicle-cli/examples/rad-config.md
+++ b/radicle-cli/examples/rad-config.md
@@ -47,7 +47,8 @@ $ rad config
       "connection": {
         "inbound": 128,
         "outbound": 16
-      }
+      },
+      "fetchPackReceive": "500.0 MiB"
     },
     "workers": 8,
     "seedingPolicy": {
diff --git a/radicle-node/src/service.rs b/radicle-node/src/service.rs
index dfb8b874..db08f00d 100644
--- a/radicle-node/src/service.rs
+++ b/radicle-node/src/service.rs
@@ -1098,7 +1098,13 @@ where
             refs_at: refs_at.clone(),
             subscribers: vec![],
         });
-        self.outbox.fetch(session, rid, refs_at, timeout);
+        self.outbox.fetch(
+            session,
+            rid,
+            refs_at,
+            timeout,
+            self.config.limits.fetch_pack_receive,
+        );
 
         Ok(fetching)
     }
diff --git a/radicle-node/src/service/io.rs b/radicle-node/src/service/io.rs
index 5232f922..09b05fbb 100644
--- a/radicle-node/src/service/io.rs
+++ b/radicle-node/src/service/io.rs
@@ -2,6 +2,7 @@ use std::collections::VecDeque;
 use std::time;
 
 use log::*;
+use radicle::node::config::FetchPackSizeLimit;
 use radicle::storage::refs::RefsAt;
 
 use crate::prelude::*;
@@ -30,6 +31,8 @@ pub enum Io {
         refs_at: Option<Vec<RefsAt>>,
         /// Fetch timeout.
         timeout: time::Duration,
+        /// Limit the number of bytes fetched.
+        reader_limit: FetchPackSizeLimit,
     },
     /// Ask for a wakeup in a specified amount of time.
     Wakeup(LocalDuration),
@@ -124,6 +127,7 @@ impl Outbox {
         rid: RepoId,
         refs_at: Vec<RefsAt>,
         timeout: time::Duration,
+        reader_limit: FetchPackSizeLimit,
     ) {
         peer.fetching(rid);
 
@@ -143,6 +147,7 @@ impl Outbox {
             refs_at,
             remote: peer.id,
             timeout,
+            reader_limit,
         });
     }
 
diff --git a/radicle-node/src/tests/e2e.rs b/radicle-node/src/tests/e2e.rs
index 8337b53a..ffa83c2a 100644
--- a/radicle-node/src/tests/e2e.rs
+++ b/radicle-node/src/tests/e2e.rs
@@ -1493,3 +1493,44 @@ fn test_multiple_offline_inits() {
         assert!(projects.contains(&repo.rid), "Bob is missing {}", repo.rid);
     }
 }
+
+#[test]
+fn test_channel_reader_limit() {
+    logger::init(log::Level::Debug);
+
+    let tmp = tempfile::tempdir().unwrap();
+    let mut alice = Node::init(tmp.path(), config::relay("alice"));
+    let limits = radicle::node::config::Limits {
+        fetch_pack_receive: radicle::node::config::FetchPackSizeLimit::bytes(1000),
+        ..radicle::node::config::Limits::default()
+    };
+    let bob = Node::init(
+        tmp.path(),
+        Config {
+            limits,
+            ..config::relay("bob")
+        },
+    );
+    let acme = alice.project("acme", "");
+
+    let mut alice = alice.spawn();
+    let mut bob = bob.spawn();
+
+    alice.connect(&bob);
+    converge([&alice, &bob]);
+
+    let updated = bob.handle.seed(acme, Scope::All).unwrap();
+    assert!(updated);
+
+    let result = bob.handle.fetch(acme, alice.id, DEFAULT_TIMEOUT).unwrap();
+    assert!(!result.is_success());
+
+    let FetchResult::Failed { reason } = result else {
+        panic!("fetch result must be failed")
+    };
+    assert!(
+        reason.contains("Failed to consume the pack sent by the remote"),
+        "actual: {}",
+        reason
+    );
+}
diff --git a/radicle-node/src/wire/protocol.rs b/radicle-node/src/wire/protocol.rs
index b4f903f2..a34a1ba2 100644
--- a/radicle-node/src/wire/protocol.rs
+++ b/radicle-node/src/wire/protocol.rs
@@ -35,7 +35,7 @@ use crate::wire::frame;
 use crate::wire::frame::{Frame, FrameData, StreamId};
 use crate::wire::Encode;
 use crate::worker;
-use crate::worker::{ChannelEvent, FetchRequest, FetchResult, Task, TaskResult};
+use crate::worker::{ChannelEvent, ChannelsConfig, FetchRequest, FetchResult, Task, TaskResult};
 use crate::Link;
 
 /// NoiseXK handshake pattern.
@@ -128,22 +128,22 @@ impl Streams {
     }
 
     /// Open a new stream.
-    fn open(&mut self, timeout: time::Duration) -> (StreamId, worker::Channels) {
+    fn open(&mut self, config: ChannelsConfig) -> (StreamId, worker::Channels) {
         self.seq += 1;
 
         let id = StreamId::git(self.link)
             .nth(self.seq)
             .expect("Streams::open: too many streams");
         let channels = self
-            .register(id, timeout)
+            .register(id, config)
             .expect("Streams::open: stream was already open");
 
         (id, channels)
     }
 
     /// Register an open stream.
-    fn register(&mut self, stream: StreamId, timeout: time::Duration) -> Option<worker::Channels> {
-        let (wire, worker) = worker::Channels::pair(timeout)
+    fn register(&mut self, stream: StreamId, config: ChannelsConfig) -> Option<worker::Channels> {
+        let (wire, worker) = worker::Channels::pair(config)
             .expect("Streams::register: fatal: unable to create channels");
 
         match self.streams.entry(stream) {
@@ -762,8 +762,12 @@ where
                                 log::debug!(target: "wire", "Received `open` command for stream {stream} from {nid}");
                                 metrics.streams_opened += 1;
                                 metrics.received_fetch_requests += 1;
-
-                                let Some(channels) = streams.register(stream, FETCH_TIMEOUT) else {
+                                let reader_limit = self.service.config().limits.fetch_pack_receive;
+                                let Some(channels) = streams.register(
+                                    stream,
+                                    ChannelsConfig::new(FETCH_TIMEOUT)
+                                        .with_reader_limit(reader_limit),
+                                ) else {
                                     log::warn!(target: "wire", "Peer attempted to open already-open stream stream {stream}");
                                     continue;
                                 };
@@ -1062,6 +1066,7 @@ where
                     rid,
                     remote,
                     timeout,
+                    reader_limit,
                     refs_at,
                 } => {
                     log::trace!(target: "wire", "Processing fetch for {rid} from {remote}..");
@@ -1076,7 +1081,8 @@ where
                         log::error!(target: "wire", "Peer {remote} is not connected: dropping fetch");
                         continue;
                     };
-                    let (stream, channels) = streams.open(timeout);
+                    let (stream, channels) =
+                        streams.open(ChannelsConfig::new(timeout).with_reader_limit(reader_limit));
 
                     log::debug!(target: "wire", "Opened new stream with id {stream} for {rid} and remote {remote}");
 
diff --git a/radicle-node/src/worker.rs b/radicle-node/src/worker.rs
index bc5277f3..eb5b0a29 100644
--- a/radicle-node/src/worker.rs
+++ b/radicle-node/src/worker.rs
@@ -23,7 +23,7 @@ use crate::service::policy;
 use crate::service::policy::SeedingPolicy;
 use crate::wire::StreamId;
 
-pub use channels::{ChannelEvent, Channels};
+pub use channels::{ChannelEvent, Channels, ChannelsConfig};
 
 /// Worker pool configuration.
 pub struct Config {
diff --git a/radicle-node/src/worker/channels.rs b/radicle-node/src/worker/channels.rs
index 298408ec..a5e721bd 100644
--- a/radicle-node/src/worker/channels.rs
+++ b/radicle-node/src/worker/channels.rs
@@ -4,6 +4,7 @@ use std::ops::Deref;
 use std::{fmt, io, time};
 
 use crossbeam_channel as chan;
+use radicle::node::config::FetchPackSizeLimit;
 use radicle::node::NodeId;
 
 use crate::runtime::Handle;
@@ -14,6 +15,32 @@ use crate::wire::StreamId;
 /// upload-pack's stdout, the data chunks are of a maximum size of 8192 bytes.
 pub const MAX_WORKER_CHANNEL_SIZE: usize = 64;
 
+#[derive(Clone, Copy, Debug)]
+pub struct ChannelsConfig {
+    timeout: time::Duration,
+    reader_limit: FetchPackSizeLimit,
+}
+
+impl ChannelsConfig {
+    pub fn new(timeout: time::Duration) -> Self {
+        Self {
+            timeout,
+            reader_limit: FetchPackSizeLimit::default(),
+        }
+    }
+
+    pub fn with_timeout(self, timeout: time::Duration) -> Self {
+        Self { timeout, ..self }
+    }
+
+    pub fn with_reader_limit(self, reader_limit: FetchPackSizeLimit) -> Self {
+        Self {
+            reader_limit,
+            ..self
+        }
+    }
+}
+
 /// A reader and writer pair that can be used in the fetch protocol.
 ///
 /// It implements [`radicle::fetch::transport::ConnectionStream`] to
@@ -92,20 +119,23 @@ impl<T: AsRef<[u8]>> Channels<T> {
     pub fn new(
         sender: chan::Sender<ChannelEvent<T>>,
         receiver: chan::Receiver<ChannelEvent<T>>,
-        timeout: time::Duration,
+        config: ChannelsConfig,
     ) -> Self {
-        let sender = ChannelWriter { sender, timeout };
-        let receiver = ChannelReader::new(receiver, timeout);
+        let sender = ChannelWriter {
+            sender,
+            timeout: config.timeout,
+        };
+        let receiver = ChannelReader::new(receiver, config.timeout, config.reader_limit);
 
         Self { sender, receiver }
     }
 
-    pub fn pair(timeout: time::Duration) -> io::Result<(Channels<T>, Channels<T>)> {
+    pub fn pair(config: ChannelsConfig) -> io::Result<(Channels<T>, Channels<T>)> {
         let (l_send, r_recv) = chan::bounded::<ChannelEvent<T>>(MAX_WORKER_CHANNEL_SIZE);
         let (r_send, l_recv) = chan::bounded::<ChannelEvent<T>>(MAX_WORKER_CHANNEL_SIZE);
 
-        let l = Channels::new(l_send, l_recv, timeout);
-        let r = Channels::new(r_send, r_recv, timeout);
+        let l = Channels::new(l_send, l_recv, config);
+        let r = Channels::new(r_send, r_recv, config);
 
         Ok((l, r))
     }
@@ -123,12 +153,41 @@ impl<T: AsRef<[u8]>> Channels<T> {
     }
 }
 
+#[derive(Clone, Copy, Debug)]
+pub struct ReadLimiter {
+    limit: FetchPackSizeLimit,
+    total_read: usize,
+}
+
+impl ReadLimiter {
+    pub fn new(limit: FetchPackSizeLimit) -> Self {
+        Self {
+            limit,
+            total_read: 0,
+        }
+    }
+
+    pub fn read(&mut self, bytes: usize) -> io::Result<()> {
+        self.total_read = self.total_read.saturating_add(bytes);
+        log::trace!(target: "worker", "limit {}, total bytes read: {}", self.limit, self.total_read);
+        if self.limit.exceeded_by(self.total_read) {
+            Err(io::Error::new(
+                io::ErrorKind::Other,
+                "sender has exceeded number of allowed bytes, aborting read",
+            ))
+        } else {
+            Ok(())
+        }
+    }
+}
+
 /// Wraps a [`chan::Receiver`] and provides it with [`io::Read`].
 #[derive(Clone)]
 pub struct ChannelReader<T = Vec<u8>> {
     buffer: io::Cursor<Vec<u8>>,
     receiver: chan::Receiver<ChannelEvent<T>>,
     timeout: time::Duration,
+    limiter: ReadLimiter,
 }
 
 impl<T> Deref for ChannelReader<T> {
@@ -140,11 +199,16 @@ impl<T> Deref for ChannelReader<T> {
 }
 
 impl<T: AsRef<[u8]>> ChannelReader<T> {
-    pub fn new(receiver: chan::Receiver<ChannelEvent<T>>, timeout: time::Duration) -> Self {
+    pub fn new(
+        receiver: chan::Receiver<ChannelEvent<T>>,
+        timeout: time::Duration,
+        limit: FetchPackSizeLimit,
+    ) -> Self {
         Self {
             buffer: io::Cursor::new(Vec::new()),
             receiver,
             timeout,
+            limiter: ReadLimiter::new(limit),
         }
     }
 }
@@ -152,6 +216,7 @@ impl<T: AsRef<[u8]>> ChannelReader<T> {
 impl Read for ChannelReader<Vec<u8>> {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         let read = self.buffer.read(buf)?;
+        self.limiter.read(read)?;
         if read > 0 {
             return Ok(read);
         }
diff --git a/radicle/Cargo.toml b/radicle/Cargo.toml
index 74853a1b..9282579b 100644
--- a/radicle/Cargo.toml
+++ b/radicle/Cargo.toml
@@ -16,6 +16,7 @@ logger = ["colored", "chrono"]
 [dependencies]
 amplify = { version = "4.0.0", default-features = false, features = ["std"] }
 base64 = { version = "0.21.3" }
+bytesize = { version = "2", features = ["serde"] }
 crossbeam-channel = { version = "0.5.6" }
 cyphernet = { version = "0.5.0", features = ["tor", "dns", "p2p-ed25519"] }
 fastrand = { version = "2.0.0" }
diff --git a/radicle/src/node/config.rs b/radicle/src/node/config.rs
index fdbe4476..15571df2 100644
--- a/radicle/src/node/config.rs
+++ b/radicle/src/node/config.rs
@@ -1,6 +1,7 @@
 use std::collections::HashSet;
-use std::net;
 use std::ops::Deref;
+use std::str::FromStr;
+use std::{fmt, net};
 
 use cyphernet::addr::PeerAddr;
 use localtime::LocalDuration;
@@ -114,6 +115,9 @@ pub struct Limits {
     /// Connection limits.
     #[serde(default)]
     pub connection: ConnectionLimits,
+    /// Channel limits.
+    #[serde(default)]
+    pub fetch_pack_receive: FetchPackSizeLimit,
 }
 
 impl Default for Limits {
@@ -126,10 +130,88 @@ impl Default for Limits {
             max_open_files: 4096,
             rate: RateLimits::default(),
             connection: ConnectionLimits::default(),
+            fetch_pack_receive: FetchPackSizeLimit::default(),
         }
     }
 }
 
+/// Limiter for byte streams.
+///
+/// Default: 500MiB
+#[derive(Debug, Clone, Copy, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+#[serde(into = "String", try_from = "String")]
+pub struct FetchPackSizeLimit {
+    limit: bytesize::ByteSize,
+}
+
+impl From<bytesize::ByteSize> for FetchPackSizeLimit {
+    fn from(limit: bytesize::ByteSize) -> Self {
+        Self { limit }
+    }
+}
+
+impl From<FetchPackSizeLimit> for String {
+    fn from(limit: FetchPackSizeLimit) -> Self {
+        limit.to_string()
+    }
+}
+
+impl TryFrom<String> for FetchPackSizeLimit {
+    type Error = String;
+
+    fn try_from(s: String) -> Result<Self, Self::Error> {
+        s.parse()
+    }
+}
+
+impl FromStr for FetchPackSizeLimit {
+    type Err = String;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        Ok(FetchPackSizeLimit { limit: s.parse()? })
+    }
+}
+
+impl fmt::Display for FetchPackSizeLimit {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.limit)
+    }
+}
+
+impl FetchPackSizeLimit {
+    /// New `FetchPackSizeLimit` in bytes.
+    pub fn bytes(size: u64) -> Self {
+        bytesize::ByteSize::b(size).into()
+    }
+
+    /// New `FetchPackSizeLimit` in kibibytes.
+    pub fn kibibytes(size: u64) -> Self {
+        bytesize::ByteSize::kib(size).into()
+    }
+
+    /// New `FetchPackSizeLimit` in mebibytes.
+    pub fn mebibytes(size: u64) -> Self {
+        bytesize::ByteSize::mib(size).into()
+    }
+
+    /// New `FetchPackSizeLimit` in gibibytes.
+    pub fn gibibytes(size: u64) -> Self {
+        bytesize::ByteSize::gib(size).into()
+    }
+
+    /// Check if this limit is exceeded by the number of `bytes` provided.
+    pub fn exceeded_by(&self, bytes: usize) -> bool {
+        bytes >= self.limit.as_u64() as usize
+    }
+}
+
+impl Default for FetchPackSizeLimit {
+    fn default() -> Self {
+        Self::mebibytes(500)
+    }
+}
+
 /// Connection limits.
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 #[serde(rename_all = "camelCase")]