From 5f4005e4149fc4fbd05091b408eab7e21f222e8d Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Tue, 29 Sep 2026 18:53:21 +0300 Subject: [PATCH] [stream] hold a channel slot for a stream's closing error a subscriber is closed when its channel stays full for the send timeout, and the ConsumerTooSlow error was then try_sent into that full channel, so it was nearly always dropped and the stream just ended. the engine now reserves one slot when the stream starts and sends the error through it, behind the outputs already queued. --- src/control/indexer.rs | 3 +- src/control/relay.rs | 3 +- src/control/stream/engine.rs | 61 +++++++++++++++++++++++------------- 3 files changed, 44 insertions(+), 23 deletions(-) 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) -> 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) -> 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: StreamSource, E: From + 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( 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( /// send the db's rows after `after` until `target`, moving `after` along. fn replay( - tx: &mpsc::Sender>, + out: &mut Output, source: &mut S, live: &mut Live, after: &mut Option>, @@ -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( - tx: &mpsc::Sender>, + out: &mut Output, output: O, opts: StreamOptions, live: &mut Live, @@ -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 { + tx: mpsc::Sender>, + closing: Option>>, +} + +impl Output { + fn new(tx: mpsc::Sender>) -> 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 Live { } } -fn send_stream_error(tx: &mpsc::Sender>, err: E) -where - E: fmt::Display, -{ - warn!(%err, "closing stream subscriber"); - let _ = tx.try_send(Err(err)); -} - fn replay_chunk_size_for( tx: &mpsc::Sender, configured: Option, @@ -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, event_rx: broadcast::Receiver, @@ -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()); } -- 2.51.2