Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162use std::time::Duration;
use backon::{ExponentialBuilder, Retryable};use tokio::time::Instant;use tokio_util::sync::CancellationToken;use tracing::{debug, info, warn};
use crate::heads::RepoHeads;use crate::index::LanguageIndex;use crate::live::run_live;use crate::source::{LanguageSource, Page, SourceFuture};
const PAGE_LIMIT: usize = 500;const RETRY_MIN: Duration = Duration::from_secs(1);const RETRY_MAX: Duration = Duration::from_secs(60);
pub async fn run( source: &dyn LanguageSource, index: &LanguageIndex, heads: &RepoHeads, cancel: &CancellationToken,) { if backfill(source, index, cancel).await.is_none() { return; } info!(repos = index.len(), "cutting over to live language updates"); run_live(source, index, heads, cancel).await;}
async fn backfill( source: &dyn LanguageSource, index: &LanguageIndex, cancel: &CancellationToken,) -> Option<()> { let started = Instant::now(); let (mut cursor, mut pages, mut repos) = (None, 0, 0); loop { let page = next_page(source, cursor.as_deref(), cancel).await?; pages += 1; repos += page.repos.len(); for repo in &page.repos { let sizes = repo .languages .iter() .map(|(name, size)| (name.as_str(), *size)); index.apply(&repo.repo, repo.commit, sizes); } debug!( page = pages, repos = page.repos.len(), "loaded a page of indexed languages" ); cursor = page.cursor; if cursor.is_none() { break; } } info!(repos, pages, elapsed = ?started.elapsed(), "backfilled repo languages from the mirror"); Some(())}
async fn next_page( source: &dyn LanguageSource, cursor: Option<&str>, cancel: &CancellationToken,) -> Option<Page> { let backoff = ExponentialBuilder::default() .with_min_delay(RETRY_MIN) .with_max_delay(RETRY_MAX) .without_max_times(); let page: SourceFuture<'_, _> = Box::pin((|| source.page(cursor, PAGE_LIMIT)).retry(backoff).notify( |error, retry_in| warn!(%error, ?retry_in, "loading indexed languages failed"), )); cancel.run_until_cancelled(page).await?.ok()}
#[cfg(test)]mod tests { use tokio::time::sleep;
use super::*; use crate::fake::{Scripted, did, oid}; use crate::source::{Fetched, IndexedRepo};
fn page(repos: &[(&str, char, &str)], cursor: Option<&str>) -> Result<Page, String> { Ok(Page { repos: repos .iter() .map(|(repo, commit, language)| IndexedRepo { repo: did(repo), commit: oid(*commit), languages: vec![((*language).into(), 1)], }) .collect(), cursor: cursor.map(Into::into), }) }
#[tokio::test(start_paused = true)] async fn backfill_follows_the_cursor_and_retries_a_failed_page() { let source = Scripted::paging( vec![ page(&[("a", 'a', "Rust")], Some("did:plc:a")), Err("mirror returned 502".into()), page(&[("b", 'b', "Go")], None), ], vec![], ); let index = LanguageIndex::new(); assert!( backfill(&source, &index, &CancellationToken::new()) .await .is_some() ); assert_eq!( source.cursors(), [None, Some("did:plc:a".into()), Some("did:plc:a".into())], ); assert_eq!(index.primary(&did("a")).unwrap().name, "Rust"); assert_eq!(index.primary(&did("b")).unwrap().name, "Go"); }
#[tokio::test(start_paused = true)] async fn heads_seen_during_backfill_are_only_fetched_when_they_moved() { let source = Scripted::paging( vec![page(&[("a", 'a', "Rust"), ("b", 'b', "Go")], None)], vec![Fetched::Languages(vec![("Zig".into(), 3)])], ); let (index, heads, cancel) = ( LanguageIndex::new(), RepoHeads::new(), CancellationToken::new(), ); heads.observe(did("a"), Some(oid('a'))); heads.observe(did("b"), Some(oid('f')));
let stop = cancel.clone(); tokio::spawn(async move { sleep(Duration::from_secs(1)).await; stop.cancel(); }); run(&source, &index, &heads, &cancel).await;
assert_eq!(source.asked(), [(did("b"), oid('f'))]); assert_eq!(index.commit(&did("a")), Some(oid('a'))); assert_eq!(index.primary(&did("b")).unwrap().name, "Zig"); }
#[tokio::test(start_paused = true)] async fn cancelling_during_backfill_stops_without_going_live() { let source = Scripted::paging(vec![], vec![]); let cancel = CancellationToken::new(); cancel.cancel(); assert!( backfill(&source, &LanguageIndex::new(), &cancel) .await .is_none() ); }}