diff --git a/src/control/indexer.rs b/src/control/indexer.rs
index ebd963a..406958b 100644
--- a/src/control/indexer.rs
+++ b/src/control/indexer.rs
@@ -99,7 +99,8 @@ impl Hydrant {
     /// the stream ends when the `EventStream` is dropped. a subscriber that falls behind
     /// the live broadcast catches up from the database, skipping only the ephemeral
     /// events it fell behind on. one that stops reading for `HYDRANT_STREAM_SEND_TIMEOUT`
-    /// receives [`StreamError::ConsumerTooSlow`] before the stream terminates when possible.
+    /// receives [`StreamError::ConsumerTooSlow`] after the events already queued, then the
+    /// stream ends.
     pub fn subscribe(&self, cursor: Option<u64>) -> EventStream {
         let (tx, rx) = mpsc::channel(stream::STREAM_CHANNEL_CAPACITY);
         let state = self.state.clone();
diff --git a/src/control/relay.rs b/src/control/relay.rs
index 6c0eb68..ccb5b20 100644
--- a/src/control/relay.rs
+++ b/src/control/relay.rs
@@ -39,7 +39,8 @@ impl Hydrant {
     ///
     /// a subscriber that falls behind the live broadcast catches up from the database. one
     /// that stops reading for `HYDRANT_STREAM_SEND_TIMEOUT` receives
-    /// [`RelayStreamError::ConsumerTooSlow`] before the stream terminates when possible.
+    /// [`RelayStreamError::ConsumerTooSlow`] after the frames already queued, then the
+    /// stream ends.
     pub fn subscribe_repos(&self, cursor: Option<u64>) -> RelayEventStream {
         let (tx, rx) = mpsc::channel(stream::STREAM_CHANNEL_CAPACITY);
         let state = self.state.clone();
diff --git a/src/control/stream/engine.rs b/src/control/stream/engine.rs
index d84e268..69b67af 100644
--- a/src/control/stream/engine.rs
+++ b/src/control/stream/engine.rs
@@ -11,6 +11,7 @@ use std::fmt;
 use std::num::NonZeroUsize;
 use std::time::{Duration, Instant};
 use tokio::sync::broadcast::error::{RecvError, TryRecvError};
+use tokio::sync::mpsc::OwnedPermit;
 use tokio::sync::mpsc::error::TrySendError;
 use tokio::sync::{broadcast, mpsc};
 use tracing::{debug, warn};
@@ -104,17 +105,18 @@ pub(crate) fn run_stream<S, E>(
     S: StreamSource,
     E: From<StreamTooSlow> + fmt::Display,
 {
+    let mut out = Output::new(tx);
     let mut live = Live::new(event_rx, opts.pending_event_limit);
     let mut after = start.after;
     let mut catch_up = start.catch_up.map(CatchUp::Through);
 
     loop {
-        if tx.is_closed() {
+        if out.tx.is_closed() {
             return;
         }
 
         if let Some(target) = catch_up.take() {
-            if replay(&tx, &mut source, &mut live, &mut after, target, opts).is_err() {
+            if replay(&mut out, &mut source, &mut live, &mut after, target, opts).is_err() {
                 return;
             }
             continue;
@@ -143,7 +145,7 @@ pub(crate) fn run_stream<S, E>(
             continue;
         }
         if let Some(output) = source.render(event)
-            && send(&tx, output, opts, &mut live).is_err()
+            && send(&mut out, output, opts, &mut live).is_err()
         {
             return;
         }
@@ -153,7 +155,7 @@ pub(crate) fn run_stream<S, E>(
 
 /// send the db's rows after `after` until `target`, moving `after` along.
 fn replay<S, E>(
-    tx: &mpsc::Sender<Result<S::Output, E>>,
+    out: &mut Output<S::Output, E>,
     source: &mut S,
     live: &mut Live<S::Broadcast>,
     after: &mut Option<Position<S>>,
@@ -172,11 +174,10 @@ where
 
     loop {
         // skip the db read while the output channel is saturated
-        let Some(limit) = replay_chunk_size_for(tx, opts.replay_chunk_size) else {
+        let Some(limit) = replay_chunk_size_for(&out.tx, opts.replay_chunk_size) else {
             live.drain();
             if let Err(err) = note_replay_blocked(&mut blocked_since, opts.send_timeout) {
-                send_stream_error(tx, err.into());
-                return Err(Ended);
+                return Err(out.close(err.into()));
             }
             std::thread::sleep(STREAM_SEND_RETRY_PAUSE);
             continue;
@@ -187,7 +188,7 @@ where
             .read(*after, through, limit)
             .map_err(|ReplayFailed| Ended)?;
         for output in chunk.events {
-            send(tx, output, opts, live)?;
+            send(out, output, opts, live)?;
         }
         *after = (*after).max(chunk.last_seen);
 
@@ -205,7 +206,7 @@ where
 
 /// send one output, waiting out a full channel for up to the send timeout.
 fn send<O, E, B>(
-    tx: &mpsc::Sender<Result<O, E>>,
+    out: &mut Output<O, E>,
     output: O,
     opts: StreamOptions,
     live: &mut Live<B>,
@@ -217,15 +218,14 @@ where
     let mut item = Ok(output);
     let started = Instant::now();
     loop {
-        match tx.try_send(item) {
+        match out.tx.try_send(item) {
             Ok(()) => return Ok(()),
             Err(TrySendError::Closed(_)) => return Err(Ended),
             Err(TrySendError::Full(returned)) => {
                 item = returned;
                 live.drain();
                 if started.elapsed() >= opts.send_timeout {
-                    send_stream_error(tx, StreamTooSlow::send_timeout(opts.send_timeout).into());
-                    return Err(Ended);
+                    return Err(out.close(StreamTooSlow::send_timeout(opts.send_timeout).into()));
                 }
                 std::thread::sleep(STREAM_SEND_RETRY_PAUSE);
             }
@@ -233,6 +233,29 @@ where
     }
 }
 
+/// the subscriber's channel, with one slot held from the start so a closing
+/// error always fits behind the outputs already queued.
+struct Output<O, E> {
+    tx: mpsc::Sender<Result<O, E>>,
+    closing: Option<OwnedPermit<Result<O, E>>>,
+}
+
+impl<O, E: fmt::Display> Output<O, E> {
+    fn new(tx: mpsc::Sender<Result<O, E>>) -> Self {
+        let closing = tx.clone().try_reserve_owned().ok();
+        Self { tx, closing }
+    }
+
+    /// queue `err` as the last item, ending the stream.
+    fn close(&mut self, err: E) -> Ended {
+        warn!(%err, "closing stream subscriber");
+        if let Some(closing) = self.closing.take() {
+            closing.send(Err(err));
+        }
+        Ended
+    }
+}
+
 /// the broadcasts a subscriber hasn't taken yet: the broadcast channel, plus
 /// a bounded backlog drained from it while a send waits, so a briefly slow
 /// subscriber doesn't lag the channel.
@@ -287,14 +310,6 @@ impl<B: Clone> Live<B> {
     }
 }
 
-fn send_stream_error<O, E>(tx: &mpsc::Sender<Result<O, E>>, err: E)
-where
-    E: fmt::Display,
-{
-    warn!(%err, "closing stream subscriber");
-    let _ = tx.try_send(Err(err));
-}
-
 fn replay_chunk_size_for<T>(
     tx: &mpsc::Sender<T>,
     configured: Option<NonZeroUsize>,
@@ -511,6 +526,8 @@ mod tests {
         }
     }
 
+    /// run a subscriber whose channel holds `out_capacity` outputs, plus the
+    /// slot the engine keeps for a closing error.
     fn spawn(
         log: &Arc<Log>,
         event_rx: broadcast::Receiver<Ev>,
@@ -519,7 +536,7 @@ mod tests {
         opts: StreamOptions,
         on_read: impl FnMut(&Log, usize) + Send + 'static,
     ) -> (JoinHandle<()>, Out) {
-        let (tx, out) = mpsc::channel(out_capacity);
+        let (tx, out) = mpsc::channel(out_capacity + 1);
         let source = Source {
             log: log.clone(),
             wants_markers: true,
@@ -729,7 +746,9 @@ mod tests {
         );
         handle.join().unwrap();
 
+        // the error comes after what was queued, even though that filled the channel
         assert_eq!(out.try_recv().unwrap().unwrap(), rows[0]);
+        assert!(matches!(out.try_recv(), Ok(Err(StreamTooSlow { .. }))));
         assert!(out.try_recv().is_err());
     }