Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873use crate::db::types::DbTid;use crate::db::{keys, ser_repo_state};use crate::state::AppState;use crate::types::{GaugeState, RepoMetadata, RepoState, RepoStatus, ResyncState};use jacquard_api::com_atproto::sync::list_repos::RepoStatus as ListedRepoStatus;use jacquard_common::IntoStatic;use jacquard_common::types::did::Did;use miette::{IntoDiagnostic, Result};use rand::Rng;use rand::rngs::SmallRng;use smol_str::ToSmolStr;use std::sync::Arc;use std::time::Duration;use tokio::sync::mpsc;use tracing::{debug, error, info, trace};
use super::{CrawlerStats, InFlightGuard};
const WORKER_CHANNEL_CAPACITY: usize = 64;const BLOCKING_TASK_TIMEOUT: Duration = Duration::from_secs(30);
#[derive(Debug, Clone, Copy, PartialEq, Eq)]enum CapacityDecision { Proceed, EnterThrottle, StayThrottled, ReleaseThrottle,}
fn capacity_decision( pending: u64, max_pending: usize, resume_pending: usize, was_throttled: bool,) -> CapacityDecision { if pending > max_pending as u64 { if was_throttled { CapacityDecision::StayThrottled } else { CapacityDecision::EnterThrottle } } else if was_throttled && pending > resume_pending as u64 { CapacityDecision::StayThrottled } else if was_throttled { CapacityDecision::ReleaseThrottle } else { CapacityDecision::Proceed }}
/// a cursor write to include atomically with the batch.pub(crate) struct CursorUpdate { pub(super) key: Vec<u8>, pub(super) value: Option<Vec<u8>>,}
#[derive(Clone)]pub(crate) struct RepoListing { pub(super) did: Did, pub(super) rev: DbTid, pub(super) active: bool, pub(super) status: Option<ListedRepoStatus>, pub(super) create_if_missing: bool,}
/// a batch of confirmed repos from any crawler source ready to enqueue.////// `guards` hold the DIDs (via `Deref`) and keep their in-flight slots occupied/// until their lock group is committed to the database. cursor updates commit/// only after every idempotent group, so a crash can replay but never skip a/// partially applied page.pub(crate) struct CrawlerBatch { pub(super) guards: Vec<InFlightGuard>, pub(super) listings: Vec<RepoListing>, pub(super) cursor_updates: Vec<CursorUpdate>,}
pub(crate) struct CrawlerWorker { pub(super) state: Arc<AppState>, pub(super) max_pending: usize, pub(super) resume_pending: usize, pub(super) stats: CrawlerStats, pub(super) rx: mpsc::Receiver<CrawlerBatch>, pub(super) was_throttled: bool,}
impl CrawlerWorker { pub(crate) fn new( state: Arc<AppState>, max_pending: usize, resume_pending: usize, stats: CrawlerStats, ) -> (Self, mpsc::Sender<CrawlerBatch>) { let (tx, rx) = mpsc::channel(WORKER_CHANNEL_CAPACITY); ( Self { state, max_pending, resume_pending, stats, rx, was_throttled: false, }, tx, ) }
pub(crate) async fn run(mut self) { while let Some(batch) = self.rx.recv().await { self.wait_for_capacity().await; if let Err(e) = self.enqueue(batch).await { error!(err = ?e, "crawler worker: enqueue failed"); } } }
/// blocks until the pending queue has capacity. mirrors the hysteresis logic /// that was previously duplicated in each crawler loop: /// - above `max_pending`: hard stop, poll every 5s /// - after crossing `max_pending`: cooldown until below `resume_pending` /// - below `resume_pending`: proceed (or release throttle) async fn wait_for_capacity(&mut self) { loop { let pending = self.state.db.get_count("pending").await; match capacity_decision( pending, self.max_pending, self.resume_pending, self.was_throttled, ) { CapacityDecision::Proceed => break, CapacityDecision::EnterThrottle => { debug!( pending, max = self.max_pending, "throttling: above max pending" ); self.was_throttled = true; self.stats.set_throttled(true); tokio::time::sleep(Duration::from_secs(5)).await; } CapacityDecision::StayThrottled => { tokio::time::sleep(Duration::from_secs(5)).await; } CapacityDecision::ReleaseThrottle => { self.was_throttled = false; self.stats.set_throttled(false); info!("throttling released"); break; } } } }
/// filters already-known repos, commits them to `repos` + `pending` (plus /// an optional cursor update), then releases the in-flight guards. async fn enqueue(&mut self, batch: CrawlerBatch) -> Result<()> { let CrawlerBatch { guards, listings, cursor_updates, } = batch;
// nothing to insert but still need to commit the cursor if present. if guards.is_empty() && listings.is_empty() { if !cursor_updates.is_empty() { self.commit_cursors(cursor_updates).await?; } return Ok(()); }
// Group by the one record lock each DID needs. Acquiring a whole crawler // page at once usually captures nearly all 256 locks and turns one long // operator redaction into a global ingestion convoy. let app_state = self.state.clone(); let surviving = app_state .db .run(move |db| { let mut rng: SmallRng = rand::make_rng(); let mut listing_groups = std::collections::BTreeMap::<u8, Vec<_>>::new(); for listing in listings { listing_groups .entry(crate::db::record_lock_index_for_did(&listing.did)) .or_default() .push(listing); } let mut guard_groups = std::collections::BTreeMap::<u8, Vec<_>>::new(); for guard in guards { guard_groups .entry(crate::db::record_lock_index_for_did(&guard)) .or_default() .push(guard); } let lock_indices = listing_groups .keys() .chain(guard_groups.keys()) .copied() .collect::<std::collections::BTreeSet<_>>(); let mut surviving = Vec::new(); let mut changed_listings = Vec::new();
for lock_index in lock_indices { let mut txn = crate::db::Txn::new(db); txn.hold_record_lock_index(lock_index); for listing in listing_groups.remove(&lock_index).unwrap_or_default() { if reconcile_listing(&mut txn, &listing)?.emits_account() { changed_listings.push(listing); } } for guard in guard_groups.remove(&lock_index).unwrap_or_default() { if db .filter .contains_key(crate::db::filter::exclude_key(guard.as_str())?) .into_diagnostic()? { continue; } let did_key = keys::repo_key(&guard); let metadata_key = keys::repo_metadata_key(&guard); if db.repos.contains_key(&did_key).into_diagnostic()? { continue; } let state = RepoState::backfilling(); let metadata = RepoMetadata::backfilling(rng.next_u64()); txn.batch .insert(&db.repos, &did_key, ser_repo_state(&state)?); txn.batch.insert( &db.repo_metadata, &metadata_key, crate::db::ser_repo_meta(&metadata)?, ); #[cfg(feature = "indexer")] txn.batch.insert( &db.indexer.pending, keys::pending_key(metadata.index_id), &did_key, ); txn.batch .remove(&db.crawler, keys::crawler_retry_key(&guard)); trace!(did = %*guard, "enqueuing repo"); txn.counts.add_repos(1); #[cfg(feature = "indexer")] txn.transition_lifecycle(&guard, crate::types::GaugeState::Pending)?; surviving.push(guard); } txn.commit()?; }
let mut txn = crate::db::Txn::new(db); for cursor in cursor_updates { stage_cursor(&mut txn.batch, db, cursor); } txn.commit()?; Ok((surviving, changed_listings)) }) .await .inspect_err(|e| error!(err = ?e, "enqueue batch commit failed")) .unwrap_or_default();
let (surviving, changed_listings) = surviving; let count = surviving.len(); // release in-flight slots now that the batch is committed drop(surviving);
if count > 0 { self.stats.record_processed(count); #[cfg(feature = "indexer")] self.state.notify_backfill(); }
#[cfg(feature = "indexer_stream")] for listing in changed_listings { let evt = crate::types::AccountEvt { did: listing.did, active: listing.active, status: listing .status .map(|status| jacquard_common::CowStr::Owned(status.as_str().to_smolstr())), }; let _ = self .state .db .stream .event_tx .send(crate::ops::make_account_event(&self.state.db, evt)); } #[cfg(not(feature = "indexer_stream"))] drop(changed_listings);
Ok(()) }
async fn commit_cursors(&self, cursors: Vec<CursorUpdate>) -> Result<()> { let state = self.state.clone(); tokio::time::timeout( BLOCKING_TASK_TIMEOUT, state.db.run(move |db| { let mut txn = crate::db::Txn::new(db); for cursor in cursors { stage_cursor(&mut txn.batch, db, cursor); } txn.commit() }), ) .await .map_err(|_| miette::miette!("cursor-only commit timed out"))??; Ok(()) }}
fn stage_cursor(batch: &mut fjall::OwnedWriteBatch, db: &crate::db::Db, cursor: CursorUpdate) { match cursor.value { Some(value) => batch.insert(&db.cursors, cursor.key, value), None => batch.remove(&db.cursors, cursor.key), }}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]enum ReconcileOutcome { Unchanged, StateOnly, AccountChanged,}
impl ReconcileOutcome { fn emits_account(self) -> bool { self == Self::AccountChanged }}
fn reconcile_listing( txn: &mut crate::db::Txn<'_>, listing: &RepoListing,) -> Result<ReconcileOutcome> { // production callers pre-acquire this listing's one lock group; this local // check keeps direct callers on the same record-lock -> lifecycle hierarchy. let db = txn.db; if txn.lock_repo_and_is_excluded(&listing.did)? { return Ok(ReconcileOutcome::Unchanged); } let did_key = keys::repo_key(&listing.did); let Some(state_bytes) = db.repos.get(&did_key).into_diagnostic()? else { if listing.active || !listing.create_if_missing { return Ok(ReconcileOutcome::Unchanged); } let status = listed_repo_status(listing.status.as_ref()); let mut state = RepoState::backfilling(); state.active = false; state.status = status.clone(); if status != RepoStatus::Deleted { txn.batch.insert( &db.indexer.resync, &did_key, rmp_serde::to_vec(&ResyncState::Gone { status: status.clone(), }) .into_diagnostic()?, ); txn.transition_lifecycle(&listing.did, GaugeState::Resync(None))?; } txn.batch .insert(&db.repos, &did_key, ser_repo_state(&state)?); txn.counts.add_repos(1); return Ok(ReconcileOutcome::StateOnly); }; let mut state = crate::db::deser_repo_state(&state_bytes)?.into_static();
if listing.active { let listing_is_newer = state .root .as_ref() .is_none_or(|root| listing.rev.as_bytes() > root.rev.as_bytes()); if state.active && !listing_is_newer { return Ok(ReconcileOutcome::Unchanged); } if listing_is_newer && listing_is_pending(db, &listing.did)? { return Ok(ReconcileOutcome::Unchanged); }
let was_active = state.active; txn.batch.remove(&db.indexer.resync, &did_key); if !listing_is_newer { txn.transition_lifecycle(&listing.did, GaugeState::Synced)?; state.status = RepoStatus::Synced; } else { enqueue_listing_backfill(txn, &listing.did, &did_key)?; txn.transition_lifecycle(&listing.did, GaugeState::Pending)?; state.status = RepoStatus::Desynchronized; } state.active = true; state.touch(); txn.batch .insert(&db.repos, did_key, ser_repo_state(&state)?); return Ok(if was_active { ReconcileOutcome::StateOnly } else { ReconcileOutcome::AccountChanged }); }
let status = listed_repo_status(listing.status.as_ref()); let account_changed = state.active || state.status != status; if !account_changed && !listing_is_pending(db, &listing.did)? { return Ok(ReconcileOutcome::Unchanged); }
txn.transition_lifecycle( &listing.did, if status == RepoStatus::Deleted { GaugeState::Synced } else { GaugeState::Resync(None) }, )?; if status == RepoStatus::Deleted { crate::ops::delete_repo(&mut *txn, db, &listing.did, &state)?; state.root = None; } else { if let Some(metadata_bytes) = db .repo_metadata .get(keys::repo_metadata_key(&listing.did)) .into_diagnostic()? { let metadata = crate::db::deser_repo_meta(&metadata_bytes)?; txn.batch .remove(&db.indexer.pending, keys::pending_key(metadata.index_id)); } txn.batch.insert( &db.indexer.resync, &did_key, rmp_serde::to_vec(&ResyncState::Gone { status: status.clone(), }) .into_diagnostic()?, ); }
state.active = false; state.status = status; state.touch(); txn.batch .insert(&db.repos, did_key, ser_repo_state(&state)?); Ok(if account_changed { ReconcileOutcome::AccountChanged } else { ReconcileOutcome::StateOnly })}
fn listing_is_pending(db: &crate::db::Db, did: &Did) -> Result<bool> { let Some(metadata_bytes) = db .repo_metadata .get(keys::repo_metadata_key(did)) .into_diagnostic()? else { return Ok(false); }; let metadata = crate::db::deser_repo_meta(&metadata_bytes)?; Ok(db .indexer .pending .get(keys::pending_key(metadata.index_id)) .into_diagnostic()? .is_some())}
fn enqueue_listing_backfill(txn: &mut crate::db::Txn<'_>, did: &Did, did_key: &[u8]) -> Result<()> { let db = txn.db; let metadata_key = keys::repo_metadata_key(did); let existing = db .repo_metadata .get(&metadata_key) .into_diagnostic()? .map(|bytes| crate::db::deser_repo_meta(&bytes)) .transpose()?; if let Some(metadata) = &existing { txn.batch .remove(&db.indexer.pending, keys::pending_key(metadata.index_id)); } let mut metadata = existing.unwrap_or_else(|| RepoMetadata::backfilling(rand::random())); metadata.index_id = rand::random(); txn.batch.insert( &db.repo_metadata, metadata_key, crate::db::ser_repo_meta(&metadata)?, ); txn.batch.insert( &db.indexer.pending, keys::pending_key(metadata.index_id), did_key, ); Ok(())}
fn listed_repo_status(status: Option<&ListedRepoStatus>) -> RepoStatus { match status { Some(ListedRepoStatus::Takendown) => RepoStatus::Takendown, Some(ListedRepoStatus::Suspended) => RepoStatus::Suspended, Some(ListedRepoStatus::Deleted) => RepoStatus::Deleted, Some(ListedRepoStatus::Deactivated) => RepoStatus::Deactivated, Some(ListedRepoStatus::Desynchronized) => RepoStatus::Desynchronized, Some(ListedRepoStatus::Throttled) => RepoStatus::Throttled, Some(ListedRepoStatus::Other(status)) => RepoStatus::Error(status.to_smolstr()), None => RepoStatus::Error("inactive".into()), }}
#[cfg(test)]mod tests { use super::*; use bytes::Bytes; use miette::IntoDiagnostic;
#[test] fn pending_between_resume_and_max_does_not_start_throttling() { assert_eq!( capacity_decision(75, 100, 50, false), CapacityDecision::Proceed ); assert_eq!( capacity_decision(100, 100, 50, false), CapacityDecision::Proceed ); }
#[test] fn pending_above_max_starts_throttling() { assert_eq!( capacity_decision(101, 100, 50, false), CapacityDecision::EnterThrottle ); }
#[test] fn throttling_stays_active_until_resume_boundary() { assert_eq!( capacity_decision(100, 100, 50, true), CapacityDecision::StayThrottled ); assert_eq!( capacity_decision(51, 100, 50, true), CapacityDecision::StayThrottled ); assert_eq!( capacity_decision(50, 100, 50, true), CapacityDecision::ReleaseThrottle ); assert_eq!( capacity_decision(0, 100, 50, true), CapacityDecision::ReleaseThrottle ); }
fn did() -> Did { Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap() }
fn rev(value: u8) -> DbTid { let mut bytes = [0; 8]; bytes[7] = value; DbTid::new_from_bytes(bytes) }
fn state_with_root(rev: DbTid) -> RepoState<'static> { let mut state = RepoState::synced(); state.root = Some(crate::types::Commit { version: 3, rev, data: jacquard_repo::mst::util::compute_cid(b"root").unwrap(), prev: None, sig: Bytes::new(), }); state }
fn open_db() -> Result<(tempfile::TempDir, crate::db::Db)> { let tmp = tempfile::tempdir().into_diagnostic()?; let config = crate::config::Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; let db = crate::db::Db::open(&config)?; Ok((tmp, db)) }
fn insert_repo( db: &crate::db::Db, state: &RepoState<'_>, index_id: u64, pending: bool, ) -> Result<()> { let did = did(); let did_key = keys::repo_key(&did); db.repos .insert(&did_key, ser_repo_state(state)?) .into_diagnostic()?; db.repo_metadata .insert( keys::repo_metadata_key(&did), crate::db::ser_repo_meta(&RepoMetadata::backfilling(index_id))?, ) .into_diagnostic()?; if pending { db.indexer .pending .insert(keys::pending_key(index_id), did_key) .into_diagnostic()?; } Ok(()) }
fn apply(db: &crate::db::Db, listing: &RepoListing) -> Result<ReconcileOutcome> { let mut txn = crate::db::Txn::new(db); let changed = reconcile_listing(&mut txn, listing)?; txn.commit()?; Ok(changed) }
fn read_state(db: &crate::db::Db) -> Result<RepoState<'static>> { let bytes = db .repos .get(keys::repo_key(&did())) .into_diagnostic()? .expect("repo state should exist"); Ok(crate::db::deser_repo_state(&bytes)?.into_static()) }
#[test] fn inactive_listing_removes_pending_backfill() -> Result<()> { let (_tmp, db) = open_db()?; insert_repo(&db, &RepoState::backfilling(), 7, true)?; let listing = RepoListing { did: did(), rev: rev(1), active: false, status: Some(ListedRepoStatus::Deactivated), create_if_missing: false, };
assert_eq!(apply(&db, &listing)?, ReconcileOutcome::AccountChanged); assert!( db.indexer .pending .get(keys::pending_key(7)) .into_diagnostic()? .is_none() ); let state = read_state(&db)?; assert!(!state.active); assert_eq!(state.status, RepoStatus::Deactivated); assert_eq!(apply(&db, &listing)?, ReconcileOutcome::Unchanged); Ok(()) }
#[test] fn relay_first_persists_unknown_inactive_listing_without_backfill() -> Result<()> { let (_tmp, db) = open_db()?; let listing = RepoListing { did: did(), rev: rev(1), active: false, status: Some(ListedRepoStatus::Takendown), create_if_missing: true, };
assert_eq!(apply(&db, &listing)?, ReconcileOutcome::StateOnly); let state = read_state(&db)?; assert!(!state.active); assert_eq!(state.status, RepoStatus::Takendown); assert_eq!(db.indexer.pending.iter().count(), 0); assert!( db.indexer .resync .contains_key(keys::repo_key(&listing.did)) .into_diagnostic()? ); assert_eq!(apply(&db, &listing)?, ReconcileOutcome::Unchanged); Ok(()) }
#[test] fn current_active_listing_does_not_reenqueue() -> Result<()> { let (_tmp, db) = open_db()?; insert_repo(&db, &state_with_root(rev(2)), 7, false)?; let listing = RepoListing { did: did(), rev: rev(2), active: true, status: None, create_if_missing: false, };
assert_eq!(apply(&db, &listing)?, ReconcileOutcome::Unchanged); assert_eq!(db.indexer.pending.iter().count(), 0); Ok(()) }
#[test] fn unchanged_inactive_listing_repairs_stale_pending_entry() -> Result<()> { let (_tmp, db) = open_db()?; let mut state = RepoState::backfilling(); state.active = false; state.status = RepoStatus::Deactivated; insert_repo(&db, &state, 7, true)?; let listing = RepoListing { did: did(), rev: rev(1), active: false, status: Some(ListedRepoStatus::Deactivated), create_if_missing: false, };
assert_eq!(apply(&db, &listing)?, ReconcileOutcome::StateOnly); assert_eq!(db.indexer.pending.iter().count(), 0); Ok(()) }
#[test] fn newer_active_listing_reenqueues_missed_revision() -> Result<()> { let (_tmp, db) = open_db()?; insert_repo(&db, &state_with_root(rev(1)), 7, false)?; let listing = RepoListing { did: did(), rev: rev(2), active: true, status: None, create_if_missing: false, };
assert_eq!(apply(&db, &listing)?, ReconcileOutcome::StateOnly); assert_eq!(db.indexer.pending.iter().count(), 1); let state = read_state(&db)?; assert!(state.active); assert_eq!(state.status, RepoStatus::Desynchronized); Ok(()) }
#[test] fn terminal_cursor_update_clears_resume_point() -> Result<()> { let (_tmp, db) = open_db()?; let key = b"crawler_cursor|relay".to_vec(); db.cursors.insert(&key, b"old").into_diagnostic()?; let mut txn = crate::db::Txn::new(&db); stage_cursor( &mut txn.batch, &db, CursorUpdate { key: key.clone(), value: None, }, ); txn.commit()?; assert!(db.cursors.get(key).into_diagnostic()?.is_none()); Ok(()) }
#[test] fn cursor_batch_can_advance_one_marker_and_clear_another() -> Result<()> { let (_tmp, db) = open_db()?; let old_key = b"relay_first_cursor|relay|hosts".to_vec(); let marker_key = b"relay_first_cursor|relay|handled".to_vec(); db.cursors.insert(&old_key, b"old").into_diagnostic()?; let mut txn = crate::db::Txn::new(&db); for cursor in [ CursorUpdate { key: marker_key.clone(), value: Some(7u64.to_be_bytes().to_vec()), }, CursorUpdate { key: old_key.clone(), value: None, }, ] { stage_cursor(&mut txn.batch, &db, cursor); } txn.commit()?;
assert_eq!( db.cursors.get(marker_key).into_diagnostic()?.as_deref(), Some(7u64.to_be_bytes().as_slice()) ); assert!(db.cursors.get(old_key).into_diagnostic()?.is_none()); Ok(()) }
#[tokio::test] #[cfg(feature = "indexer_stream")] async fn changed_listing_emits_one_account_event() -> Result<()> { let tmp = tempfile::tempdir().into_diagnostic()?; let config = crate::config::Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; let state = Arc::new(AppState::new(&config)?); insert_repo(&state.db, &RepoState::backfilling(), 7, true)?; let stats = CrawlerStats::new( state.clone(), Vec::new(), crate::util::throttle::Throttler::new(1, 1), ); let (mut worker, _tx) = CrawlerWorker::new(state.clone(), 100, 50, stats); let mut events = state.db.stream.event_tx.subscribe(); let listing = RepoListing { did: did(), rev: rev(1), active: false, status: Some(ListedRepoStatus::Deactivated), create_if_missing: false, };
worker .enqueue(CrawlerBatch { guards: Vec::new(), listings: vec![listing.clone()], cursor_updates: Vec::new(), }) .await?; let crate::types::BroadcastEvent::Ephemeral(event) = events.recv().await.unwrap() else { panic!("expected account event"); }; let account = event.account.expect("account payload"); assert!(!account.active); assert_eq!(account.status.as_deref(), Some("deactivated"));
worker .enqueue(CrawlerBatch { guards: Vec::new(), listings: vec![listing], cursor_updates: Vec::new(), }) .await?; assert!(events.try_recv().is_err()); Ok(()) }
#[tokio::test] #[cfg(feature = "indexer_stream")] async fn missed_revision_does_not_emit_account_event() -> Result<()> { let tmp = tempfile::tempdir().into_diagnostic()?; let config = crate::config::Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; let state = Arc::new(AppState::new(&config)?); insert_repo(&state.db, &state_with_root(rev(1)), 7, false)?; let stats = CrawlerStats::new( state.clone(), Vec::new(), crate::util::throttle::Throttler::new(1, 1), ); let (mut worker, _tx) = CrawlerWorker::new(state.clone(), 100, 50, stats); let mut events = state.db.stream.event_tx.subscribe();
worker .enqueue(CrawlerBatch { guards: Vec::new(), listings: vec![RepoListing { did: did(), rev: rev(2), active: true, status: None, create_if_missing: false, }], cursor_updates: Vec::new(), }) .await?;
assert!(events.try_recv().is_err()); assert_eq!(state.db.indexer.pending.iter().count(), 1); Ok(()) }}