diff --git a/crates/superbank/Cargo.toml b/crates/superbank/Cargo.toml index ea1730a..4e4c5b5 100644 --- a/crates/superbank/Cargo.toml +++ b/crates/superbank/Cargo.toml @@ -54,3 +54,4 @@ wincode05 = { package = "wincode", version = "0.5.0" } [dev-dependencies] solana-address = "1.1.0" solana-hash = "4.3.0" +tokio = { version = "1.46.1", features = ["full", "test-util"] } diff --git a/crates/superbank/src/ingest/rpc.rs b/crates/superbank/src/ingest/rpc.rs index 65fcbc4..b854cd0 100644 --- a/crates/superbank/src/ingest/rpc.rs +++ b/crates/superbank/src/ingest/rpc.rs @@ -535,6 +535,7 @@ async fn run_rpc_inserter(args: RpcInserterArgs<'_>) -> Result) -> Result { + progress = progress_rx.changed(), if progress_open => { if progress.is_ok() { let new_cursor = *progress_rx.borrow(); let new_processed = new_cursor @@ -587,6 +588,8 @@ async fn run_rpc_inserter(args: RpcInserterArgs<'_>) -> Result { @@ -2403,6 +2406,62 @@ mod tests { } } + #[test] + fn inserter_parks_after_discovery_drops_progress_sender() { + let (done_tx, done_rx) = std::sync::mpsc::channel(); + + std::thread::spawn(move || { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .start_paused(true) + .build() + .expect("build paused runtime"); + + runtime.block_on(async move { + let args = sample_args(); + + let (result_tx, result_rx) = mpsc::channel::(1); + let (progress_tx, progress_rx) = watch::channel(0u64); + let (_shutdown_tx, shutdown_rx) = watch::channel(0u64); + let (_fatal_tx, fatal_rx) = mpsc::channel::(1); + + drop(progress_tx); + + let inserter = tokio::spawn(async move { + run_rpc_inserter(RpcInserterArgs { + clickhouse: Arc::new(build_clickhouse_client(&args)), + insert_tables: Arc::new(InsertTables::from_args(&args)), + insert_concurrency: 2, + args: &args, + rpc_clients: Arc::new(Vec::new()), + result_rx, + progress_rx, + shutdown_rx, + fatal_rx, + range: RpcRange { start: 0, end: 0 }, + start_time: std::time::Instant::now(), + }) + .await + }); + + tokio::time::sleep(Duration::from_secs(30)).await; + + drop(result_tx); + inserter + .await + .expect("inserter task panicked") + .expect("inserter returned an error"); + }); + + let _ = done_tx.send(()); + }); + + assert!( + done_rx.recv_timeout(Duration::from_secs(10)).is_ok(), + "inserter kept polling the closed progress channel instead of parking", + ); + } + #[test] fn uningested_slots_drops_present_and_preserves_order() { let present: HashSet = [101, 106].into_iter().collect();