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 {