//! the replay-then-tail loop every event stream runs. //! //! positions come from `db::sequencer`, so broadcasts arrive in position //! order, and every committed row is either broadcast or covered by a later //! persisted marker. a subscriber only has to take broadcasts in the order //! they arrive, reading the db for a marker's rows, and one that falls behind //! the broadcast channel can always resume from the db at its position. use futures::future::{Either, select}; use std::collections::VecDeque; use std::fmt; use std::num::NonZeroUsize; use std::sync::Arc; use std::task::{Context, Poll, Wake, Waker}; use std::time::Instant; use tokio::sync::broadcast::error::{RecvError, TryRecvError}; use tokio::sync::mpsc::error::TrySendError; use tokio::sync::mpsc::{OwnedPermit, Permit}; use tokio::sync::{broadcast, mpsc}; use tracing::{debug, warn}; use super::replay_cap::{Pacer, Read}; use super::types::{ReplayChunk, ReplayFailed, StreamOptions, StreamTooSlow}; /// an event broadcast after its commit. pub(crate) trait StreamBroadcast: Clone { /// where the event sits in commit order. type Position: Ord + Copy; fn position(&self) -> Self::Position; /// a persisted marker carries no event: every row through its position is /// committed, and the ones since the previous broadcast weren't broadcast. fn is_marker(&self) -> bool; } /// what one stream endpoint reads and sends. pub(crate) trait StreamSource { type Broadcast: StreamBroadcast; type Output; /// up to `limit` outputs for the committed rows after `after`, through /// `through` when it is set. fn read( &mut self, after: Option>, through: Option>, limit: usize, ) -> Result>, ReplayFailed>; /// the output for a live broadcast, or none when there's nothing to send. fn render(&mut self, event: Self::Broadcast) -> Option; /// how many positions the stream has taken, which a capped replay reads /// as the live rate. an atomic load, so the commit path pays nothing. fn written(&self) -> u64; /// about how many bytes `output` takes on the wire, which a capped replay /// pays for. fn frame_bytes(output: &Self::Output) -> usize; /// whether a marker's rows are worth reading. a subscriber that filters /// them all out can skip the read. fn wants_marker(&self, _marker: &Self::Broadcast) -> bool { true } } pub(crate) type Position = <::Broadcast as StreamBroadcast>::Position; /// where a subscriber begins. pub(crate) struct Start
{ /// the last position it has seen, none for before the first event after: Option
, /// replay the db through this position before tailing catch_up: Option
, } impl Start
{ /// tail live broadcasts after `head`. pub(crate) fn live(head: Option
) -> Self { Self { after: head, catch_up: None, } } /// replay the rows after `after` through `head`, then tail. pub(crate) fn replay(after: Option
, head: Option
) -> Self { Self { after, catch_up: head.filter(|head| after < Some(*head)), } } } enum CatchUp
{ Through(P), ToEnd, } /// the stream is over: the subscriber hung up, was sent a closing error, or /// the db couldn't be read. struct Ended; /// subscribe `event_rx` before reading the head `start` was built from, so /// nothing committed in between is missed. pub(crate) fn run_stream( tx: mpsc::Sender>, event_rx: broadcast::Receiver, start: Start>, opts: StreamOptions, mut source: S, ) where 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); // only the replay the cursor asked for is capped. markers and lag catch ups are as big as // what got written meanwhile, so they're live work, and capping them would lag the tail let mut pacer = catch_up .as_ref() .and(opts.replay_cap) .map(|cap| Pacer::new(cap, source.written(), Instant::now())); loop { if out.tx.is_closed() { return; } if let Some(target) = catch_up.take() { let replayed = replay( &mut out, &mut source, &mut live, &mut after, target, opts, pacer.take(), ); if replayed.is_err() { return; } continue; } let event = match live.next(&out.tx) { Next::Event(event) => event, Next::Lagged => { debug!("stream subscriber fell behind the broadcast, catching up from the db"); catch_up = Some(CatchUp::ToEnd); continue; } Next::Closed => return, }; let position = event.position(); if after.is_some_and(|after| position <= after) { continue; } if event.is_marker() { if source.wants_marker(&event) { catch_up = Some(CatchUp::Through(position)); } else { after = Some(position); } continue; } if let Some(output) = source.render(event) && send(&mut out, output, opts, &mut live).is_err() { return; } after = Some(position); } } /// send the db's rows after `after` until `target`, moving `after` along. fn replay( out: &mut Output, source: &mut S, live: &mut Live, after: &mut Option>, target: CatchUp>, opts: StreamOptions, mut pacer: Option, ) -> Result<(), Ended> where S: StreamSource, E: From + fmt::Display, { let through = match target { CatchUp::Through(position) => Some(position), CatchUp::ToEnd => None, }; let mut blocked_since: Option = None; loop { // skip the db read while the output channel is saturated let Some(limit) = replay_chunk_size_for(&out.tx, opts.replay_chunk_size) else { let deadline = *blocked_since.get_or_insert_with(Instant::now) + opts.send_timeout; match live.wait_for_room(&out.tx, deadline) { Room::Ready(_) => continue, Room::Closed => return Err(Ended), Room::TimedOut => {} } return Err(out.close(StreamTooSlow::send_timeout(opts.send_timeout).into())); }; blocked_since = None; let chunk = source .read(*after, through, limit) .map_err(|ReplayFailed| Ended)?; let mut read = Read { scanned: chunk.scanned, sent: chunk.events.len(), sent_bytes: 0, }; for output in chunk.events { read.sent_bytes += S::frame_bytes(&output); send(out, output, opts, live)?; } *after = (*after).max(chunk.last_seen); if chunk.exhausted { // every row through the target is committed, so none are left *after = (*after).max(through); return Ok(()); } if let Some(pacer) = &mut pacer { let wait = pacer.charge(read, source.written(), Instant::now()); // a reader that hangs up mid wait ends the thread right away if !wait.is_zero() && block_on_until(out.tx.closed(), Instant::now() + wait).is_some() { return Err(Ended); } } } } /// send one output, waiting out a full channel for up to the send timeout. fn send( out: &mut Output, output: O, opts: StreamOptions, live: &mut Live, ) -> Result<(), Ended> where B: Clone, E: From + fmt::Display, { match live.wait_for_room(&out.tx, Instant::now() + opts.send_timeout) { Room::Ready(permit) => { permit.send(Ok(output)); return Ok(()); } Room::Closed => return Err(Ended), Room::TimedOut => {} } Err(out.close(StreamTooSlow::send_timeout(opts.send_timeout).into())) } /// what waiting for a slot in the subscriber's channel came to enum Room<'a, T> { Ready(Permit<'a, T>), /// the subscriber hung up Closed, /// still full at the deadline TimedOut, } /// runs `fut` on this thread until it finishes or `deadline` passes. the thread sleeps until /// something wakes the future, so a wait costs nothing however long it is fn block_on_until(fut: F, deadline: Instant) -> Option { struct Unpark(std::thread::Thread); impl Wake for Unpark { fn wake(self: Arc) { self.0.unpark(); } } let waker = Waker::from(Arc::new(Unpark(std::thread::current()))); let mut cx = Context::from_waker(&waker); let mut fut = std::pin::pin!(fut); loop { if let Poll::Ready(output) = fut.as_mut().poll(&mut cx) { return Some(output); } let left = deadline.saturating_duration_since(Instant::now()); if left.is_zero() { return None; } std::thread::park_timeout(left); } } /// 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. struct Live { rx: broadcast::Receiver, backlog: VecDeque, limit: usize, /// broadcasts after the backlog were lost lagged: bool, } enum Next { Event(B), Lagged, /// the broadcast channel closed, or the subscriber hung up Closed, } impl Live { fn new(rx: broadcast::Receiver, limit: usize) -> Self { Self { rx, backlog: VecDeque::new(), limit, lagged: false, } } /// move waiting broadcasts into the backlog, up to its limit. past it the /// channel holds them, until it lags. fn drain(&mut self) { while !self.lagged && self.backlog.len() < self.limit { match self.rx.try_recv() { Ok(event) => self.backlog.push_back(event), Err(TryRecvError::Lagged(_)) => self.lagged = true, Err(TryRecvError::Empty | TryRecvError::Closed) => return, } } } /// waits for a slot in `tx` until `deadline`, taking broadcasts into the backlog meanwhile /// so a subscriber blocked for a moment doesn't lag the channel fn wait_for_room<'a, T>(&mut self, tx: &'a mpsc::Sender, deadline: Instant) -> Room<'a, T> { match tx.try_reserve() { Ok(permit) => return Room::Ready(permit), Err(TrySendError::Closed(())) => return Room::Closed, Err(TrySendError::Full(())) => {} } loop { self.drain(); let taking = !self.lagged && self.backlog.len() < self.limit; let rx = &mut self.rx; let woke = block_on_until( async { tokio::select! { biased; permit = tx.reserve() => Ok(permit), event = rx.recv(), if taking => Err(event), } }, deadline, ); match woke { None => return Room::TimedOut, Some(Ok(Ok(permit))) => return Room::Ready(permit), Some(Ok(Err(_))) => return Room::Closed, Some(Err(Ok(event))) => self.backlog.push_back(event), Some(Err(Err(RecvError::Lagged(_)))) => self.lagged = true, // a closed broadcast ends the stream once the backlog is sent Some(Err(Err(RecvError::Closed))) => { return match block_on_until(tx.reserve(), deadline) { None => Room::TimedOut, Some(Ok(permit)) => Room::Ready(permit), Some(Err(_)) => Room::Closed, }; } } } } /// the next broadcast, waiting for one unless the subscriber on `tx` /// hangs up first. fn next(&mut self, tx: &mpsc::Sender) -> Next { if let Some(event) = self.backlog.pop_front() { return Next::Event(event); } if std::mem::take(&mut self.lagged) { return Next::Lagged; } let recv = std::pin::pin!(self.rx.recv()); let hung_up = std::pin::pin!(tx.closed()); match futures::executor::block_on(select(recv, hung_up)) { Either::Left((Ok(event), _)) => Next::Event(event), Either::Left((Err(RecvError::Lagged(_)), _)) => Next::Lagged, Either::Left((Err(RecvError::Closed), _)) | Either::Right(_) => Next::Closed, } } } fn replay_chunk_size_for( tx: &mpsc::Sender, configured: Option, ) -> Option { let available = tx.capacity(); if available == 0 { return None; } let max_cap = tx.max_capacity().max(1); let desired = configured .map(NonZeroUsize::get) .unwrap_or_else(|| (max_cap / 2).max(1)) .min(max_cap); Some(desired.min(available).max(1)) } #[cfg(all(test, any(feature = "indexer_stream", feature = "relay")))] mod tests { use super::*; use crate::control::stream::replay_cap::ReplayCap; use std::collections::BTreeSet; use std::num::NonZeroU64; use std::sync::Mutex; use std::thread::JoinHandle; use std::time::Duration; #[derive(Clone, Debug)] enum Ev { /// a committed row, broadcast live Row(u64), /// a broadcast that has no row Ephemeral(u64), Marker(u64), } impl StreamBroadcast for Ev { type Position = u64; fn position(&self) -> u64 { match self { Self::Row(id) | Self::Ephemeral(id) | Self::Marker(id) => *id, } } fn is_marker(&self) -> bool { matches!(self, Self::Marker(_)) } } /// the db and broadcast channel the way the sequencer drives them: /// commit, then broadcast, in position order. struct Log { db: Mutex>, tx: Mutex>>, next: Mutex, reads: Mutex, reads_fail: Mutex, } impl Log { fn new(capacity: usize) -> (Arc, broadcast::Receiver) { let (tx, rx) = broadcast::channel(capacity); let log = Self { db: Mutex::new(BTreeSet::new()), tx: Mutex::new(Some(tx)), next: Mutex::new(1), reads: Mutex::new(0), reads_fail: Mutex::new(false), }; (Arc::new(log), rx) } fn take(&self) -> u64 { let mut next = self.next.lock().unwrap(); *next += 1; *next - 1 } fn broadcast(&self, ev: Ev) { if let Some(tx) = &*self.tx.lock().unwrap() { let _ = tx.send(ev); } } fn live(&self) -> u64 { let id = self.take(); self.db.lock().unwrap().insert(id); self.broadcast(Ev::Row(id)); id } fn ephemeral(&self) -> u64 { let id = self.take(); self.broadcast(Ev::Ephemeral(id)); id } /// rows written without a broadcast, like a backfill fn backfill(&self, n: usize) -> Vec { let ids: Vec<_> = (0..n).map(|_| self.take()).collect(); self.db.lock().unwrap().extend(&ids); self.broadcast(Ev::Marker(*ids.last().unwrap())); ids } fn head(&self) -> Option { self.next .lock() .unwrap() .checked_sub(1) .filter(|id| *id > 0) } fn close(&self) { self.tx.lock().unwrap().take(); } fn fail_reads(&self) { *self.reads_fail.lock().unwrap() = true; } /// wait until the subscriber has taken every broadcast off the channel fn wait_taken(&self) { let deadline = Instant::now() + Duration::from_secs(5); while self .tx .lock() .unwrap() .as_ref() .is_some_and(|tx| !tx.is_empty()) { assert!( Instant::now() < deadline, "the subscriber never took a broadcast" ); std::thread::sleep(Duration::from_millis(1)); } } } /// runs before each db read, to commit at exact points of a replay type OnRead = Box; struct Source { log: Arc, wants_markers: bool, /// the rows the subscriber wants, the rest are scanned and filtered out sends: fn(u64) -> bool, on_read: OnRead, } impl StreamSource for Source { type Broadcast = Ev; type Output = u64; fn read( &mut self, after: Option, through: Option, limit: usize, ) -> Result, ReplayFailed> { let call = { let mut reads = self.log.reads.lock().unwrap(); *reads += 1; *reads - 1 }; (self.on_read)(&self.log, call); if *self.log.reads_fail.lock().unwrap() { return Err(ReplayFailed); } let start = after.map_or(0, |after| after + 1); let end = through.unwrap_or(u64::MAX); let db = self.log.db.lock().unwrap(); let mut rows = db.range(start..=end).copied().peekable(); let mut chunk = ReplayChunk { events: Vec::new(), last_seen: after, exhausted: false, scanned: 0, }; // up to four rows scanned per output, like the real reads while chunk.events.len() < limit && chunk.scanned < limit * 4 { let Some(row) = rows.next() else { break }; chunk.scanned += 1; chunk.last_seen = Some(row); chunk .events .extend(Some(row).filter(|row| (self.sends)(*row))); } chunk.exhausted = rows.peek().is_none(); Ok(chunk) } fn written(&self) -> u64 { *self.log.next.lock().unwrap() } fn frame_bytes(_: &u64) -> usize { FRAME_BYTES } fn render(&mut self, event: Ev) -> Option { match event { Ev::Row(id) | Ev::Ephemeral(id) => Some(id), Ev::Marker(_) => unreachable!("markers are handled by the engine"), } } fn wants_marker(&self, _marker: &Ev) -> bool { self.wants_markers } } type Out = mpsc::Receiver>; const FRAME_BYTES: usize = 100; fn opts() -> StreamOptions { StreamOptions { replay_chunk_size: NonZeroUsize::new(2), replay_cap: None, pending_event_limit: 4, send_timeout: Duration::from_secs(5), } } /// 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, start: Start, out_capacity: usize, opts: StreamOptions, on_read: impl FnMut(&Log, usize) + Send + 'static, ) -> (JoinHandle<()>, Out) { let (tx, out) = mpsc::channel(out_capacity + 1); let source = Source { log: log.clone(), wants_markers: true, sends: |_| true, on_read: Box::new(on_read), }; let handle = std::thread::spawn(move || run_stream(tx, event_rx, start, opts, source)); (handle, out) } fn take(out: &mut Out, n: usize) -> Vec { (0..n) .map(|_| { let deadline = Instant::now() + Duration::from_secs(5); loop { match out.try_recv() { Ok(item) => break item.expect("stream error"), Err(mpsc::error::TryRecvError::Empty) if Instant::now() < deadline => { std::thread::sleep(Duration::from_millis(5)); } Err(err) => panic!("no output: {err}"), } } }) .collect() } fn finish(log: &Log, handle: JoinHandle<()>, mut out: Out) { log.close(); handle.join().unwrap(); assert!(out.try_recv().is_err(), "unexpected extra output"); } /// wait for the stream thread to end by itself fn ends(handle: JoinHandle<()>) { let deadline = Instant::now() + Duration::from_secs(5); while !handle.is_finished() { assert!(Instant::now() < deadline, "the stream never ended"); std::thread::sleep(Duration::from_millis(5)); } handle.join().unwrap(); } fn nothing_on_read(_: &Log, _: usize) {} #[test] fn replays_in_chunks_then_tails() { let (log, rx) = Log::new(16); let rows = log.backfill(5); let (handle, mut out) = spawn( &log, rx, Start::replay(None, log.head()), 16, opts(), nothing_on_read, ); assert_eq!(take(&mut out, 5), rows); let live = log.live(); assert_eq!(take(&mut out, 1), [live]); finish(&log, handle, out); } #[test] fn a_live_start_skips_everything_through_its_head() { let (log, rx) = Log::new(16); // already broadcast, so they wait in the subscriber's channel log.backfill(3); log.live(); let (handle, mut out) = spawn(&log, rx, Start::live(log.head()), 16, opts(), |_, _| { panic!("rows through the head need no db read"); }); let live = log.live(); assert_eq!(take(&mut out, 1), [live]); finish(&log, handle, out); } #[test] fn a_replay_start_reads_only_past_its_cursor() { let (log, rx) = Log::new(16); let rows = log.backfill(4); let (handle, mut out) = spawn( &log, rx, Start::replay(Some(rows[1]), log.head()), 16, opts(), nothing_on_read, ); assert_eq!(take(&mut out, 2), rows[2..]); let live = log.live(); assert_eq!(take(&mut out, 1), [live]); finish(&log, handle, out); let (log, rx) = Log::new(16); log.backfill(2); let (handle, mut out) = spawn( &log, rx, Start::replay(log.head(), log.head()), 16, opts(), |_, _| panic!("a cursor at the head has nothing to replay"), ); let live = log.live(); assert_eq!(take(&mut out, 1), [live]); finish(&log, handle, out); } #[test] fn a_marker_reads_its_rows_from_the_db() { let (log, rx) = Log::new(16); let (handle, mut out) = spawn( &log, rx, Start::live(log.head()), 16, opts(), nothing_on_read, ); let first = log.live(); let rows = log.backfill(3); let last = log.live(); let expected: Vec<_> = [first].into_iter().chain(rows).chain([last]).collect(); assert_eq!(take(&mut out, 5), expected); finish(&log, handle, out); } #[test] fn broadcast_only_events_keep_their_place() { let (log, rx) = Log::new(16); let (handle, mut out) = spawn( &log, rx, Start::live(log.head()), 16, opts(), nothing_on_read, ); let rows = log.backfill(2); let ephemeral = log.ephemeral(); let more = log.backfill(2); let expected: Vec<_> = rows.into_iter().chain([ephemeral]).chain(more).collect(); assert_eq!(take(&mut out, 5), expected); finish(&log, handle, out); } #[test] fn commits_during_a_replay_are_sent_once() { let (log, rx) = Log::new(16); let rows = log.backfill(3); let (handle, mut out) = spawn( &log, rx, Start::replay(None, log.head()), 16, opts(), |log, call| { if call == 0 { // lands past the head the replay started with, broadcast // before the replay reaches it log.live(); log.backfill(1); } }, ); let sent = take(&mut out, 5); assert_eq!(sent[..3], rows[..]); assert_eq!(sent, (rows[0]..rows[0] + 5).collect::>()); finish(&log, handle, out); } #[test] fn a_lagging_subscriber_catches_up_from_the_db() { let (log, rx) = Log::new(2); let opts = StreamOptions { pending_event_limit: 1, ..opts() }; let (handle, mut out) = spawn(&log, rx, Start::live(log.head()), 1, opts, nothing_on_read); let first = log.live(); std::thread::sleep(Duration::from_millis(50)); // the subscriber is stuck sending while these overflow the channel // and its backlog let rest: Vec<_> = (0..8).map(|_| log.live()).collect(); let expected: Vec<_> = [first].into_iter().chain(rest).collect(); assert_eq!(take(&mut out, 9), expected); let live = log.live(); assert_eq!(take(&mut out, 1), [live]); finish(&log, handle, out); } #[test] fn the_backlog_holds_broadcasts_while_a_send_waits() { let (log, rx) = Log::new(2); let (handle, mut out) = spawn(&log, rx, Start::live(log.head()), 1, opts(), |_, _| { panic!("a backlog within its limit needs no db read"); }); // fill the output, then leave the subscriber waiting to send the next let mut sent = vec![log.live()]; log.wait_taken(); sent.push(log.ephemeral()); log.wait_taken(); // more than the broadcast channel holds, taken into the backlog while // that send waits. they have no rows, so a lag would lose them for _ in 0..3 { sent.push(log.ephemeral()); log.wait_taken(); } assert_eq!(take(&mut out, 5), sent); finish(&log, handle, out); } #[test] fn unwanted_markers_are_skipped_without_a_read() { let (log, rx) = Log::new(16); let (tx, mut out) = mpsc::channel(16); let source = Source { log: log.clone(), wants_markers: false, sends: |_| true, on_read: Box::new(|_, _| panic!("an unwanted marker needs no db read")), }; let handle = std::thread::spawn(move || run_stream(tx, rx, Start::live(None), opts(), source)); log.backfill(3); let live = log.live(); assert_eq!(take(&mut out, 1), [live]); finish(&log, handle, out); } #[test] fn a_stuck_subscriber_is_closed() { let (log, rx) = Log::new(16); let rows = log.backfill(2); let opts = StreamOptions { send_timeout: Duration::ZERO, ..opts() }; let (handle, mut out) = spawn( &log, rx, Start::replay(None, log.head()), 1, opts, nothing_on_read, ); 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()); } #[test] fn a_hung_up_subscriber_ends_without_a_broadcast() { let (log, rx) = Log::new(16); let (handle, out) = spawn( &log, rx, Start::live(log.head()), 16, opts(), nothing_on_read, ); // most likely waiting for a broadcast by now std::thread::sleep(Duration::from_millis(50)); drop(out); ends(handle); } #[test] fn a_failed_read_ends_the_stream() { let (log, rx) = Log::new(16); let rows = log.backfill(5); let (handle, mut out) = spawn( &log, rx, Start::replay(None, log.head()), 16, opts(), |log, call| { if call == 1 { log.fail_reads(); } }, ); // rather than skip the rows it couldn't read and go on tailing assert_eq!(take(&mut out, 2), rows[..2]); ends(handle); log.live(); assert!(out.try_recv().is_err()); } /// a subscriber that reads as fast as it can never waits on a timer: the thread sends again /// the moment there's room, so many times the channel's size goes through in no time #[test] fn a_fast_subscriber_is_never_held_back_by_a_full_channel() { const N: usize = 20_000; const CAPACITY: usize = 16; let (log, rx) = Log::new(16); let rows = log.backfill(N); let opts = StreamOptions { replay_chunk_size: NonZeroUsize::new(4), ..opts() }; let started = Instant::now(); let (handle, mut out) = spawn( &log, rx, Start::replay(None, log.head()), CAPACITY, opts, nothing_on_read, ); let sent: Vec<_> = (0..N) .map(|_| out.blocking_recv().unwrap().unwrap()) .collect(); let took = started.elapsed(); assert_eq!(sent, rows); // sleeping 10 ms whenever the channel filled took N / CAPACITY x 10 ms, 12.5 s here assert!(took < Duration::from_secs(2), "{took:?}"); finish(&log, handle, out); } /// whether it fills up mid chunk or between chunks, a subscriber that stops reading is /// closed once it has been full for the send timeout, not before #[test] fn a_subscriber_that_stops_reading_is_closed_after_the_send_timeout() { for chunk in [1, 2] { let (log, rx) = Log::new(16); let rows = log.backfill(4); let opts = StreamOptions { replay_chunk_size: NonZeroUsize::new(chunk), send_timeout: Duration::from_millis(300), ..opts() }; let started = Instant::now(); let (handle, mut out) = spawn( &log, rx, Start::replay(None, log.head()), 1, opts, nothing_on_read, ); ends(handle); let took = started.elapsed(); assert!( took >= Duration::from_millis(300), "chunk {chunk}: {took:?}" ); assert_eq!(out.try_recv().unwrap().unwrap(), rows[0]); assert!(matches!(out.try_recv(), Ok(Err(StreamTooSlow { .. })))); assert!(out.try_recv().is_err()); } } /// a reader whose replay is capped at a floor of `floor` bytes a second, on a box whose /// head never moves fn capped(floor: u64) -> StreamOptions { StreamOptions { replay_chunk_size: NonZeroUsize::new(250), replay_cap: Some(ReplayCap::new(10, NonZeroU64::new(floor).unwrap())), ..opts() } } fn spawn_capped( log: &Arc, event_rx: broadcast::Receiver, start: Start, opts: StreamOptions, sends: fn(u64) -> bool, ) -> (JoinHandle<()>, Out) { let (tx, out) = mpsc::channel(1024); let source = Source { log: log.clone(), wants_markers: true, sends, on_read: Box::new(nothing_on_read), }; let handle = std::thread::spawn(move || run_stream(tx, event_rx, start, opts, source)); (handle, out) } fn receive(out: &mut Out, n: usize) -> Vec { (0..n) .map(|_| out.blocking_recv().unwrap().unwrap()) .collect() } #[test] fn a_cursor_replay_is_held_to_the_cap() { let (log, rx) = Log::new(16); let rows = log.backfill(3000); let started = Instant::now(); let start = Start::replay(None, log.head()); // 1000 rows a second let (handle, mut out) = spawn_capped(&log, rx, start, capped(100_000), |_| true); assert_eq!(receive(&mut out, 3000), rows); // a second banked, then the last chunk is read once 2750 rows are paid for let took = started.elapsed(); assert!(took >= Duration::from_millis(1500), "{took:?}"); finish(&log, handle, out); } #[test] fn a_filtered_replay_pays_for_the_rows_it_scans() { let (log, rx) = Log::new(16); let rows = log.backfill(4000); let wanted: Vec<_> = rows.into_iter().filter(|row| row % 50 == 0).collect(); let started = Instant::now(); let start = Start::replay(None, log.head()); let (handle, mut out) = spawn_capped(&log, rx, start, capped(32_000), |row| row % 50 == 0); assert_eq!(receive(&mut out, wanted.len()), wanted); // reads of 1000 rows for 20 sends cost 33 kb each, so the middle two wait a second, where // the 2 kb they send would never wait let took = started.elapsed(); assert!(took >= Duration::from_millis(1500), "{took:?}"); finish(&log, handle, out); } #[test] fn a_quiet_box_still_replays_at_the_floor() { let (log, rx) = Log::new(16); let rows = log.backfill(1500); let start = Start::replay(None, log.head()); let (handle, mut out) = spawn_capped(&log, rx, start, capped(100_000), |_| true); // the head never moves, so without the floor the replay would never get past its first second assert_eq!(take(&mut out, 1500), rows); finish(&log, handle, out); } #[test] fn a_capped_replay_leaves_the_live_tail_alone() { let (log, rx) = Log::new(16); let rows = log.backfill(2000); let start = Start::replay(None, log.head()); let (handle, mut out) = spawn_capped(&log, rx, start, capped(100_000), |_| true); assert_eq!(take(&mut out, 2000), rows); // a backfill's marker on the tail costs as much as a replay but isn't one, so it reads // its rows at once. even a fresh second of budget would leave 1500 of them to wait out let started = Instant::now(); let backfilled = log.backfill(2500); let live = log.live(); let expected: Vec<_> = backfilled.into_iter().chain([live]).collect(); assert_eq!(take(&mut out, 2501), expected); let took = started.elapsed(); assert!(took < Duration::from_millis(500), "{took:?}"); finish(&log, handle, out); } #[test] fn a_reader_that_hangs_up_mid_wait_ends_at_once() { let (log, rx) = Log::new(16); log.backfill(1000); // the first chunk leaves 200 rows owed at 50 a second, a 4 s wait let start = Start::replay(None, log.head()); let (handle, mut out) = spawn_capped(&log, rx, start, capped(5000), |_| true); take(&mut out, 250); drop(out); let started = Instant::now(); ends(handle); let took = started.elapsed(); assert!(took < Duration::from_secs(2), "{took:?}"); } fn chunk_size(n: usize) -> Option { NonZeroUsize::new(n) } #[test] fn replay_chunk_auto_uses_half_max_capacity() { let (tx, _rx) = mpsc::channel::<()>(500); assert_eq!(replay_chunk_size_for(&tx, None), Some(250)); } #[test] fn replay_chunk_manual_is_respected() { let (tx, _rx) = mpsc::channel::<()>(500); assert_eq!(replay_chunk_size_for(&tx, chunk_size(64)), Some(64)); } #[test] fn replay_chunk_manual_is_capped_to_max_capacity() { let (tx, _rx) = mpsc::channel::<()>(500); assert_eq!(replay_chunk_size_for(&tx, chunk_size(1000)), Some(500)); } #[test] fn replay_chunk_is_capped_to_current_available_capacity() { let (tx, mut rx) = mpsc::channel::<()>(4); tx.try_send(()).unwrap(); tx.try_send(()).unwrap(); tx.try_send(()).unwrap(); assert_eq!(replay_chunk_size_for(&tx, None), Some(1)); let _ = rx.try_recv(); } #[test] fn replay_chunk_none_when_channel_full() { let (tx, _rx) = mpsc::channel::<()>(2); tx.try_send(()).unwrap(); tx.try_send(()).unwrap(); assert_eq!(replay_chunk_size_for(&tx, None), None); } #[test] fn replay_chunk_manual_none_when_channel_full() { let (tx, _rx) = mpsc::channel::<()>(2); tx.try_send(()).unwrap(); tx.try_send(()).unwrap(); assert_eq!(replay_chunk_size_for(&tx, chunk_size(1)), None); } #[test] fn replay_chunk_auto_min_one_for_small_channel() { let (tx, _rx) = mpsc::channel::<()>(1); assert_eq!(replay_chunk_size_for(&tx, None), Some(1)); } }