use std::fmt; use std::num::NonZeroUsize; use std::time::Duration; use tracing::error; use crate::config::{Config, StreamKind}; use super::engine::StreamBroadcast; use super::replay_cap::ReplayCap; pub(crate) const STREAM_CHANNEL_CAPACITY: usize = 500; #[derive(Debug, Clone, Copy)] pub(crate) struct StreamOptions { pub(crate) replay_chunk_size: Option, /// none for in-process subscribers, which pay for their own replays pub(crate) replay_cap: Option, pub(crate) pending_event_limit: usize, pub(crate) send_timeout: Duration, } impl StreamOptions { pub(crate) fn from_config(config: &Config) -> Self { Self { replay_chunk_size: NonZeroUsize::new(config.stream_replay_chunk_size), replay_cap: None, pending_event_limit: config.stream_pending_event_limit.max(1), send_timeout: config.stream_send_timeout, } } /// for a subscriber on the network, whose cursor replay is capped when `stream` is /// throttled. pub(crate) fn network(config: &Config, stream: StreamKind) -> Self { Self { replay_cap: ReplayCap::from_config(config, stream), ..Self::from_config(config) } } } #[derive(Debug, Clone)] pub(crate) struct StreamTooSlow { pub(crate) reason: String, } impl StreamTooSlow { pub(crate) fn send_timeout(timeout: Duration) -> Self { Self { reason: format!( "stream delivery blocked for at least {} seconds", timeout.as_secs() ), } } } impl fmt::Display for StreamTooSlow { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { write!(f, "stream consumer too slow: {}", self.reason) } } #[cfg(feature = "indexer_stream")] impl From for super::super::StreamError { fn from(err: StreamTooSlow) -> Self { Self::ConsumerTooSlow { reason: err.reason } } } #[cfg(feature = "relay")] impl From for super::super::RelayStreamError { fn from(err: StreamTooSlow) -> Self { Self::ConsumerTooSlow { reason: err.reason } } } #[cfg(feature = "jetstream")] impl From for super::super::JetstreamStreamError { fn from(err: StreamTooSlow) -> Self { Self::ConsumerTooSlow { reason: err.reason } } } #[cfg(feature = "indexer_stream")] impl StreamBroadcast for crate::types::BroadcastEvent { type Position = u64; fn position(&self) -> u64 { match self { Self::Persisted(id) => *id, Self::LiveRecord(evt) => evt.id, Self::Ephemeral(evt) => evt.id, } } fn is_marker(&self) -> bool { matches!(self, Self::Persisted(_)) } } #[cfg(feature = "relay")] impl StreamBroadcast for crate::types::RelayBroadcast { type Position = u64; fn position(&self) -> u64 { match self { Self::Persisted(seq) | Self::Ephemeral(seq, _) => *seq, } } fn is_marker(&self) -> bool { matches!(self, Self::Persisted(_)) } } #[cfg(feature = "jetstream")] impl StreamBroadcast for crate::types::JetstreamBroadcast { type Position = crate::types::JetstreamPosition; fn position(&self) -> Self::Position { match self { Self::Historical(position) => *position, Self::Live(live) => live.position, } } fn is_marker(&self) -> bool { matches!(self, Self::Historical(_)) } } /// a db read failed partway. the rows past it can't be skipped, so the /// stream ends and its subscriber resumes from the last event it got. #[derive(Debug)] pub(crate) struct ReplayFailed; /// outputs for the rows one db read scanned. pub(crate) struct ReplayChunk { pub(crate) events: Vec, /// the last row scanned, sent or filtered out pub(crate) last_seen: Option

, /// no rows are left in the range read pub(crate) exhausted: bool, /// rows the read went through, sent or not, since a capped replay pays for dropped ones too pub(crate) scanned: usize, } impl ReplayChunk { /// read `rows` after `after` into up to `limit` outputs. scans at most /// four times as many rows, so a range the subscriber mostly filters out /// still returns promptly. rows whose key doesn't parse are skipped. pub(crate) fn read( mut rows: fjall::Iter, after: Option

, limit: usize, position: impl Fn(&[u8]) -> Option

, mut output: impl FnMut(P, &[u8]) -> Option, ) -> Result where P: Copy, { let mut chunk = Self { events: Vec::with_capacity(limit), last_seen: after, exhausted: false, scanned: 0, }; let max_scanned = limit.saturating_mul(4).max(limit); while chunk.events.len() < limit && chunk.scanned < max_scanned { let Some(row) = rows.next() else { chunk.exhausted = true; break; }; chunk.scanned += 1; let (key, value) = row.into_inner().map_err(|e| { error!(err = %e, "failed to read stream rows from db, closing the stream"); ReplayFailed })?; let Some(position) = position(&key) else { continue; }; chunk.last_seen = Some(position); chunk.events.extend(output(position, &value)); } Ok(chunk) } } #[cfg(test)] mod tests { use super::*; #[test] fn only_network_subscribers_are_capped() { let all = [ StreamKind::Indexer, StreamKind::Jetstream, StreamKind::Relay, ]; let config = Config { throttled_streams: all.to_vec(), ..Config::default() }; assert_eq!(StreamOptions::from_config(&config).replay_cap, None); for stream in all { assert!(StreamOptions::network(&config, stream).replay_cap.is_some()); } } /// keys 1 through 10 as big-endian positions, values naming them, plus one /// key that isn't a position between 3 and 4 fn rows() -> (tempfile::TempDir, fjall::Database, fjall::Keyspace) { let tmp = tempfile::tempdir().unwrap(); let db = fjall::Database::builder(tmp.path()).open().unwrap(); let ks = db .keyspace("rows", fjall::KeyspaceCreateOptions::default) .unwrap(); for n in 1..=10_u64 { ks.insert(n.to_be_bytes(), format!("v{n}")).unwrap(); } ks.insert([0, 0, 0, 0, 0, 0, 0, 3, 0], "junk").unwrap(); (tmp, db, ks) } fn position(key: &[u8]) -> Option { <[u8; 8]>::try_from(key).ok().map(u64::from_be_bytes) } fn evens(position: u64, value: &[u8]) -> Option<(u64, String)> { position .is_multiple_of(2) .then(|| (position, String::from_utf8(value.to_vec()).unwrap())) } #[test] fn a_chunk_stops_at_its_limit_and_remembers_filtered_rows() { let (_tmp, _db, ks) = rows(); let chunk = ReplayChunk::read(ks.iter(), None, 2, position, evens).unwrap(); assert_eq!(chunk.events, [(2, "v2".into()), (4, "v4".into())]); assert_eq!(chunk.last_seen, Some(4)); assert!(!chunk.exhausted); // filtered rows and the junk key count, since reading them cost the same assert_eq!(chunk.scanned, 5); // four rows scanned for one wanted output, the junk key among them let chunk = ReplayChunk::read(ks.iter(), None, 1, position, |_, _| None::<()>).unwrap(); assert!(chunk.events.is_empty()); assert_eq!(chunk.last_seen, Some(3)); assert!(!chunk.exhausted); assert_eq!(chunk.scanned, 4); let chunk = ReplayChunk::read(ks.range(9_u64.to_be_bytes()..), Some(8), 4, position, evens) .unwrap(); assert_eq!(chunk.events, [(10, "v10".into())]); assert_eq!(chunk.last_seen, Some(10)); assert!(chunk.exhausted); let chunk = ReplayChunk::read( ks.range(11_u64.to_be_bytes()..), Some(10), 4, position, evens, ) .unwrap(); assert!(chunk.events.is_empty()); assert_eq!(chunk.last_seen, Some(10)); assert!(chunk.exhausted); } }