//! 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());
}
}