diff --git a/Cargo.toml b/Cargo.toml --- a/Cargo.toml +++ b/Cargo.toml @@ -71,6 +71,11 @@ [dev-dependencies] tempfile = "3.26.0" +[[bin]] +name = "replay_bench" +path = "src/bin/replay_bench.rs" +required-features = ["indexer_stream"] + [[example]] name = "statusphere" required-features = ["indexer_stream"] diff --git a/src/config.rs b/src/config.rs --- a/src/config.rs +++ b/src/config.rs @@ -457,10 +457,22 @@ /// set via `HYDRANT_DB_RECORDS_MEMTABLE_SIZE_MB`. pub db_records_memtable_size_mb: u64, - /// maximum number of persisted events read from the database per replay batch. + /// replay batch size. + /// + /// `0` (the default) means auto: use about half the subscriber channel capacity + /// (`STREAM_CHANNEL_CAPACITY / 2`), capped to the channel's current available + /// capacity. this prevents DB reads when the output buffer is already saturated. + /// + /// set to a non-zero value to fix the batch size manually. + /// /// set via `HYDRANT_STREAM_REPLAY_CHUNK_SIZE`. pub stream_replay_chunk_size: usize, - /// pause between replay batches, giving database maintenance work a chance to run. + /// optional pause between replay batches. + /// + /// normally zero. slow consumers are handled by bounded-channel backpressure + /// and `HYDRANT_STREAM_SEND_TIMEOUT`. only set this as an emergency knob if + /// you need to artificially throttle replay throughput. + /// /// set via `HYDRANT_STREAM_REPLAY_CHUNK_PAUSE` (humantime duration, e.g. `2ms`). pub stream_replay_chunk_pause: Duration, /// maximum number of live in-memory stream events buffered per subscriber while it catches up. @@ -543,8 +555,8 @@ db_repos_memtable_size_mb: BASE_MEMTABLE_MB / 2, db_events_memtable_size_mb: BASE_MEMTABLE_MB, db_records_memtable_size_mb: BASE_MEMTABLE_MB / 3 * 2, - stream_replay_chunk_size: 64, - stream_replay_chunk_pause: Duration::from_millis(2), + stream_replay_chunk_size: 0, + stream_replay_chunk_pause: Duration::ZERO, stream_pending_event_limit: 4096, stream_send_timeout: Duration::from_secs(30), } @@ -944,7 +956,12 @@ "db records memtable", format_args!("{} mb", self.db_records_memtable_size_mb) )?; - config_line!(f, "stream replay chunk", self.stream_replay_chunk_size)?; + let replay_chunk = if self.stream_replay_chunk_size == 0 { + "auto".to_owned() + } else { + self.stream_replay_chunk_size.to_string() + }; + config_line!(f, "stream replay chunk", replay_chunk)?; config_line!( f, "stream replay pause", diff --git a/src/bin/replay_bench.rs b/src/bin/replay_bench.rs new file mode 100644 --- /dev/null +++ b/src/bin/replay_bench.rs @@ -0,0 +1,248 @@ +//! replay throughput benchmark. +//! +//! populates a temporary hydrant database with a configurable number of synthetic +//! events, then measures how long it takes to drain them through `Hydrant::subscribe` +//! under two config profiles: +//! +//! - **old**: `stream_replay_chunk_size = 64`, `stream_replay_chunk_pause = 2ms` +//! - **new**: `stream_replay_chunk_size = 0` (auto), `stream_replay_chunk_pause = 0ms` +//! +//! run with: +//! cargo run --bin replay_bench +//! +//! optional env vars: +//! REPLAY_BENCH_EVENTS= number of events to write (default: 20_000) +//! REPLAY_BENCH_RUNS= number of timed runs per profile (default: 5) + +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; + +use futures::StreamExt; +use hydrant::config::Config; +use hydrant::control::Hydrant; + +fn n_events() -> usize { + std::env::var("REPLAY_BENCH_EVENTS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(20_000) +} + +fn n_runs() -> usize { + std::env::var("REPLAY_BENCH_RUNS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(5) +} + +struct TempDir(PathBuf); + +impl Drop for TempDir { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } +} + +/// Drain a cursor=0 subscriber until `total` events are received. +/// Returns (wall-clock elapsed, event count received). +async fn drain_stream(hydrant: &Hydrant, total: usize) -> (Duration, usize) { + let mut stream = hydrant.subscribe(Some(0)); + let start = Instant::now(); + let mut count = 0usize; + while let Some(item) = stream.next().await { + match item { + Ok(_) => count += 1, + Err(err) => { + eprintln!("stream error after {count} events: {err}"); + break; + } + } + if count >= total { + break; + } + } + (start.elapsed(), count) +} + +fn make_config(db_path: &Path, chunk_size: usize, chunk_pause: Duration) -> Config { + Config { + database_path: db_path.to_path_buf(), + enable_firehose: false, + enable_crawler: Some(false), + stream_replay_chunk_size: chunk_size, + stream_replay_chunk_pause: chunk_pause, + ..Config::default() + } +} + +fn stats(samples: &[Duration]) -> (Duration, Duration, Duration) { + let mut sorted = samples.to_vec(); + sorted.sort(); + let min = *sorted.first().unwrap(); + let max = *sorted.last().unwrap(); + let mean = sorted.iter().sum::() / sorted.len() as u32; + (min, mean, max) +} + +fn fmt_ms(d: Duration) -> String { + format!("{:.1}ms", d.as_secs_f64() * 1000.0) +} + +#[tokio::main] +async fn main() -> miette::Result<()> { + rustls::crypto::aws_lc_rs::default_provider() + .install_default() + .ok(); + + let args: Vec = std::env::args().collect(); + if args.len() > 1 && args[1] == "--run-profile" { + // subprocess mode: runs a single profile and prints run times as space-separated milliseconds + let profile = &args[2]; + let dir_path = Path::new(&args[3]); + let n = args[4].parse::().unwrap(); + let runs = args[5].parse::().unwrap(); + + let cfg = if profile == "old" { + make_config(dir_path, 64, Duration::from_millis(2)) + } else { + make_config(dir_path, 0, Duration::ZERO) + }; + + let hydrant = Hydrant::new(cfg).await?; + let mut times = Vec::with_capacity(runs); + for _ in 0..runs { + let (elapsed, got) = drain_stream(&hydrant, n).await; + assert_eq!(got, n, "[{profile}] expected {n} events, got {got}"); + times.push(elapsed.as_secs_f64() * 1000.0); + } + + // output results to stdout so parent can parse them + let formatted: Vec = times.into_iter().map(|t| format!("{t}")).collect(); + println!("{}", formatted.join(" ")); + return Ok(()); + } + + // parent mode + let n = n_events(); + let runs = n_runs(); + + println!("replay_bench: {n} events, {runs} runs per profile"); + println!(); + + // ── seed ───────────────────────────────────────────────────────────────── + // write events into a temp dir once. + let dir_path = std::env::temp_dir().join(format!( + "hydrant_replay_bench_{}_{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + )); + std::fs::create_dir_all(&dir_path).expect("failed to create temp dir"); + let _cleanup = TempDir(dir_path.clone()); + + { + let seed_cfg = make_config(&dir_path, 64, Duration::from_millis(2)); + let seed = Hydrant::new(seed_cfg).await?; + let bytes = seed.seed_events_for_bench(n); + println!( + "wrote {n} events to {} ({:.1} KB, ~{} B/event)", + dir_path.display(), + bytes as f64 / 1024.0, + bytes / n.max(1), + ); + // dropped here — closes db cleanly + } + + let current_exe = std::env::current_exe().expect("failed to get current executable path"); + + // ── old profile: chunk_size=64, pause=2ms ──────────────────────────────── + println!(); + println!("Running old profile (64 / 2ms)..."); + let output_old = std::process::Command::new(¤t_exe) + .args([ + "--run-profile", + "old", + dir_path.to_str().unwrap(), + &n.to_string(), + &runs.to_string(), + ]) + .output() + .expect("failed to execute old profile subprocess"); + + if !output_old.status.success() { + eprintln!("{}", String::from_utf8_lossy(&output_old.stderr)); + miette::bail!("old profile subprocess failed"); + } + + let stdout_old = String::from_utf8(output_old.stdout).unwrap(); + let old_times: Vec = stdout_old + .split_whitespace() + .map(|s| Duration::from_secs_f64(s.parse::().unwrap() / 1000.0)) + .collect(); + + for (i, t) in old_times.iter().enumerate() { + println!(" [old] run {}/{}: {}", i + 1, runs, fmt_ms(*t)); + } + + // ── new profile: chunk_size=0 (auto), pause=0ms ────────────────────────── + println!(); + println!("Running new profile (auto / 0ms)..."); + let output_new = std::process::Command::new(¤t_exe) + .args([ + "--run-profile", + "new", + dir_path.to_str().unwrap(), + &n.to_string(), + &runs.to_string(), + ]) + .output() + .expect("failed to execute new profile subprocess"); + + if !output_new.status.success() { + eprintln!("{}", String::from_utf8_lossy(&output_new.stderr)); + miette::bail!("new profile subprocess failed"); + } + + let stdout_new = String::from_utf8(output_new.stdout).unwrap(); + let new_times: Vec = stdout_new + .split_whitespace() + .map(|s| Duration::from_secs_f64(s.parse::().unwrap() / 1000.0)) + .collect(); + + for (i, t) in new_times.iter().enumerate() { + println!(" [new] run {}/{}: {}", i + 1, runs, fmt_ms(*t)); + } + + // ── summary ─────────────────────────────────────────────────────────────── + let (o_min, o_mean, o_max) = stats(&old_times); + let (n_min, n_mean, n_max) = stats(&new_times); + let speedup = o_mean.as_secs_f64() / n_mean.as_secs_f64(); + + println!(); + println!("┌──────────────────────────────────────────────────────────────────┐"); + println!("│ replay throughput: {n} events, {runs} runs per profile"); + println!("├──────────────┬───────────┬───────────┬───────────┬───────────────┤"); + println!("│ profile │ min │ mean │ max │ Kev/s (mean) │"); + println!("├──────────────┼───────────┼───────────┼───────────┼───────────────┤"); + println!( + "│ old(64/2ms) │ {:>8} │ {:>8} │ {:>8} │ {:>11.1} │", + fmt_ms(o_min), + fmt_ms(o_mean), + fmt_ms(o_max), + n as f64 / o_mean.as_secs_f64() / 1000.0, + ); + println!( + "│ new(auto/0) │ {:>8} │ {:>8} │ {:>8} │ {:>11.1} │", + fmt_ms(n_min), + fmt_ms(n_mean), + fmt_ms(n_max), + n as f64 / n_mean.as_secs_f64() / 1000.0, + ); + println!("├──────────────┴───────────┴───────────┴───────────┴───────────────┤"); + println!("│ speedup (mean): {speedup:.2}×"); + println!("└──────────────────────────────────────────────────────────────────┘"); + + Ok(()) +} diff --git a/src/control/indexer.rs b/src/control/indexer.rs --- a/src/control/indexer.rs +++ b/src/control/indexer.rs @@ -78,7 +78,7 @@ /// the stream ends when the `EventStream` is dropped. slow consumers receive /// [`StreamError::ConsumerTooSlow`] before the stream terminates when possible. pub fn subscribe(&self, cursor: Option) -> EventStream { - let (tx, rx) = mpsc::channel(500); + let (tx, rx) = mpsc::channel(stream::STREAM_CHANNEL_CAPACITY); let state = self.state.clone(); let runtime = tokio::runtime::Handle::current(); let opts = stream::StreamOptions::from_config(&self.config); @@ -96,5 +96,46 @@ pub(crate) fn stream_send_timeout(&self) -> std::time::Duration { self.config.stream_send_timeout + } + + #[cfg(feature = "indexer_stream")] + #[doc(hidden)] + pub fn seed_events_for_bench(&self, count: usize) -> usize { + use crate::db::keys; + use crate::db::types::{DbAction, DbRkey, DbTid, TrimmedDid}; + use crate::types::{StoredData, StoredEvent}; + use jacquard_common::types::did::Did; + use jacquard_common::{CowStr, IntoStatic}; + use std::sync::atomic::Ordering; + + let db = &self.state.db; + let mut batch = db.inner.batch(); + let mut total_bytes = 0usize; + + let did_str = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaaaaaa"; + let did = Did::new(did_str).expect("valid plc did"); + let trimmed = TrimmedDid::from(&did).into_static(); + let rev = DbTid::new_from_bytes([0u8; 8]); + let collection = CowStr::Borrowed("app.bsky.feed.post").into_static(); + + for i in 0..count { + let event_id = db.next_event_id.fetch_add(1, Ordering::SeqCst); + let rkey = DbRkey::Str(smol_str::format_smolstr!("r{i}")); + let evt = StoredEvent { + live: false, + did: trimmed.clone(), + rev, + collection: collection.clone(), + rkey, + action: DbAction::Create, + data: StoredData::Nothing, + }; + let bytes = rmp_serde::to_vec(&evt).expect("msgpack serialization cannot fail"); + total_bytes += bytes.len(); + batch.insert(&db.events, keys::event_key(event_id), bytes); + } + + batch.commit().expect("failed to commit events batch"); + total_bytes } } diff --git a/src/control/jetstream.rs b/src/control/jetstream.rs --- a/src/control/jetstream.rs +++ b/src/control/jetstream.rs @@ -139,7 +139,7 @@ cursor: Option, filter: JetstreamFilter, ) -> JetstreamEventStream { - let (tx, rx) = mpsc::channel(500); + let (tx, rx) = mpsc::channel(stream::STREAM_CHANNEL_CAPACITY); let state = self.state.clone(); let runtime = tokio::runtime::Handle::current(); let opts = stream::StreamOptions::from_config(&self.config); diff --git a/src/control/relay.rs b/src/control/relay.rs --- a/src/control/relay.rs +++ b/src/control/relay.rs @@ -40,7 +40,7 @@ /// slow consumers receive [`RelayStreamError::ConsumerTooSlow`] before the stream terminates /// when possible. pub fn subscribe_repos(&self, cursor: Option) -> RelayEventStream { - let (tx, rx) = mpsc::channel(500); + let (tx, rx) = mpsc::channel(stream::STREAM_CHANNEL_CAPACITY); let state = self.state.clone(); let runtime = tokio::runtime::Handle::current(); let opts = stream::StreamOptions::from_config(&self.config); diff --git a/src/control/stream.rs b/src/control/stream.rs --- a/src/control/stream.rs +++ b/src/control/stream.rs @@ -1,5 +1,6 @@ use std::collections::VecDeque; use std::fmt; +use std::num::NonZeroUsize; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -38,13 +39,17 @@ #[cfg(all(feature = "relay", feature = "jetstream"))] use crate::ingest::stream::{SubscribeReposMessage, decode_frame}; -#[cfg(any(feature = "indexer_stream", feature = "relay"))] +#[cfg(any(feature = "indexer_stream", feature = "relay", feature = "jetstream"))] const STREAM_SEND_RETRY_PAUSE: Duration = Duration::from_millis(10); + +pub(super) const STREAM_CHANNEL_CAPACITY: usize = 500; #[cfg(any(feature = "indexer_stream", feature = "relay"))] #[derive(Debug, Clone, Copy)] pub(crate) struct StreamOptions { - replay_chunk_size: usize, + /// `None` = auto: use roughly half the output channel capacity, capped by + /// current available capacity. `Some(n)` = manual override. + replay_chunk_size: Option, replay_chunk_pause: Duration, pending_event_limit: usize, send_timeout: Duration, @@ -54,7 +59,8 @@ impl StreamOptions { pub(crate) fn from_config(config: &Config) -> Self { Self { - replay_chunk_size: config.stream_replay_chunk_size.max(1), + // 0 means auto; NonZeroUsize::new returns None for 0. + replay_chunk_size: NonZeroUsize::new(config.stream_replay_chunk_size), replay_chunk_pause: config.stream_replay_chunk_pause, pending_event_limit: config.stream_pending_event_limit.max(1), send_timeout: config.stream_send_timeout, @@ -246,6 +252,7 @@ { let mut replay_gap_target = None; let mut pending = PendingLiveEvents::new(opts.pending_event_limit); + let mut replay_blocked_since: Option = None; loop { if let Err(err) = drain_pending_broadcasts(&mut event_rx, &mut pending) { @@ -275,7 +282,27 @@ .and_then(|seq| seq.checked_sub(1)) .map(|before_pending| before_pending.min(target)) .unwrap_or(target); - let chunk = read_replay_chunk(current_seq, effective_target, opts.replay_chunk_size); + + // skip the DB read entirely when the output channel is saturated. + // this mirrors the send_timeout semantics already enforced inside + // send_stream_event: a continuously-full channel will eventually + // trigger StreamTooSlow::send_timeout and close the stream. + let Some(chunk_size) = replay_chunk_size_for(&tx, opts.replay_chunk_size) else { + if let Err(err) = drain_pending_broadcasts(&mut event_rx, &mut pending) { + send_stream_error(&tx, err.into()); + return; + } + if let Err(err) = note_replay_blocked(&mut replay_blocked_since, opts.send_timeout) + { + send_stream_error(&tx, err.into()); + return; + } + std::thread::sleep(STREAM_SEND_RETRY_PAUSE); + continue; + }; + clear_replay_blocked(&mut replay_blocked_since); + + let chunk = read_replay_chunk(current_seq, effective_target, chunk_size); current_seq = chunk.last_seen_seq.or(current_seq); for event in chunk.events { @@ -295,6 +322,7 @@ } replay_gap_target = Some(effective_target); } else if !opts.replay_chunk_pause.is_zero() { + // emergency compatibility knob; zero by default. std::thread::sleep(opts.replay_chunk_pause); } @@ -570,6 +598,59 @@ current_id.map(|id| id.saturating_add(1)).unwrap_or(0) } +/// Returns the chunk size to use for the next DB replay read, or `None` if the +/// output channel has no remaining capacity and the read should be skipped. +/// +/// - In auto mode (`configured = None`): target half of the channel's maximum +/// capacity so the reader stays roughly half a window ahead. +/// - In manual mode (`configured = Some(n)`): use `n`, but still cap to the +/// channel's hard maximum to avoid over-reading. +/// +/// In both modes the result is further capped to the current *available* +/// capacity so we never read more events than the channel can immediately absorb. +#[cfg(any(feature = "indexer_stream", feature = "relay", feature = "jetstream"))] +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)) +} + +/// Records the instant the replay loop first found the channel full. Returns +/// `Err(StreamTooSlow)` if the channel has been continuously full for longer +/// than `timeout`. +#[cfg(any(feature = "indexer_stream", feature = "relay"))] +fn note_replay_blocked( + blocked_since: &mut Option, + timeout: Duration, +) -> Result<(), StreamTooSlow> { + let started = blocked_since.get_or_insert_with(Instant::now); + + if started.elapsed() >= timeout { + return Err(StreamTooSlow::send_timeout(timeout)); + } + + Ok(()) +} + +/// Resets the replay-blocked timer once the channel has capacity again. +#[cfg(any(feature = "indexer_stream", feature = "relay"))] +fn clear_replay_blocked(blocked_since: &mut Option) { + *blocked_since = None; +} + #[cfg(feature = "relay")] pub(super) fn relay_stream_thread( state: Arc, @@ -802,7 +883,7 @@ send_stream_error(tx, StreamTooSlow::send_timeout(opts.send_timeout).into()); return Err(()); } - std::thread::sleep(Duration::from_millis(10)); + std::thread::sleep(STREAM_SEND_RETRY_PAUSE); } } } @@ -819,13 +900,25 @@ opts: StreamOptions, ) -> Option> { let mut max_id_seen: Option = None; + let mut replay_blocked_since: Option = None; loop { // drain live events without buffering so the broadcast receiver never // lags regardless of how many events arrive during the replay window. drain_jetstream_broadcast(event_rx); - let chunk = - read_jetstream_replay_chunk(state, &next_key, target_time_us, opts.replay_chunk_size); + // skip the DB read when the output channel is already saturated. + let Some(chunk_size) = replay_chunk_size_for(tx, opts.replay_chunk_size) else { + drain_jetstream_broadcast(event_rx); + if let Err(err) = note_replay_blocked(&mut replay_blocked_since, opts.send_timeout) { + send_stream_error(tx, err.into()); + return None; + } + std::thread::sleep(STREAM_SEND_RETRY_PAUSE); + continue; + }; + clear_replay_blocked(&mut replay_blocked_since); + + let chunk = read_jetstream_replay_chunk(state, &next_key, target_time_us, chunk_size); for event in chunk.events { let event_id = event.id; next_key = keys::jetstream_event_key(event.time_us as u64, event_id.saturating_add(1)) @@ -843,6 +936,7 @@ return Some(max_id_seen); } if !opts.replay_chunk_pause.is_zero() { + // emergency compatibility knob; zero by default. std::thread::sleep(opts.replay_chunk_pause); } } @@ -1190,10 +1284,14 @@ } } + fn chunk_size(n: usize) -> Option { + NonZeroUsize::new(n) + } + #[test] fn ordered_stream_replays_chunks_then_live_tail() { let opts = StreamOptions { - replay_chunk_size: 2, + replay_chunk_size: chunk_size(2), replay_chunk_pause: Duration::ZERO, pending_event_limit: 4, send_timeout: Duration::from_secs(1), @@ -1248,8 +1346,11 @@ #[test] fn ordered_stream_closes_when_output_queue_is_full() { + // channel capacity 1: one event is enqueued by replay_chunk_size_for, + // then the channel is full. send_stream_event's retry loop fires + // send_timeout immediately (Duration::ZERO) and the stream closes. let opts = StreamOptions { - replay_chunk_size: 2, + replay_chunk_size: chunk_size(2), replay_chunk_pause: Duration::ZERO, pending_event_limit: 4, send_timeout: Duration::ZERO, @@ -1273,6 +1374,70 @@ TestBroadcast::Live(seq) => Some(seq), }, ); + } + + // --- replay_chunk_size_for unit tests --- + + #[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); + // requesting more than the channel can ever hold is silently capped. + 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); + + // fill 3 of 4 slots — only 1 is free. + tx.try_send(()).unwrap(); + tx.try_send(()).unwrap(); + tx.try_send(()).unwrap(); + + // auto mode: half of 4 is 2, but only 1 slot is free. + 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(); + + // even an explicit size of 1 returns None when the channel is full. + assert_eq!(replay_chunk_size_for(&tx, chunk_size(1)), None); + } + + #[test] + fn replay_chunk_auto_min_one_for_small_channel() { + // channel of 1 — half rounds down to 0, should be floored to 1. + let (tx, _rx) = mpsc::channel::<()>(1); + assert_eq!(replay_chunk_size_for(&tx, None), Some(1)); } }