From 8999521aa4f6c0f3baadb758bdbe22f78ab08315 Mon Sep 17 00:00:00 2001 From: Yorgos Saslis Date: Thu, 29 Oct 2026 11:34:26 +0000 Subject: [PATCH] httpd: Keep reading node events during rescans A rescan can take tens of minutes on permissive seeds and blocked the event loop meanwhile. radicle-node drops a subscriber once 8192 events are queued, so every rescan disconnected us and the events emitted until the resubscribe were lost. Rescans and batch reindexes now run alongside the event loop, and the subscription reconnects on its own task. A repo that an event adds during a rescan only gets back into the seeded cache when its batch runs, so it can be missing for up to RADICLE_SEARCH_BATCH_SECS. Co-authored-by: Daniel Norman --- explorer/CHANGELOG.md | 9 + explorer/crates/radicle-search/src/indexer.rs | 178 +++++++++++------- 2 files changed, 116 insertions(+), 71 deletions(-) diff --git a/explorer/CHANGELOG.md b/explorer/CHANGELOG.md index 834a60729..ef847f6c0 100644 --- a/explorer/CHANGELOG.md +++ b/explorer/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog +## Unreleased + +- **Event subscription kept during rescans**: radicle-search keeps reading node + events while it rescans or reindexes, so the node no longer drops its + subscription during every periodic rescan, and the events emitted meanwhile + are no longer lost. Rescans and batched reindexes also keep running while the + event stream is down, and the catch-up rescan after a reconnect runs in the + background instead of delaying the resubscription + ## radicle-httpd + radicle-search 0.30.0 - **`tar.zst` archive downloads**: `/raw/{rid}/archive/{ref}.tar.zst` joins the diff --git a/explorer/crates/radicle-search/src/indexer.rs b/explorer/crates/radicle-search/src/indexer.rs index 7c1ca2e08..5bbb61e74 100644 --- a/explorer/crates/radicle-search/src/indexer.rs +++ b/explorer/crates/radicle-search/src/indexer.rs @@ -3,6 +3,8 @@ mod build; mod event; use std::collections::HashSet; +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; use std::time::Duration; @@ -198,6 +200,7 @@ impl Indexer { match action { ReindexAction::Upsert(doc) => { self.index.upsert(std::slice::from_ref(&*doc)).await?; + self.seeded.insert(key).await; } ReindexAction::Delete => { tracing::info!("removing {rid} from index (no longer seeded)"); @@ -214,60 +217,8 @@ impl Indexer { res = self.bootstrap() => res?, } - loop { - let sub_shutdown = shutdown.clone(); - tokio::select! { - _ = shutdown.changed() => return Ok(()), - res = self.subscribe_loop(sub_shutdown) => match res { - Ok(()) => tracing::warn!("event stream ended without error; reconnecting"), - Err(e) => tracing::warn!("event stream error: {e:#}; reconnecting"), - }, - } - tokio::select! { - _ = shutdown.changed() => return Ok(()), - _ = tokio::time::sleep(self.config.reconnect_backoff) => {} - } - tokio::select! { - _ = shutdown.changed() => return Ok(()), - res = self.bootstrap() => { - if let Err(e) = res { - tracing::error!("rescan after disconnect failed: {e:#}"); - } - } - } - } - } - - async fn subscribe_loop(&self, mut shutdown: watch::Receiver) -> Result<()> { - let socket = self.profile.home().socket_from_env(); - tracing::info!("subscribing to node events at {}", socket.display()); - let node = radicle::Node::new(&socket); - - let (tx, mut rx) = mpsc::channel::(EVENT_CHANNEL_CAPACITY); - let blocking_shutdown = shutdown.clone(); - let blocking = tokio::task::spawn_blocking(move || -> Result<()> { - let events = node - .subscribe(Duration::from_secs(1)) - .context("subscribe to node socket failed")?; - for event in events { - if *blocking_shutdown.borrow() { - break; - } - let event = match event { - Ok(e) => e, - Err(radicle::node::Error::TimedOut) => continue, - Err(e) => { - return Err(anyhow::Error::from(e).context("event read failed")); - } - }; - if tx.blocking_send(event).is_err() { - break; - } - } - Ok(()) - }); - let mut rescan_timer = tokio::time::interval(self.config.rescan_interval); + rescan_timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); rescan_timer.tick().await; let mut pending = HashSet::new(); @@ -275,11 +226,27 @@ impl Indexer { batch_timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); batch_timer.tick().await; + let (tx, mut rx) = mpsc::channel::(EVENT_CHANNEL_CAPACITY); + let reader = tokio::spawn(read_events( + self.profile.clone(), + self.config.reconnect_backoff, + shutdown.clone(), + tx, + )); + + let mut work: Option + '_>>> = None; loop { tokio::select! { _ = shutdown.changed() => break, - event = rx.recv() => { - let Some(event) = event else { break; }; + feed = rx.recv() => { + let event = match feed { + Some(Feed::Event(event)) => event, + Some(Feed::Resubscribed) => { + rescan_timer.reset_immediately(); + continue; + } + None => break, + }; let Some((rid, category)) = event::classify_event(&event) else { tracing::debug!("ignored event: {}", event::event_kind(&event)); continue; @@ -307,31 +274,100 @@ impl Indexer { ); } } - _ = batch_timer.tick() => { - if pending.is_empty() { - continue; + () = async { + if let Some(work) = work.as_mut() { + work.await; } - tracing::info!("reindexing {} repositories (batched)", pending.len()); - for rid in pending.drain().collect::>() { - if let Err(e) = self.reindex(rid).await { - tracing::warn!("reindex {} failed: {e:#}", rid); - } + }, if work.is_some() => work = None, + _ = batch_timer.tick(), if work.is_none() => { + if !pending.is_empty() { + work = Some(Box::pin(self.reindex_batch(pending.drain().collect()))); } } - _ = rescan_timer.tick() => { - if let Err(e) = self.bootstrap().await { - tracing::warn!("periodic rescan failed: {e:#}"); - } + _ = rescan_timer.tick(), if work.is_none() => { + work = Some(Box::pin(self.rescan())); } } } - blocking - .await - .context("event reader task panicked")? - .context("event reader task returned an error")?; + drop(rx); + reader.await.context("event reader task panicked")?; Ok(()) } + + async fn reindex_batch(&self, rids: Vec) { + tracing::info!("reindexing {} repositories (batched)", rids.len()); + for rid in rids { + if let Err(e) = self.reindex(rid).await { + tracing::warn!("reindex {} failed: {e:#}", rid); + } + } + } + + async fn rescan(&self) { + if let Err(e) = self.bootstrap().await { + tracing::warn!("periodic rescan failed: {e:#}"); + } + } +} + +enum Feed { + Event(Event), + Resubscribed, +} + +async fn read_events( + profile: Arc, + reconnect_backoff: Duration, + mut shutdown: watch::Receiver, + tx: mpsc::Sender, +) { + let mut resubscribing = false; + loop { + let socket = profile.home().socket_from_env(); + tracing::info!("subscribing to node events at {}", socket.display()); + let node = radicle::Node::new(&socket); + let feed = tx.clone(); + let blocking_shutdown = shutdown.clone(); + let res = tokio::task::spawn_blocking(move || -> Result<()> { + let events = node + .subscribe(Duration::from_secs(1)) + .context("subscribe to node socket failed")?; + if resubscribing && feed.blocking_send(Feed::Resubscribed).is_err() { + return Ok(()); + } + for event in events { + if *blocking_shutdown.borrow() { + break; + } + let event = match event { + Ok(e) => e, + Err(radicle::node::Error::TimedOut) => continue, + Err(e) => { + return Err(anyhow::Error::from(e).context("event read failed")); + } + }; + if feed.blocking_send(Feed::Event(event)).is_err() { + break; + } + } + Ok(()) + }) + .await; + if *shutdown.borrow() || tx.is_closed() { + return; + } + match res { + Ok(Ok(())) => tracing::warn!("event stream ended without error; reconnecting"), + Ok(Err(e)) => tracing::warn!("event stream error: {e:#}; reconnecting"), + Err(e) => tracing::warn!("event reader task panicked: {e}; reconnecting"), + } + tokio::select! { + _ = shutdown.changed() => return, + _ = tokio::time::sleep(reconnect_backoff) => {} + } + resubscribing = true; + } } enum ReindexAction {