442 unmodified lines
Expand all
let (
Applied {
added,
changed_listings,
reconciled,
..
},
failed,
) = apply_on_owners(&self.state, listings, dids).await;
drop(guards);
if count > 0 {
self.stats.record_processed(count);
}
if count > 0 || reconciled > 0 {
self.state.notify_backfill();
}
#[cfg(feature = "indexer_stream")]
/// listings that changed whether the account is active
changed_listings: Vec<RepoListing>,
/// deleted accounts whose wipe a listing started
wipes: Vec<Did>,
/// listings that changed their repo, a queued backfill among them
reconciled: usize,
}
/// one write a page asks of a repo's owner shard
#[derive(Clone)]
Ok(batch) => {
applied.added.extend(batch.added);
applied.changed_listings.extend(batch.changed_listings);
applied.wipes.extend(batch.wipes);
applied.reconciled += batch.reconciled;
}
Err(e) => failed = Some(e),
}
}
for listing in listings {
let mut txn = crate::db::Txn::new(db);
let outcome = reconcile_listing(&mut txn, listing)?;
txn.commit()?;
if outcome != ReconcileOutcome::Unchanged && listing.starts_a_wipe() {
applied.wipes.push(listing.did.clone());
if outcome != ReconcileOutcome::Unchanged {
applied.reconciled += 1;
if listing.starts_a_wipe() {
applied.wipes.push(listing.did.clone());
}
}
if outcome.emits_account() {
applied.changed_listings.push(listing.clone());
}
473 unmodified lines
Expand all
Ok(())
}
#[tokio::test]
async fn a_listing_that_queues_a_backfill_wakes_the_backfill_worker() -> Result<()> {
let (_tmp, state, tx) = crawl_over_a_synced_repo()?;
tx.send(CrawlerBatch {
guards: Vec::new(),
listings: newer_rev(),
cursor_updates: Vec::new(),
applied: None,
})
.await
.into_diagnostic()?;
tokio::time::timeout(Duration::from_secs(5), state.backfill_notify.notified())
.await
.into_diagnostic()?;
assert_eq!(state.db.indexer.pending_iter().count(), 1);
Ok(())
}
#[tokio::test]
async fn a_newer_rev_the_firehose_brought_in_while_it_waited_queues_nothing() -> Result<()> {
let (_tmp, state, tx) = crawl_over_a_synced_repo()?;
let relay = relay_firehose(&state, "relay.example", 500);
let waiting = WaitingPages::spawn(tx.clone(), state.clone(), QUICK, WAITING_PAGES);
744 unmodified lines
Expand all