//! 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<Position<Self>>,
through: Option<Position<Self>>,
limit: usize,
) -> Result<ReplayChunk<Self::Output, Position<Self>>, ReplayFailed>;
/// the output for a live broadcast, or none when there's nothing to send.
fn render(&mut self, event: Self::Broadcast) -> Option<Self::Output>;
/// 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<S> = <<S as StreamSource>::Broadcast as StreamBroadcast>::Position;
/// where a subscriber begins.
pub(crate) struct Start<P> {
/// the last position it has seen, none for before the first event
after: Option<P>,
/// replay the db through this position before tailing
catch_up: Option<P>,
}
impl<P: Ord + Copy> Start<P> {
/// tail live broadcasts after `head`.
pub(crate) fn live(head: Option<P>) -> Self {
Self {
after: head,
catch_up: None,
}
}
/// replay the rows after `after` through `head`, then tail.
pub(crate) fn replay(after: Option<P>, head: Option<P>) -> Self {
Self {
after,
catch_up: head.filter(|head| after < Some(*head)),
}
}
}
enum CatchUp<P> {
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<S, E>(
tx: mpsc::Sender<Result<S::Output, E>>,
event_rx: broadcast::Receiver<S::Broadcast>,
start: Start<Position<S>>,
opts: StreamOptions,
mut source: S,
) where
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);
// 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<S, E>(
out: &mut Output<S::Output, E>,
source: &mut S,
live: &mut Live<S::Broadcast>,
after: &mut Option<Position<S>>,
target: CatchUp<Position<S>>,
opts: StreamOptions,
mut pacer: Option<Pacer>,
) -> Result<(), Ended>
where
S: StreamSource,
E: From<StreamTooSlow> + fmt::Display,
{
let through = match target {
CatchUp::Through(position) => Some(position),
CatchUp::ToEnd => None,
};
let mut blocked_since: Option<Instant> = 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<O, E, B>(
out: &mut Output<O, E>,
output: O,
opts: StreamOptions,
live: &mut Live<B>,
) -> Result<(), Ended>
where
B: Clone,
E: From<StreamTooSlow> + 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<F: Future>(fut: F, deadline: Instant) -> Option<F::Output> {
struct Unpark(std::thread::Thread);
impl Wake for Unpark {
fn wake(self: Arc<Self>) {
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<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.
struct Live<B> {
rx: broadcast::Receiver<B>,
backlog: VecDeque<B>,
limit: usize,
/// broadcasts after the backlog were lost
lagged: bool,
}
enum Next<B> {
Event(B),
Lagged,
/// the broadcast channel closed, or the subscriber hung up
Closed,
}
impl<B: Clone> Live<B> {
fn new(rx: broadcast::Receiver<B>, 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<T>, 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<T>(&mut self, tx: &mpsc::Sender<T>) -> Next<B> {
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<T>(
tx: &mpsc::Sender<T>,
configured: Option<NonZeroUsize>,
) -> Option<usize> {
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<BTreeSet<u64>>,
tx: Mutex<Option<broadcast::Sender<Ev>>>,
next: Mutex<u64>,
reads: Mutex<usize>,
reads_fail: Mutex<bool>,
}
impl Log {
fn new(capacity: usize) -> (Arc<Self>, broadcast::Receiver<Ev>) {
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<u64> {
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<u64> {
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<dyn FnMut(&Log, usize) + Send>;
struct Source {
log: Arc<Log>,
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<u64>,
through: Option<u64>,
limit: usize,
) -> Result<ReplayChunk<u64, u64>, 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<u64> {
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<Result<u64, StreamTooSlow>>;
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<Log>,
event_rx: broadcast::Receiver<Ev>,
start: Start<u64>,
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<u64> {
(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::<Vec<_>>());
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<Log>,
event_rx: broadcast::Receiver<Ev>,
start: Start<u64>,
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<u64> {
(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> {
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));
}
}