Something went wrong. Try again.
very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
rust fjall at-protocol atproto indexer
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302//! the writes a full backfill makes: the new repo's records merged against the repo's stored//! records, both read in key order, so neither side is ever held whole.
use std::cmp::Ordering;use std::iter::Fuse;
use cid::Cid as IpldCid;use miette::Result;use crate::mst::MstKey;use tracing::trace;
/// one write that brings the stored records in line with the new repo#[derive(Debug)]pub(crate) enum RecordOp<K, T> { /// a record that `stored` does not have Create { key: MstKey, record: T }, /// a record whose cid differs from the stored one Update { key: MstKey, record: T }, /// a stored record the new repo doesn't have Delete(K),}
/// the writes that turn `stored` into `new`, in key order. a record whose cid didn't change/// is left alone.////// `new` yields each record's mst key, its cid and whatever the caller needs to write it./// `stored` yields each stored record's key after the did prefix and its cid. a heads key/// ends in the same `collection/rkey` text as the mst key, so both sides sort the same way.pub(crate) fn diff<N, S, K, T>(new: N, stored: S) -> Diff<N, S, K, T>where N: Iterator<Item = Result<(MstKey, IpldCid, T)>>, S: Iterator<Item = Result<(K, IpldCid)>>, K: AsRef<[u8]>,{ Diff { new: new.fuse(), stored: stored.fuse(), next_new: None, next_stored: None, last_new: None, failed: false, }}
pub(crate) struct Diff<N: Iterator, S: Iterator, K, T> { new: Fuse<N>, stored: Fuse<S>, next_new: Option<(MstKey, IpldCid, T)>, next_stored: Option<(K, IpldCid)>, last_new: Option<MstKey>, failed: bool,}
impl<N, S, K, T> Diff<N, S, K, T>where N: Iterator<Item = Result<(MstKey, IpldCid, T)>>, S: Iterator<Item = Result<(K, IpldCid)>>, K: AsRef<[u8]>,{ fn step(&mut self) -> Result<Option<RecordOp<K, T>>> { loop { if self.next_new.is_none() { self.next_new = self.pull_new()?; } if self.next_stored.is_none() { self.next_stored = self.stored.next().transpose()?; } let (new, stored) = match (self.next_new.take(), self.next_stored.take()) { (None, None) => return Ok(None), (Some((key, _, record)), None) => { return Ok(Some(RecordOp::Create { key, record })); } (None, Some((stored, _))) => return Ok(Some(RecordOp::Delete(stored))), (Some(new), Some(stored)) => (new, stored), }; match new.0.as_bytes().cmp(stored.0.as_ref()) { Ordering::Less => { self.next_stored = Some(stored); let (key, _, record) = new; return Ok(Some(RecordOp::Create { key, record })); } Ordering::Greater => { self.next_new = Some(new); return Ok(Some(RecordOp::Delete(stored.0))); } Ordering::Equal if new.1 != stored.1 => { let (key, _, record) = new; return Ok(Some(RecordOp::Update { key, record })); } Ordering::Equal => trace!(key = %new.0, cid = %new.1, "skip unchanged record"), } } }
/// the next new record. a merge against records out of order would delete and recreate /// them, so that's an error instead fn pull_new(&mut self) -> Result<Option<(MstKey, IpldCid, T)>> { let Some(next) = self.new.next().transpose()? else { return Ok(None); }; if let Some(last) = self.last_new.as_ref().filter(|last| **last >= next.0) { miette::bail!("record {} does not sort after {last}", next.0); } self.last_new = Some(next.0.clone()); Ok(Some(next)) }}
impl<N, S, K, T> Iterator for Diff<N, S, K, T>where N: Iterator<Item = Result<(MstKey, IpldCid, T)>>, S: Iterator<Item = Result<(K, IpldCid)>>, K: AsRef<[u8]>,{ type Item = Result<RecordOp<K, T>>;
fn next(&mut self) -> Option<Self::Item> { if self.failed { return None; } let step = self.step(); self.failed = step.is_err(); step.transpose() }}
#[cfg(test)]mod tests { use std::collections::HashMap;
use smol_str::{SmolStr, ToSmolStr};
use super::*; use crate::db::keys; use crate::db::types::{DbAction, DbRkey, DbTid}; use crate::mst::tests::cid;
/// every stored record in a map, each new record taken out of it, and whatever stays behind /// deleted. returns (key, action) in key order fn oracle( new: &[(String, IpldCid)], stored: &[(String, IpldCid)], ) -> Vec<(String, &'static str)> { let mut existing: HashMap<(SmolStr, DbRkey), IpldCid> = stored .iter() .map(|(key, cid)| { let (collection, rkey) = keys::split_record_suffix(key.as_bytes()).unwrap(); ((collection.into(), rkey), *cid) }) .collect(); let mut ops = Vec::new(); for (key, cid) in new { let (collection, rkey) = crate::ops::parse_path(key).unwrap(); let path = (SmolStr::new(collection), DbRkey::new(rkey)); let action = existing .remove(&path) .map_or(Some(DbAction::Create), |existing| { (existing != *cid).then_some(DbAction::Update) }); if let Some(action) = action { ops.push((key.clone(), action.as_str())); } } ops.extend( existing.into_keys().map(|(collection, rkey)| { (format!("{collection}/{}", rkey.to_smolstr()), "delete") }), ); ops.sort(); ops }
fn merged( new: &[(String, IpldCid)], stored: &[(String, IpldCid)], ) -> Vec<(String, &'static str)> { let new = new .iter() .map(|(key, cid)| Ok((MstKey::new(key), *cid, ()))); let stored = stored.iter().map(|(key, cid)| Ok((key.clone(), *cid))); diff(new, stored) .map(|op| match op.unwrap() { RecordOp::Create { key, .. } => (key.to_string(), "create"), RecordOp::Update { key, .. } => (key.to_string(), "update"), RecordOp::Delete(key) => (key, "delete"), }) .collect() }
/// xorshift, so a failing case can be rerun from its seed struct Rng(u64);
impl Rng { fn next(&mut self) -> u64 { self.0 ^= self.0 << 13; self.0 ^= self.0 >> 7; self.0 ^= self.0 << 17; self.0 }
fn below(&mut self, n: usize) -> usize { (self.next() % n as u64) as usize } }
#[test] fn the_merge_makes_the_same_writes_as_the_map_it_replaced() { // collections where one is a prefix of another or differs by a char either side of // `/`, which is where byte order and (collection, rkey) order could part let collections = [ "a.b.c", "a.b.c.d", "a.b.cd", "a.b-c.d", "app.bsky.feed.post", ]; let strings = [ "self", "a", "a.b", "a-b", "a_b", "~", "z:z", "3k", "2222222222222x", ]; for seed in 1..=2000u64 { let mut rng = Rng(seed.wrapping_mul(0x9e37_79b9_7f4a_7c15)); let tids = (0..4) .map(|_| DbTid::new_from_bytes((rng.next() >> 1).to_be_bytes()).to_smolstr()) .collect::<Vec<_>>(); let mut keys = (0..12) .map(|_| { let collection = collections[rng.below(collections.len())]; let rkey = match rng.below(2) { 0 => tids[rng.below(tids.len())].to_string(), _ => strings[rng.below(strings.len())].to_string(), }; format!("{collection}/{rkey}") }) .collect::<Vec<_>>(); keys.sort(); keys.dedup(); // either side is empty now and then, and cids come from a few values so // unchanged records are common let mut pick = |odds: usize| { let mut picked = Vec::new(); for key in &keys { if odds > 0 && rng.below(odds + 1) > 0 { picked.push((key.clone(), cid(rng.below(3) as u8))); } } picked }; let new = pick(seed as usize % 4); let stored = pick(seed as usize % 3);
let merged = merged(&new, &stored); assert_eq!(merged, oracle(&new, &stored), "seed {seed}"); assert!( merged.windows(2).all(|pair| pair[0].0 < pair[1].0), "seed {seed}: {merged:?}" ); } }
#[test] fn deletes_land_between_the_puts_in_key_order() { let new = [ ("a.b.c/1".to_string(), cid(1)), ("a.b.c/3".to_string(), cid(1)), ]; let stored = [ ("a.b.c/2".to_string(), cid(1)), ("a.b.c/3".to_string(), cid(2)), ]; assert_eq!( merged(&new, &stored), [ ("a.b.c/1".to_string(), "create"), ("a.b.c/2".to_string(), "delete"), ("a.b.c/3".to_string(), "update"), ] ); }
#[test] fn new_records_out_of_order_are_an_error() { let new = [("a.b.c/2", cid(1)), ("a.b.c/1", cid(1))] .map(|(key, cid)| Ok((MstKey::new(key), cid, ()))); let mut ops = diff( new.into_iter(), std::iter::empty::<Result<(Vec<u8>, IpldCid)>>(), ); assert!(matches!(ops.next(), Some(Ok(RecordOp::Create { .. })))); let err = ops.next().unwrap().unwrap_err(); assert!(err.to_string().contains("does not sort after"), "{err}"); assert!(ops.next().is_none()); }}