grain.social/grain
photography-focused photo sharing platform
grain.social
1 collaborator
- 1 collaborator
0
2
grain-serve backfills the index from hubble, or each account's PDS
Only what this pull adds. It is built on grain runs on a Rust server modelled on knotless, without hatk.
crates/grain/src/repo/car.rsnew+168−0
| @@ -0,0 +1,168 @@ | ||
//! A repo as a CAR (CARv1), as a PDS's `com.atproto.sync.getRepo` answers: | ||
//! a header naming the commit's block, then blocks, each its CID and its | ||
//! bytes, in whatever order the PDS wrote them. | ||
//! | ||
//! Each block is held to its CID as it comes, and kept in a spill file, | ||
//! only where each is held in memory. Once all are in, the commit's | ||
//! signature is checked, and its tree walked from the root in key order, | ||
//! each record given out as [`super::star::StarLite`] gives them out and | ||
//! rebuilt into a root of its own the same way: a tree a PDS shaped other | ||
//! than atproto would is refused, whatever its blocks hash to. | ||
| ||
use super::mst::{Node, RootBuilder}; | ||
use super::{Commit, Frames, MAX_RECORD, Record, Spill, verify_commit}; | ||
use crate::crypto::PublicKey; | ||
use crate::drisl; | ||
use anyhow::{Context, Result, ensure}; | ||
use ciborium::Value as V; | ||
use sha2::{Digest, Sha256}; | ||
use std::collections::HashMap; | ||
use std::io::{BufWriter, Read, Seek, SeekFrom, Write}; | ||
| ||
/// The most a CAR's header may be. | ||
const MAX_HEADER: usize = 4096; | ||
| ||
/// The most a CAR's blocks may come to in all, spilled: past any repo | ||
/// the network has. | ||
pub const MAX_CAR: u64 = 16 << 30; | ||
| ||
/// The CAR's blocks, in the spill, by CID: where each starts and how long. | ||
struct Blocks { | ||
spill: Box<dyn Spill>, | ||
at: HashMap<Vec<u8>, (u64, usize)>, | ||
} | ||
| ||
impl Blocks { | ||
fn get(&mut self, cid: &[u8]) -> Result<Vec<u8>> { | ||
let what = || format!("a CAR without the block {}", drisl::cid_text(cid)); | ||
let &(from, length) = self.at.get(cid).with_context(what)?; | ||
self.spill.seek(SeekFrom::Start(from))?; | ||
let mut block = vec![0; length]; | ||
self.spill.read_exact(&mut block)?; | ||
Ok(block) | ||
} | ||
} | ||
| ||
/// What is left of the tree to walk, last first. | ||
enum Next { | ||
Node(Vec<u8>), | ||
Record(String, Vec<u8>), | ||
} | ||
| ||
/// A CAR being read. | ||
pub struct Car { | ||
blocks: Blocks, | ||
commit: Commit, | ||
walk: Vec<Next>, | ||
tree: RootBuilder, | ||
} | ||
| ||
impl Car { | ||
/// Reads `did`'s CAR from `r`, its blocks into `spill`, and checks that | ||
/// `key` signed the commit it names. | ||
pub fn read(r: impl Read, did: &str, key: &PublicKey, spill: Box<dyn Spill>) -> Result<Car> { | ||
let mut frames = Frames::new(r); | ||
let header = frames | ||
.frame(MAX_HEADER, "a CAR header")? | ||
.unwrap_or_default(); | ||
let header: V = ciborium::from_reader(&header[..]).context("not a CAR")?; | ||
let field = |name: &str| { | ||
let pairs = header.as_map()?; | ||
pairs | ||
.iter() | ||
.find(|(k, _)| k.as_text() == Some(name)) | ||
.map(|(_, v)| v) | ||
}; | ||
let version = field("version").and_then(V::as_integer); | ||
ensure!(version == Some(1.into()), "not a CAR of version 1"); | ||
let root = field("roots").and_then(|r| drisl::link(r.as_array()?.first()?)); | ||
let root = root.context("a CAR with no root")?.to_vec(); | ||
let mut spill = BufWriter::new(spill); | ||
let (mut at, mut end) = (HashMap::new(), 0); | ||
while let Some(block) = frames.frame(MAX_RECORD + 36, "a CAR block")? { | ||
ensure!( | ||
block.len() >= 36 | ||
&& block[0] == 1 | ||
&& [0x71, 0x55].contains(&block[1]) | ||
&& block[2..4] == [0x12, 0x20], | ||
"a CAR block whose CID is not atproto's" | ||
); | ||
let (cid, bytes) = block.split_at(36); | ||
ensure!( | ||
Sha256::digest(bytes)[..] == cid[4..], | ||
"a CAR block that is not what its CID says" | ||
); | ||
ensure!( | ||
end + bytes.len() as u64 <= MAX_CAR, | ||
"a CAR of more than {MAX_CAR} bytes" | ||
); | ||
spill.write_all(bytes)?; | ||
at.insert(cid.to_vec(), (end, bytes.len())); | ||
end += bytes.len() as u64; | ||
} | ||
let spill = spill.into_inner().map_err(|e| e.into_error())?; | ||
let mut blocks = Blocks { spill, at }; | ||
let commit: V = | ||
ciborium::from_reader(&blocks.get(&root)?[..]).context("a commit that is not CBOR")?; | ||
let commit = verify_commit(commit, did, key)?; | ||
Ok(Car { | ||
blocks, | ||
walk: vec![Next::Node(commit.data.clone())], | ||
commit, | ||
tree: RootBuilder::new(), | ||
}) | ||
} | ||
| ||
pub fn commit(&self) -> &Commit { | ||
&self.commit | ||
} | ||
| ||
/// The next record, its key and its DRISL; none at the end. | ||
pub fn record(&mut self) -> Result<Option<Record>> { | ||
while let Some(next) = self.walk.pop() { | ||
match next { | ||
Next::Node(cid) => { | ||
let node = Node::decode(&self.blocks.get(&cid)?)?; | ||
for entry in node.entries.into_iter().rev() { | ||
self.walk.extend(entry.right.map(Next::Node)); | ||
self.walk.push(Next::Record(entry.key, entry.value)); | ||
} | ||
self.walk.extend(node.left.map(Next::Node)); | ||
} | ||
Next::Record(key, cid) => { | ||
let record = self.blocks.get(&cid)?; | ||
self.tree.add(&key, &record)?; | ||
return Ok(Some((key, record))); | ||
} | ||
} | ||
} | ||
Ok(None) | ||
} | ||
| ||
/// The commit, once the records read make the tree it signed. | ||
pub fn finish(self) -> Result<Commit> { | ||
let root = self.tree.root(); | ||
ensure!( | ||
root == self.commit.data, | ||
"the records do not make the tree the commit signed" | ||
); | ||
Ok(self.commit) | ||
} | ||
} | ||
| ||
/// Writes a CAR of `blocks` naming `root` to `out`, each block's CID | ||
/// worked out from it as DRISL. | ||
pub fn write(out: &mut impl Write, root: &[u8], blocks: &[Vec<u8>]) -> std::io::Result<()> { | ||
let header = V::Map(vec![ | ||
( | ||
V::Text("roots".into()), | ||
V::Array(vec![drisl::link_to(root)]), | ||
), | ||
(V::Text("version".into()), V::Integer(1.into())), | ||
]); | ||
super::star::frame(out, &drisl::encode(&header))?; | ||
for block in blocks { | ||
super::star::frame(out, &[drisl::cid(block), block.clone()].concat())?; | ||
} | ||
Ok(()) | ||
} | ||
crates/grain/src/repo/mod.rsnew+229−0
| @@ -0,0 +1,229 @@ | ||
//! A whole repo, read as an export and checked as it is read: every record | ||
//! of it in key order, the tree's root rebuilt from them and held to the | ||
//! root the account's commit names, and the commit held to the account's | ||
//! signing key. | ||
//! | ||
//! An export is STAR-lite ([`star`]), as microcosm's hubble serves it, or | ||
//! a CAR ([`car`]), as any PDS's `com.atproto.sync.getRepo` does; zstd it | ||
//! may come in. [`open`] tells which by its first bytes. STAR-lite streams: | ||
//! a record at a time, nothing of the repo held but one tree node a layer. | ||
//! A CAR's blocks come in no order a reader can rely on, so they are kept | ||
//! in a spill file, a few bytes each held in memory, and walked from its | ||
//! commit once all are in. | ||
//! | ||
//! The parts are grain's own and depend on nothing of the rest of it but | ||
//! [`crate::drisl`] and [`crate::crypto`], for knotless to take back. | ||
| ||
pub mod car; | ||
pub mod mst; | ||
pub mod star; | ||
| ||
use crate::crypto::PublicKey; | ||
use crate::drisl; | ||
use anyhow::{Context, Result, anyhow, ensure}; | ||
use ciborium::Value as V; | ||
use std::io::{BufReader, Read, Seek, Write}; | ||
| ||
/// The most a record's key may be: a collection's 317 bytes, a slash and | ||
/// a record key's 512. | ||
pub const MAX_KEY: usize = 830; | ||
/// The most a record may be, as atproto recommends. | ||
pub const MAX_RECORD: usize = 1_048_576; | ||
/// The most a commit may be. | ||
pub const MAX_COMMIT: usize = 4096; | ||
| ||
/// The bytes a zstd frame starts with. | ||
pub const ZSTD: [u8; 4] = [0x28, 0xb5, 0x2f, 0xfd]; | ||
| ||
/// A record as a repo gives it out: its key, `collection/rkey`, and its | ||
/// DRISL. | ||
pub type Record = (String, Vec<u8>); | ||
| ||
/// An account's commit, checked: whose repo it is, at which revision, and | ||
/// the root of the tree it signed. | ||
#[derive(Debug, Clone, PartialEq)] | ||
pub struct Commit { | ||
pub did: String, | ||
pub rev: String, | ||
/// The tree's root, a CID as bytes. | ||
pub data: Vec<u8>, | ||
/// The commit as signed, less `data`: as STAR-lite carries it. | ||
pub partial: V, | ||
} | ||
| ||
/// `commit`, a commit's DRISL map, if it is `did`'s and `key` signed it. | ||
pub fn verify_commit(commit: V, did: &str, key: &PublicKey) -> Result<Commit> { | ||
let V::Map(mut fields) = commit else { | ||
return Err(anyhow!("a commit that is not a map")); | ||
}; | ||
let at = | ||
|fields: &[(V, V)], name: &str| fields.iter().position(|(k, _)| k.as_text() == Some(name)); | ||
let sig = at(&fields, "sig").map(|i| fields.remove(i).1); | ||
let sig = sig | ||
.and_then(|s| s.into_bytes().ok()) | ||
.context("a commit with no signature")?; | ||
let text = |name: &str| { | ||
at(&fields, name) | ||
.and_then(|i| fields[i].1.as_text()) | ||
.map(str::to_string) | ||
}; | ||
let theirs = text("did").context("a commit with no DID")?; | ||
ensure!(theirs == did, "a commit for {theirs}, not {did}"); | ||
let rev = text("rev").filter(|r| crate::lexicon::valid_tid(r)); | ||
let rev = rev.context("a commit with no revision")?; | ||
let data = at(&fields, "data").and_then(|i| drisl::link(&fields[i].1).map(<[u8]>::to_vec)); | ||
let data = data.context("a commit with no tree")?; | ||
let signed = drisl::encode(&V::Map(fields.clone())); | ||
key.verify(&signed, &sig) | ||
.context("a commit the account's key did not sign")?; | ||
fields.retain(|(k, _)| k.as_text() != Some("data")); | ||
fields.push((V::Text("sig".into()), V::Bytes(sig))); | ||
Ok(Commit { | ||
did: theirs, | ||
rev, | ||
data, | ||
partial: V::Map(fields), | ||
}) | ||
} | ||
| ||
/// A stream of frames, each an unsigned varint length and that many | ||
/// bytes, as STAR-lite and CAR both write them (and knotless's STAR v1). | ||
pub struct Frames<R> { | ||
r: BufReader<R>, | ||
} | ||
| ||
impl<R: Read> Frames<R> { | ||
pub fn new(r: R) -> Frames<R> { | ||
Frames { | ||
r: BufReader::new(r), | ||
} | ||
} | ||
| ||
/// The next length, in its shortest form: seven bits a byte, least | ||
/// significant first, the high bit set on all but the last. None | ||
/// where the stream ends before it. | ||
pub fn length(&mut self) -> Result<Option<usize>> { | ||
let (mut length, mut i) = (0, 0); | ||
loop { | ||
ensure!(i < 6, "a length too long"); | ||
let Some(byte) = self.r.by_ref().bytes().next().transpose()? else { | ||
ensure!(i == 0, "it ends inside a length"); | ||
return Ok(None); | ||
}; | ||
length += usize::from(byte & 0x7f) << (7 * i); | ||
if byte < 0x80 { | ||
ensure!(byte != 0 || i == 0, "a length not in its shortest form"); | ||
return Ok(Some(length)); | ||
} | ||
i += 1; | ||
} | ||
} | ||
| ||
/// Exactly `n` bytes, of `what`. | ||
pub fn exact(&mut self, n: usize, what: &str) -> Result<Vec<u8>> { | ||
let mut bytes = vec![0; n]; | ||
self.r.read_exact(&mut bytes).map_err(|e| match e.kind() { | ||
std::io::ErrorKind::UnexpectedEof => anyhow!("it ends inside {what}"), | ||
_ => e.into(), | ||
})?; | ||
Ok(bytes) | ||
} | ||
| ||
/// The next frame, of `what`, at most `max` bytes; none at the end. | ||
pub fn frame(&mut self, max: usize, what: &str) -> Result<Option<Vec<u8>>> { | ||
let Some(length) = self.length()? else { | ||
return Ok(None); | ||
}; | ||
ensure!(length <= max, "{what} of more than {max} bytes"); | ||
self.exact(length, what).map(Some) | ||
} | ||
} | ||
| ||
/// An unsigned varint, as [`Frames::length`] reads it. | ||
pub fn varint(mut n: usize) -> Vec<u8> { | ||
let mut out = Vec::new(); | ||
while n >= 0x80 { | ||
out.push((n as u8 & 0x7f) | 0x80); | ||
n >>= 7; | ||
} | ||
out.push(n as u8); | ||
out | ||
} | ||
| ||
/// Where a CAR's blocks are kept while it is read: a file, or for tests | ||
/// memory. | ||
pub trait Spill: Read + Write + Seek + Send {} | ||
| ||
impl<T: Read + Write + Seek + Send> Spill for T {} | ||
| ||
/// A body read from. | ||
pub type Body = Box<dyn Read + Send>; | ||
| ||
/// A repo export being read: its commit, checked, then its records in key | ||
/// order, then its tree's root checked by [`Archive::finish`]. Until then | ||
/// nothing read from it is proven, so nothing is to be kept for good. | ||
pub enum Archive { | ||
Star(star::StarLite<Body>), | ||
Car(car::Car), | ||
} | ||
| ||
impl Archive { | ||
/// The commit, its signature checked. | ||
pub fn commit(&self) -> &Commit { | ||
match self { | ||
Archive::Star(s) => s.commit(), | ||
Archive::Car(c) => c.commit(), | ||
} | ||
} | ||
| ||
/// The next record: its key, `<collection>/<rkey>`, and its DRISL. | ||
pub fn record(&mut self) -> Result<Option<Record>> { | ||
match self { | ||
Archive::Star(s) => s.record(), | ||
Archive::Car(c) => c.record(), | ||
} | ||
} | ||
| ||
/// The commit, once every record has been read and they make the tree | ||
/// it signed. | ||
pub fn finish(self) -> Result<Commit> { | ||
match self { | ||
Archive::Star(s) => s.finish(), | ||
Archive::Car(c) => c.finish(), | ||
} | ||
} | ||
} | ||
| ||
/// `did`'s repo, from `body`, a STAR-lite archive or a CAR, zstd'd or not, | ||
/// its commit checked against `key`. A CAR's blocks go to `spill`. | ||
pub fn open( | ||
body: impl Read + Send + 'static, | ||
did: &str, | ||
key: &PublicKey, | ||
spill: Box<dyn Spill>, | ||
) -> Result<Archive> { | ||
let (head, body) = peek(Box::new(body))?; | ||
let body: Body = if head.starts_with(&ZSTD) { | ||
Box::new(zstd::stream::read::Decoder::new(body)?) | ||
} else { | ||
body | ||
}; | ||
let (head, body) = peek(body)?; | ||
Ok(if head.starts_with(&star::MAGIC) { | ||
Archive::Star(star::StarLite::open(body, did, key)?) | ||
} else { | ||
Archive::Car(car::Car::read(body, did, key, spill)?) | ||
}) | ||
} | ||
| ||
/// The first bytes of `body`, and `body` as it was. | ||
fn peek(body: Body) -> Result<(Vec<u8>, Body)> { | ||
let mut head = Vec::new(); | ||
let mut body = body; | ||
body.by_ref().take(4).read_to_end(&mut head)?; | ||
let again = std::io::Cursor::new(head.clone()).chain(body); | ||
Ok((head, Box::new(again))) | ||
} | ||
| ||
#[cfg(test)] | ||
pub(crate) mod tests; | ||
crates/grain/src/repo/mst.rsnew+264−0
| @@ -0,0 +1,264 @@ | ||
//! A repo's Merkle Search Tree: its root rebuilt from the keys and records | ||
//! in key order, holding one node a layer at a time, and its nodes read | ||
//! back from their blocks. As opensocial's `RootBuilder` | ||
//! (`packages/group-sites/src/repoArchive.ts`), after STAR-lite's | ||
//! reference algorithm. | ||
| ||
use crate::drisl; | ||
use anyhow::{Context, Result, ensure}; | ||
use ciborium::Value as V; | ||
use sha2::{Digest, Sha256}; | ||
| ||
/// How many layers up the tree `key` sits: its SHA-256's leading zero | ||
/// bits, two to a layer, as atproto's fanout of four puts it. | ||
pub fn layer(key: &[u8]) -> u32 { | ||
let mut zeros = 0; | ||
for byte in Sha256::digest(key) { | ||
zeros += byte.leading_zeros(); | ||
if byte != 0 { | ||
break; | ||
} | ||
} | ||
zeros / 2 | ||
} | ||
| ||
/// A node's entry: a key, its record's CID, and the subtree right of it. | ||
#[derive(Debug, Clone, PartialEq)] | ||
pub struct Entry { | ||
pub key: String, | ||
pub value: Vec<u8>, | ||
pub right: Option<Vec<u8>>, | ||
} | ||
| ||
/// A tree node: the subtree left of its first entry, and its entries. | ||
#[derive(Debug, Clone, Default, PartialEq)] | ||
pub struct Node { | ||
pub left: Option<Vec<u8>>, | ||
pub entries: Vec<Entry>, | ||
} | ||
| ||
impl Node { | ||
fn is_empty(&self) -> bool { | ||
self.left.is_none() && self.entries.is_empty() | ||
} | ||
| ||
/// The node as atproto writes it: each key after the first as how much | ||
/// of the key before it it shares, and the rest. | ||
pub fn encode(&self) -> Vec<u8> { | ||
let link = |cid: &Option<Vec<u8>>| cid.as_deref().map_or(V::Null, drisl::link_to); | ||
let mut last: &[u8] = &[]; | ||
let entries = self.entries.iter().map(|e| { | ||
let key = e.key.as_bytes(); | ||
let shared = last.iter().zip(key).take_while(|(a, b)| a == b).count(); | ||
last = key; | ||
V::Map(vec![ | ||
(V::Text("k".into()), V::Bytes(key[shared..].to_vec())), | ||
(V::Text("p".into()), V::Integer((shared as u64).into())), | ||
(V::Text("t".into()), link(&e.right)), | ||
(V::Text("v".into()), drisl::link_to(&e.value)), | ||
]) | ||
}); | ||
let node = V::Map(vec![ | ||
(V::Text("e".into()), V::Array(entries.collect())), | ||
(V::Text("l".into()), link(&self.left)), | ||
]); | ||
drisl::encode(&node) | ||
} | ||
| ||
/// A node from its block, its keys made whole again. | ||
pub fn decode(block: &[u8]) -> Result<Node> { | ||
let node: V = ciborium::from_reader(block).context("a tree node that is not CBOR")?; | ||
let field = |map: &V, name: &str| { | ||
let pairs = map.as_map()?; | ||
pairs | ||
.iter() | ||
.find(|(k, _)| k.as_text() == Some(name)) | ||
.map(|(_, v)| v.clone()) | ||
}; | ||
let link = |v: Option<V>| v.as_ref().and_then(drisl::link).map(<[u8]>::to_vec); | ||
let entries = field(&node, "e").and_then(|e| e.into_array().ok()); | ||
let entries = entries.context("a tree node with no entries")?; | ||
let mut last = Vec::new(); | ||
let mut node = Node { | ||
left: link(field(&node, "l")), | ||
entries: Vec::with_capacity(entries.len()), | ||
}; | ||
for e in &entries { | ||
let shared = | ||
field(e, "p").and_then(|p| usize::try_from(i128::from(p.as_integer()?)).ok()); | ||
let suffix = field(e, "k").and_then(|k| k.into_bytes().ok()); | ||
let (shared, suffix) = shared.zip(suffix).context("a tree entry with no key")?; | ||
ensure!( | ||
shared <= last.len(), | ||
"a tree key sharing more than the key before it" | ||
); | ||
last.truncate(shared); | ||
last.extend(suffix); | ||
node.entries.push(Entry { | ||
key: String::from_utf8(last.clone()).context("a tree key that is not text")?, | ||
value: link(field(e, "v")).context("a tree entry with no record")?, | ||
right: link(field(e, "t")), | ||
}); | ||
} | ||
Ok(node) | ||
} | ||
} | ||
| ||
/// The tree's root, rebuilt from its keys and records in order. What is | ||
/// left of the stack is finished, so only one node a layer is held. | ||
#[derive(Default)] | ||
pub struct RootBuilder { | ||
stack: Vec<Node>, | ||
prev: Option<u32>, | ||
last: String, | ||
} | ||
| ||
impl RootBuilder { | ||
pub fn new() -> RootBuilder { | ||
RootBuilder::default() | ||
} | ||
| ||
/// Adds the record `record` at `key`, which must come after the last. | ||
pub fn add(&mut self, key: &str, record: &[u8]) -> Result<()> { | ||
ensure!(key > self.last.as_str(), "keys out of order at {key}"); | ||
self.last = key.to_string(); | ||
let layer = layer(key.as_bytes()); | ||
let at = layer as usize; | ||
if self.stack.len() <= at { | ||
self.stack.resize_with(at + 1, Node::default); | ||
} | ||
if self.prev.is_none_or(|prev| layer > prev) { | ||
self.fold(at); | ||
} | ||
self.stack[at].entries.push(Entry { | ||
key: key.to_string(), | ||
value: drisl::cid(record), | ||
right: None, | ||
}); | ||
self.prev = Some(layer); | ||
Ok(()) | ||
} | ||
| ||
/// Finishes every node below `layer`, each hung under the one above: | ||
/// as its left subtree, or right of its last entry. | ||
fn fold(&mut self, layer: usize) { | ||
for i in 0..layer { | ||
let node = std::mem::take(&mut self.stack[i]); | ||
if node.is_empty() { | ||
continue; | ||
} | ||
let cid = drisl::cid(&node.encode()); | ||
let parent = &mut self.stack[i + 1]; | ||
match parent.entries.last_mut() { | ||
Some(last) => last.right = Some(cid), | ||
None => parent.left = Some(cid), | ||
} | ||
} | ||
} | ||
| ||
/// The root's CID: of one empty node when no key was added, as | ||
/// atproto writes an empty repo. | ||
pub fn root(mut self) -> Vec<u8> { | ||
let top = self.stack.len().saturating_sub(1); | ||
self.fold(top); | ||
drisl::cid(&self.stack.pop().unwrap_or_default().encode()) | ||
} | ||
} | ||
| ||
#[cfg(test)] | ||
mod tests { | ||
use super::*; | ||
| ||
#[test] | ||
fn layers() { | ||
// atproto's interop vectors for the fanout of four. | ||
assert_eq!(layer(b""), 0); | ||
assert_eq!(layer(b"asdf"), 0); | ||
assert_eq!(layer(b"blue"), 1); | ||
assert_eq!(layer(b"2653ae71"), 0); | ||
assert_eq!(layer(b"88bfafc7"), 2); | ||
assert_eq!(layer(b"2a92d355"), 4); | ||
assert_eq!(layer(b"884976f5"), 6); | ||
assert_eq!(layer(b"app.bsky.feed.post/454397e440ec"), 4); | ||
assert_eq!(layer(b"app.bsky.feed.post/9adeb165882c"), 8); | ||
} | ||
| ||
#[test] | ||
fn the_empty_tree() { | ||
let root = drisl::cid_text(&RootBuilder::new().root()); | ||
assert_eq!( | ||
root, | ||
"bafyreie5737gdxlw5i64vzichcalba3z2v5n6icifvx5xytvske7mr3hpm" | ||
); | ||
} | ||
| ||
/// The root of a tree of one key, worked by hand: a node of one entry. | ||
/// Deeper trees are checked against the reference PDS's. | ||
#[test] | ||
fn known_roots() { | ||
let record = drisl::encode(&V::Map(vec![( | ||
V::Text("name".into()), | ||
V::Text("value".into()), | ||
)])); | ||
let one = |keys: &[&str]| { | ||
let mut tree = RootBuilder::new(); | ||
for key in keys { | ||
tree.add(key, &record).unwrap(); | ||
} | ||
drisl::cid_text(&tree.root()) | ||
}; | ||
assert_eq!( | ||
one(&["com.example.record/3jqfcqzm3fo2j"]), | ||
"bafyreidgbec6p2odwcdq6phvbl2mwo66e63aurdl5z6bnfhgiilqlzlyx4" | ||
); | ||
} | ||
| ||
#[test] | ||
fn nodes_round_trip() { | ||
let mut tree = RootBuilder::new(); | ||
tree.add("a/1", &[0xa0]).unwrap(); | ||
assert!(tree.add("a/1", &[0xa0]).is_err(), "a key again"); | ||
assert!(tree.add("a/0", &[0xa0]).is_err(), "a key before the last"); | ||
let node = Node { | ||
left: Some(drisl::cid(b"l")), | ||
entries: vec![ | ||
Entry { | ||
key: "app.x/abc".into(), | ||
value: drisl::cid(b"1"), | ||
right: None, | ||
}, | ||
Entry { | ||
key: "app.x/abd".into(), | ||
value: drisl::cid(b"2"), | ||
right: Some(drisl::cid(b"r")), | ||
}, | ||
], | ||
}; | ||
assert_eq!(Node::decode(&node.encode()).unwrap(), node); | ||
let bad = |node: V| Node::decode(&drisl::encode(&node)).unwrap_err().to_string(); | ||
let entry = |p: i64, k: &[u8], v: V| { | ||
V::Map(vec![ | ||
(V::Text("k".into()), V::Bytes(k.to_vec())), | ||
(V::Text("p".into()), V::Integer(p.into())), | ||
(V::Text("v".into()), v), | ||
]) | ||
}; | ||
let node = |entries: Vec<V>| V::Map(vec![(V::Text("e".into()), V::Array(entries))]); | ||
let v = drisl::link_to(&drisl::cid(b"1")); | ||
assert_eq!(bad(V::Null), "a tree node with no entries"); | ||
assert_eq!(bad(node(vec![V::Null])), "a tree entry with no key"); | ||
assert_eq!( | ||
bad(node(vec![entry(1, b"a", v.clone())])), | ||
"a tree key sharing more than the key before it" | ||
); | ||
assert_eq!( | ||
bad(node(vec![entry(0, b"\xff", v)])), | ||
"a tree key that is not text" | ||
); | ||
assert_eq!( | ||
bad(node(vec![entry(0, b"a", V::Null)])), | ||
"a tree entry with no record" | ||
); | ||
assert!(Node::decode(b"\xff").is_err()); | ||
} | ||
} | ||
crates/grain/src/repo/star.rsnew+118−0
| @@ -0,0 +1,118 @@ | ||
//! STAR-lite, version 0: microcosm's export of a whole repo, every record | ||
//! in key order after its commit, with no tree. The tree's root is rebuilt | ||
//! from the records as they stream past ([`super::mst::RootBuilder`]) and | ||
//! must come out as the root the commit signed. | ||
//! | ||
//! ```text | ||
//! *l\0 | root CID (36 bytes) | len | commit less `data` | (len | key | len | record)* | ||
//! ``` | ||
//! | ||
//! The format: <https://tangled.org/microcosm.blue/star/tree/main/v0-star/star-lite>. | ||
//! Its frames are knotless's STAR v1's (`space_sync.rs` there), whose magic | ||
//! is `*l\x01\x00` and whose records hash to a set rather than a tree. | ||
| ||
use super::mst::RootBuilder; | ||
use super::{Commit, Frames, MAX_COMMIT, MAX_KEY, MAX_RECORD, Record, varint, verify_commit}; | ||
use crate::crypto::PublicKey; | ||
use crate::drisl; | ||
use anyhow::{Context, Result, ensure}; | ||
use ciborium::Value as V; | ||
use std::io::{Read, Write}; | ||
| ||
/// `*l\0`: STAR, lite, version 0. | ||
pub const MAGIC: [u8; 3] = [0x2a, 0x6c, 0x00]; | ||
| ||
/// The media type hubble serves it as. | ||
pub const MEDIA_TYPE: &str = "application/x.microcosm.star-lite"; | ||
| ||
/// A STAR-lite archive being read. | ||
pub struct StarLite<R> { | ||
frames: Frames<R>, | ||
commit: Commit, | ||
tree: RootBuilder, | ||
} | ||
| ||
impl<R: Read> StarLite<R> { | ||
/// Reads the header of `did`'s archive from `r`, and checks that `key` | ||
/// signed its commit, with the root it opens with as the commit's | ||
/// `data`. | ||
pub fn open(r: R, did: &str, key: &PublicKey) -> Result<StarLite<R>> { | ||
let mut frames = Frames::new(r); | ||
let magic = frames.exact(MAGIC.len(), "its magic")?; | ||
ensure!(magic == MAGIC, "not a STAR-lite archive"); | ||
let root = frames.exact(36, "its root")?; | ||
ensure!( | ||
root.starts_with(&drisl::CID_PREFIX), | ||
"a root that is not a tree's CID" | ||
); | ||
let commit = frames.frame(MAX_COMMIT, "a commit")?.unwrap_or_default(); | ||
ensure!( | ||
!commit.is_empty(), | ||
"no commit, so nothing to check it against" | ||
); | ||
let commit: V = ciborium::from_reader(&commit[..]).context("a commit that is not CBOR")?; | ||
let mut commit = commit | ||
.into_map() | ||
.ok() | ||
.context("a commit that is not a map")?; | ||
ensure!( | ||
commit.iter().all(|(k, _)| k.as_text() != Some("data")), | ||
"a commit naming a tree of its own" | ||
); | ||
commit.push((V::Text("data".into()), drisl::link_to(&root))); | ||
Ok(StarLite { | ||
frames, | ||
commit: verify_commit(V::Map(commit), did, key)?, | ||
tree: RootBuilder::new(), | ||
}) | ||
} | ||
| ||
pub fn commit(&self) -> &Commit { | ||
&self.commit | ||
} | ||
| ||
/// The next record, its key and its DRISL; none at the end. | ||
pub fn record(&mut self) -> Result<Option<Record>> { | ||
let Some(key) = self.frames.frame(MAX_KEY, "a key")? else { | ||
return Ok(None); | ||
}; | ||
let key = String::from_utf8(key).context("a key that is not text")?; | ||
let record = self.frames.frame(MAX_RECORD, "a record")?; | ||
let record = record.with_context(|| format!("{key} with no record"))?; | ||
self.tree.add(&key, &record)?; | ||
Ok(Some((key, record))) | ||
} | ||
| ||
/// The commit, once the records read make the tree it signed. | ||
pub fn finish(self) -> Result<Commit> { | ||
let root = self.tree.root(); | ||
ensure!( | ||
root == self.commit.data, | ||
"the records do not make the tree the commit signed" | ||
); | ||
Ok(self.commit) | ||
} | ||
} | ||
| ||
/// Writes `commit`'s repo as STAR-lite to `out`, from `records`, its keys | ||
/// and DRISL in key order. | ||
pub fn write( | ||
out: &mut impl Write, | ||
commit: &Commit, | ||
records: impl IntoIterator<Item = Record>, | ||
) -> std::io::Result<()> { | ||
out.write_all(&MAGIC)?; | ||
out.write_all(&commit.data)?; | ||
frame(out, &drisl::encode(&commit.partial))?; | ||
for (key, record) in records { | ||
frame(out, key.as_bytes())?; | ||
frame(out, &record)?; | ||
} | ||
Ok(()) | ||
} | ||
| ||
/// Writes `bytes` to `out` as a frame: its length, then it. | ||
pub fn frame(out: &mut impl Write, bytes: &[u8]) -> std::io::Result<()> { | ||
out.write_all(&varint(bytes.len()))?; | ||
out.write_all(bytes) | ||
} | ||
crates/grain/src/repo/tests.rsnew+439−0
439 lines changed: not loaded until asked for.
crates/grain/src/drisl.rs+70−6
| @@ -1,5 +1,55 @@ | ||
//! DRISL, atproto's deterministic CBOR (DAG-CBOR), as the JSON atproto | ||
//! writes records in. | ||
//! DRISL, atproto's deterministic CBOR (DAG-CBOR): records as the JSON | ||
//! atproto writes them in, CIDs, and values encoded as they are hashed. | ||
| ||
use sha2::{Digest, Sha256}; | ||
| ||
/// The bytes a CIDv1 of DRISL hashed with SHA-256 starts with: version, | ||
/// codec (dag-cbor) and multihash (sha2-256, 32 bytes). | ||
pub const CID_PREFIX: [u8; 4] = [0x01, 0x71, 0x12, 0x20]; | ||
| ||
/// The CID of `block`, a DRISL encoding, as bytes. | ||
pub fn cid(block: &[u8]) -> Vec<u8> { | ||
[&CID_PREFIX[..], &Sha256::digest(block)].concat() | ||
} | ||
| ||
/// A CID's bytes as atproto writes them in text: base32, `b` first. | ||
pub fn cid_text(cid: &[u8]) -> String { | ||
format!("b{}", crate::base32::encode(crate::base32::LOWER, cid)) | ||
} | ||
| ||
/// A DAG-CBOR link to `cid`: tag 42 over a zero byte and the CID. | ||
pub fn link_to(cid: &[u8]) -> ciborium::Value { | ||
ciborium::Value::Tag(42, Box::new(ciborium::Value::Bytes([&[0], cid].concat()))) | ||
} | ||
| ||
/// `value` as DRISL: each map's keys shortest first, then bytewise, as | ||
/// a commit is signed and a tree node hashed. | ||
pub fn encode(value: &ciborium::Value) -> Vec<u8> { | ||
let mut out = Vec::new(); | ||
ciborium::into_writer(&canonical(value), &mut out).expect("a Vec takes any write"); | ||
out | ||
} | ||
| ||
fn canonical(value: &ciborium::Value) -> ciborium::Value { | ||
use ciborium::Value as V; | ||
let order = |k: &V| { | ||
let text = k.as_text().unwrap_or_default(); | ||
(text.len(), text.to_string()) | ||
}; | ||
match value { | ||
V::Map(pairs) => { | ||
let mut pairs: Vec<_> = pairs | ||
.iter() | ||
.map(|(k, v)| (k.clone(), canonical(v))) | ||
.collect(); | ||
pairs.sort_by_key(|(k, _)| order(k)); | ||
V::Map(pairs) | ||
} | ||
V::Array(items) => V::Array(items.iter().map(canonical).collect()), | ||
V::Tag(tag, inner) => V::Tag(*tag, Box::new(canonical(inner))), | ||
other => other.clone(), | ||
} | ||
} | ||
| ||
/// The CID a DAG-CBOR link holds, as bytes. | ||
pub fn link(value: &ciborium::Value) -> Option<&[u8]> { | ||
| @@ -27,10 +77,7 @@ fn json(value: &ciborium::Value) -> Option<serde_json::Value> { | ||
V::Bytes(b) => { | ||
json!({"$bytes": base64::engine::general_purpose::STANDARD_NO_PAD.encode(b)}) | ||
} | ||
V::Tag(42, _) => { | ||
let cid = crate::base32::encode(crate::base32::LOWER, link(value)?); | ||
json!({ "$link": format!("b{cid}") }) | ||
} | ||
V::Tag(42, _) => json!({ "$link": cid_text(link(value)?) }), | ||
V::Array(items) => J::Array(items.iter().map(json).collect::<Option<_>>()?), | ||
V::Map(pairs) => J::Object( | ||
pairs | ||
| @@ -97,6 +144,23 @@ mod tests { | ||
assert_eq!(link(&V::Tag(42, Box::new(V::Bytes(vec![1, 2])))), None); | ||
} | ||
| ||
#[test] | ||
fn cids_and_encodings() { | ||
// The empty map, {}: as atproto's CID for it. | ||
assert_eq!( | ||
cid_text(&cid(&[0xa0])), | ||
"bafyreigbtj4x7ip5legnfznufuopl4sg4knzc2cof6duas4b3q2fy6swua" | ||
); | ||
assert_eq!(link(&link_to(&cid(b"x"))).unwrap(), cid(b"x")); | ||
// Keys shortest first, then bytewise, in maps however deep. | ||
let text = |t: &str| V::Text(t.into()); | ||
let map = |keys: &[&str]| V::Map(keys.iter().map(|k| (text(k), V::Null)).collect()); | ||
let nested = V::Tag(7, Box::new(V::Array(vec![map(&["bb", "a", "ab", "c"])]))); | ||
let sorted = V::Tag(7, Box::new(V::Array(vec![map(&["a", "c", "ab", "bb"])]))); | ||
assert_eq!(encode(&nested), cbor(&sorted)); | ||
assert_eq!(encode(&V::Integer(1.into())), vec![1]); | ||
} | ||
| ||
#[test] | ||
fn what_no_record_holds_is_refused() { | ||
assert_eq!(record(&cbor(&V::Float(1.5))), None); | ||
crates/grain/src/lib.rs+1−0
| @@ -13,6 +13,7 @@ pub mod identity; | ||
pub mod keys; | ||
pub mod lexicon; | ||
pub mod oauth; | ||
pub mod repo; | ||
pub mod session; | ||
pub mod store; | ||
pub mod telemetry; | ||
crates/grain/tests/repo.rsnew+145−0
| @@ -0,0 +1,145 @@ | ||
//! Repos as the reference PDS exports them: its CARs read and checked | ||
//! against the account's key from the PLC directory, the same repo as | ||
//! STAR-lite, as hubble would serve it, and either tampered with refused. | ||
| ||
mod common; | ||
use common::*; | ||
| ||
use grain::crypto::PublicKey; | ||
use grain::identity; | ||
use grain::repo::{self, Commit, Record, star}; | ||
use serde_json::{Value, json}; | ||
| ||
/// An account with `n` galleries and a profile, and its session token. | ||
async fn account_with(n: usize) -> (Account, String) { | ||
let a = account().await; | ||
let http = reqwest::Client::new(); | ||
let session: Value = http | ||
.post(format!( | ||
"{}/xrpc/com.atproto.server.createSession", | ||
net().pds | ||
)) | ||
.json(&json!({"identifier": a.did, "password": a.password})) | ||
.send() | ||
.await | ||
.unwrap() | ||
.json() | ||
.await | ||
.unwrap(); | ||
let token = session["accessJwt"].as_str().unwrap().to_string(); | ||
let mut writes: Vec<Value> = (0..n) | ||
.map(|i| { | ||
json!({ | ||
"$type": "com.atproto.repo.applyWrites#create", | ||
"collection": "social.grain.gallery", | ||
"value": {"$type": "social.grain.gallery", "title": format!("roll {i}"), | ||
"createdAt": "2026-10-31T12:00:00.000Z"}, | ||
}) | ||
}) | ||
.collect(); | ||
writes.push(json!({ | ||
"$type": "com.atproto.repo.applyWrites#create", | ||
"collection": "social.grain.actor.profile", | ||
"rkey": "self", | ||
"value": {"$type": "social.grain.actor.profile", "displayName": "Film"}, | ||
})); | ||
let res = http | ||
.post(format!("{}/xrpc/com.atproto.repo.applyWrites", net().pds)) | ||
.bearer_auth(&token) | ||
.json(&json!({"repo": a.did, "writes": writes})) | ||
.send() | ||
.await | ||
.unwrap(); | ||
assert!(res.status().is_success(), "{}", res.text().await.unwrap()); | ||
(a, token) | ||
} | ||
| ||
async fn car_of(did: &str) -> Vec<u8> { | ||
let url = format!("{}/xrpc/com.atproto.sync.getRepo?did={did}", net().pds); | ||
let res = reqwest::get(url).await.unwrap().error_for_status().unwrap(); | ||
res.bytes().await.unwrap().to_vec() | ||
} | ||
| ||
async fn key_of(did: &str) -> PublicKey { | ||
let dir = tempfile::tempdir().unwrap(); | ||
let doc = identity::document(&ctx(dir.path(), &[]), did) | ||
.await | ||
.unwrap(); | ||
identity::signing_key(&doc).unwrap() | ||
} | ||
| ||
fn read(bytes: Vec<u8>, did: &str, key: &PublicKey) -> anyhow::Result<(Commit, Vec<Record>)> { | ||
let spill = Box::new(tempfile::tempfile().unwrap()); | ||
let mut archive = repo::open(std::io::Cursor::new(bytes), did, key, spill)?; | ||
let mut records = Vec::new(); | ||
while let Some(record) = archive.record()? { | ||
records.push(record); | ||
} | ||
Ok((archive.finish()?, records)) | ||
} | ||
| ||
#[tokio::test] | ||
async fn the_reference_pdss_repos_read_and_check() { | ||
let (a, token) = account_with(150).await; | ||
let key = key_of(&a.did).await; | ||
let car = car_of(&a.did).await; | ||
let (commit, records) = read(car.clone(), &a.did, &key).unwrap(); | ||
let latest: Value = reqwest::get(format!( | ||
"{}/xrpc/com.atproto.sync.getLatestCommit?did={}", | ||
net().pds, | ||
a.did | ||
)) | ||
.await | ||
.unwrap() | ||
.json() | ||
.await | ||
.unwrap(); | ||
assert_eq!(commit.rev, latest["rev"]); | ||
assert_eq!(grain::drisl::cid_text(&commit.data).len(), 59); | ||
assert_eq!(records.len(), 151); | ||
assert!(records.windows(2).all(|w| w[0].0 < w[1].0)); | ||
assert_eq!(records[0].0, "social.grain.actor.profile/self"); | ||
let profile = grain::drisl::record(&records[0].1).unwrap(); | ||
assert_eq!(profile["displayName"], "Film"); | ||
| ||
// The same repo as STAR-lite, plain and zstd'd. | ||
let mut lite = Vec::new(); | ||
star::write(&mut lite, &commit, records.clone()).unwrap(); | ||
assert!(lite.len() < car.len(), "STAR-lite carries no tree"); | ||
let (again, same) = read(lite.clone(), &a.did, &key).unwrap(); | ||
let parts = |c: &Commit| (c.did.clone(), c.rev.clone(), c.data.clone()); | ||
assert_eq!((parts(&again), &same), (parts(&commit), &records)); | ||
let zstd = zstd::encode_all(&lite[..], 3).unwrap(); | ||
assert_eq!(read(zstd, &a.did, &key).unwrap().1, records); | ||
| ||
// Tampered with, or cut short, either is refused. | ||
let mut forged = records.clone(); | ||
forged[3].1 = | ||
grain::drisl::encode(&ciborium::Value::serialized(&json!({"title": "x"})).unwrap()); | ||
let mut lite_forged = Vec::new(); | ||
star::write(&mut lite_forged, &commit, forged).unwrap(); | ||
let why = read(lite_forged, &a.did, &key).unwrap_err().to_string(); | ||
assert_eq!(why, "the records do not make the tree the commit signed"); | ||
assert!(read(lite[..lite.len() - 1].to_vec(), &a.did, &key).is_err()); | ||
assert!(read(car[..car.len() - 1].to_vec(), &a.did, &key).is_err()); | ||
let mut flipped = car.clone(); | ||
let at = flipped.len() / 2; | ||
flipped[at] ^= 0x40; | ||
assert!(read(flipped, &a.did, &key).is_err()); | ||
let other = key_of(&account().await.did).await; | ||
assert!(read(car.clone(), &a.did, &other).is_err()); | ||
| ||
// A record deleted: the next export is at a later revision, without it. | ||
let (collection, rkey) = records[0].0.split_once('/').unwrap(); | ||
let res = reqwest::Client::new() | ||
.post(format!("{}/xrpc/com.atproto.repo.deleteRecord", net().pds)) | ||
.bearer_auth(&token) | ||
.json(&json!({"repo": a.did, "collection": collection, "rkey": rkey})) | ||
.send() | ||
.await | ||
.unwrap(); | ||
assert!(res.status().is_success()); | ||
let (later, fewer) = read(car_of(&a.did).await, &a.did, &key).unwrap(); | ||
assert!(later.rev > commit.rev); | ||
assert_eq!(fewer[..], records[1..]); | ||
} | ||
crates/grain/Cargo.toml+1−0
| @@ -23,6 +23,7 @@ sha2.workspace = true | ||
tokio.workspace = true | ||
tracing.workspace = true | ||
unicode-segmentation.workspace = true | ||
zstd.workspace = true | ||
| ||
[dev-dependencies] | ||
axum.workspace = true | ||
crates/grain-serve/src/index/backfill.rsnew+147−0
| @@ -0,0 +1,147 @@ | ||
//! Backfill's side of the index: the queue of accounts waiting to be read, | ||
//! kept in `repo`, and a repo read whole made the index's, what is newer | ||
//! standing ([`super::schema`]'s `record_rev`). | ||
| ||
use super::{Index, records}; | ||
use rusqlite::{OptionalExtension, params}; | ||
use std::collections::HashSet; | ||
| ||
/// Where an account stands in backfill's queue. | ||
#[derive(Debug, Clone, PartialEq)] | ||
pub struct Queued { | ||
/// `pending`, `active`, `failed` or `skipped`. | ||
pub state: String, | ||
/// The revision its repo was last read at. | ||
pub rev: Option<String>, | ||
pub retry_count: i64, | ||
/// When it is next tried, in seconds since the epoch. | ||
pub retry_after: i64, | ||
pub last_error: Option<String>, | ||
} | ||
| ||
impl Index { | ||
/// Up to `limit` accounts waiting to be read by `now`, in seconds | ||
/// since the epoch, the longest waiting first: of those whose host | ||
/// does not say they are gone. | ||
pub fn pending(&self, now: i64, limit: usize) -> rusqlite::Result<Vec<String>> { | ||
let db = self.db.lock().unwrap(); | ||
let mut stmt = db.prepare_cached( | ||
"SELECT did FROM repo WHERE state = 'pending' AND retry_after <= ?1 | ||
AND host_status IS NULL ORDER BY retry_after LIMIT ?2", | ||
)?; | ||
let dids = stmt.query_map(params![now, limit as i64], |r| r.get(0))?; | ||
dids.collect() | ||
} | ||
| ||
/// Where `did` stands in backfill's queue, if it is tracked. | ||
pub fn queued(&self, did: &str) -> rusqlite::Result<Option<Queued>> { | ||
let db = self.db.lock().unwrap(); | ||
let sql = | ||
"SELECT state, rev, retry_count, retry_after, last_error FROM repo WHERE did = ?1"; | ||
db.query_row(sql, [did], |r| { | ||
Ok(Queued { | ||
state: r.get(0)?, | ||
rev: r.get(1)?, | ||
retry_count: r.get(2)?, | ||
retry_after: r.get(3)?, | ||
last_error: r.get(4)?, | ||
}) | ||
}) | ||
.optional() | ||
} | ||
| ||
/// Queues `did` to be read again now, tracking it if it was not. | ||
pub fn resync(&self, did: &str) -> rusqlite::Result<()> { | ||
let db = self.db.lock().unwrap(); | ||
db.execute( | ||
"INSERT INTO repo (did) VALUES (?1) ON CONFLICT (did) DO UPDATE SET | ||
state = 'pending', retry_count = 0, retry_after = 0, last_error = NULL", | ||
[did], | ||
)?; | ||
self.tracked.lock().unwrap().insert(did.to_string()); | ||
Ok(()) | ||
} | ||
| ||
/// A read of `did` that failed, for `error`: tried again at | ||
/// `retry_after`, or given up on (`failed`) when that is none. | ||
pub fn backfill_failed( | ||
&self, | ||
did: &str, | ||
error: &str, | ||
retry_after: Option<i64>, | ||
) -> rusqlite::Result<()> { | ||
let db = self.db.lock().unwrap(); | ||
db.execute( | ||
"UPDATE repo SET retry_count = retry_count + 1, last_error = ?2, | ||
state = CASE WHEN ?3 IS NULL THEN 'failed' ELSE state END, | ||
retry_after = coalesce(?3, retry_after) | ||
WHERE did = ?1", | ||
params![did, error, retry_after], | ||
)?; | ||
Ok(()) | ||
} | ||
| ||
/// `did` left unread, for `why`: its host is one backfill skips. | ||
pub fn backfill_skipped(&self, did: &str, why: &str) -> rusqlite::Result<()> { | ||
let db = self.db.lock().unwrap(); | ||
let sql = "UPDATE repo SET state = 'skipped', last_error = ?2 WHERE did = ?1"; | ||
db.execute(sql, [did, why])?; | ||
Ok(()) | ||
} | ||
| ||
/// `did`'s repo, read whole at `rev`, holds `present`: the URIs of its | ||
/// records of the collections the index keeps, each put already with | ||
/// [`Index::put_rev`]. What the index has of `did` that it does not, and | ||
/// had from before `rev`, was deleted since, and is deleted here; the | ||
/// deletions `rev` covers are forgotten; and the account is backfilled. | ||
/// How many records were deleted. | ||
pub fn backfilled( | ||
&self, | ||
did: &str, | ||
rev: &str, | ||
present: &HashSet<String>, | ||
) -> rusqlite::Result<usize> { | ||
let mut db = self.db.lock().unwrap(); | ||
let tx = db.transaction()?; | ||
let mut gone = Vec::new(); | ||
for (_, table) in records::COLLECTIONS { | ||
let sql = format!( | ||
"SELECT t.uri FROM {table} t LEFT JOIN record_rev r ON r.uri = t.uri | ||
WHERE t.did = ?1 AND (r.rev IS NULL OR r.rev < ?2)" | ||
); | ||
let mut stmt = tx.prepare_cached(&sql)?; | ||
let uris = stmt.query_map(params![did, rev], |r| r.get::<_, String>(0))?; | ||
for uri in uris { | ||
let uri = uri?; | ||
if !present.contains(&uri) { | ||
gone.push((table, uri)); | ||
} | ||
} | ||
} | ||
for (table, uri) in &gone { | ||
tx.execute(&format!("DELETE FROM {table} WHERE uri = ?1"), [uri])?; | ||
tx.execute("DELETE FROM self_label WHERE uri = ?1", [uri])?; | ||
tx.execute("DELETE FROM record_rev WHERE uri = ?1", [uri])?; | ||
} | ||
let (from, to) = (format!("at://{did}/"), format!("at://{did}0")); | ||
tx.execute( | ||
"DELETE FROM record_rev WHERE uri >= ?1 AND uri < ?2 AND gone = 1 AND rev <= ?3", | ||
[&from, &to, rev], | ||
)?; | ||
tx.execute( | ||
"UPDATE repo SET state = 'active', rev = max(coalesce(rev, ''), ?2), | ||
backfilled_at = ?3, retry_count = 0, retry_after = 0, last_error = NULL | ||
WHERE did = ?1", | ||
[did, rev, &grain::keys::now_iso()], | ||
)?; | ||
tx.commit()?; | ||
Ok(gone.len()) | ||
} | ||
| ||
/// Forgets the deletions from before the revision `rev`, which no repo | ||
/// read now is as old as; how many. | ||
pub fn forget_deletions(&self, rev: &str) -> rusqlite::Result<usize> { | ||
let db = self.db.lock().unwrap(); | ||
db.execute("DELETE FROM record_rev WHERE gone = 1 AND rev < ?1", [rev]) | ||
} | ||
} | ||
crates/grain-serve/src/index/mod.rs+76−0
| @@ -8,7 +8,12 @@ | ||
//! is applied once, and one deleted from its PDS is deleted here. Each is | ||
//! checked against its lexicon first ([`grain::lexicon`]), and refused if | ||
//! it does not fit. | ||
//! | ||
//! Each write keeps the revision it was made at ([`schema`]'s | ||
//! `record_rev`): backfill writes at the revision of the repo it read, and | ||
//! leaves alone what the index has from a later one ([`Applied::Stale`]). | ||
| ||
pub mod backfill; | ||
pub mod records; | ||
pub mod schema; | ||
| ||
| @@ -38,6 +43,9 @@ pub enum Applied { | ||
Updated, | ||
/// It had this version already, by its CID. | ||
Unchanged, | ||
/// It had a version, or the record's deletion, from the same revision | ||
/// or a later one, which stands. | ||
Stale, | ||
} | ||
| ||
/// Why a record was not put. | ||
| @@ -127,6 +135,29 @@ fn migrate(db: &mut Connection, steps: &[&str]) -> anyhow::Result<()> { | ||
Ok(()) | ||
} | ||
| ||
/// Whether a write at `rev` is past what the index has for `uri`. One | ||
/// that names no revision always is: it is the stream's, as it happens. | ||
fn newer(tx: &Connection, uri: &str, rev: Option<&str>) -> rusqlite::Result<bool> { | ||
let Some(rev) = rev else { | ||
return Ok(true); | ||
}; | ||
let sql = "SELECT rev FROM record_rev WHERE uri = ?1"; | ||
let held: Option<String> = tx.query_row(sql, [uri], |r| r.get(0)).optional()?; | ||
Ok(held.is_none_or(|held| held.as_str() < rev)) | ||
} | ||
| ||
/// Keeps that `uri` was written, or deleted (`gone`), at `rev`, or now if | ||
/// it names none: never back from a later revision than it has. | ||
fn stamp(tx: &Connection, uri: &str, rev: Option<&str>, gone: bool) -> rusqlite::Result<()> { | ||
let rev = rev.map_or_else(grain::keys::tid, str::to_string); | ||
tx.execute( | ||
"INSERT INTO record_rev VALUES (?1, ?2, ?3) | ||
ON CONFLICT (uri) DO UPDATE SET rev = max(rev, excluded.rev), gone = excluded.gone", | ||
params![uri, rev, gone], | ||
)?; | ||
Ok(()) | ||
} | ||
| ||
impl Index { | ||
/// The index at `path`, made or migrated to this version's tables. | ||
pub fn open(path: &Path) -> anyhow::Result<Index> { | ||
| @@ -173,6 +204,28 @@ impl Index { | ||
cid: &str, | ||
record: &Value, | ||
indexed_at: &str, | ||
) -> Result<Applied, PutError> { | ||
self.write(uri, cid, record, indexed_at, None) | ||
} | ||
| ||
/// [`Index::put`] of the version the repo had at revision `rev`. | ||
pub fn put_rev( | ||
&self, | ||
uri: &str, | ||
cid: &str, | ||
record: &Value, | ||
rev: &str, | ||
) -> Result<Applied, PutError> { | ||
self.write(uri, cid, record, &grain::keys::now_iso(), Some(rev)) | ||
} | ||
| ||
fn write( | ||
&self, | ||
uri: &str, | ||
cid: &str, | ||
record: &Value, | ||
indexed_at: &str, | ||
rev: Option<&str>, | ||
) -> Result<Applied, PutError> { | ||
let (did, collection, _) = | ||
record_uri(uri).ok_or_else(|| refused("", "not a record URI"))?; | ||
| @@ -193,9 +246,14 @@ impl Index { | ||
let row = row.map_err(|e| refused("", e.to_string()))?; | ||
let mut db = self.db.lock().unwrap(); | ||
let tx = db.transaction()?; | ||
if !newer(&tx, uri, rev)? { | ||
return Ok(Applied::Stale); | ||
} | ||
stamp(&tx, uri, rev, false)?; | ||
let held = format!("SELECT cid FROM {table} WHERE uri = ?1"); | ||
let was: Option<String> = tx.query_row(&held, [uri], |r| r.get(0)).optional()?; | ||
if was.as_deref() == Some(cid) { | ||
tx.commit()?; | ||
return Ok(Applied::Unchanged); | ||
} | ||
let columns = row.columns(); | ||
| @@ -225,12 +283,26 @@ impl Index { | ||
| ||
/// Deletes the record at `uri`; whether the index had it. | ||
pub fn delete(&self, uri: &str) -> rusqlite::Result<bool> { | ||
self.remove(uri, None) | ||
} | ||
| ||
/// [`Index::delete`], as of the repo's revision `rev`: not where the | ||
/// index has the record from that revision or a later one. | ||
pub fn delete_rev(&self, uri: &str, rev: &str) -> rusqlite::Result<bool> { | ||
self.remove(uri, Some(rev)) | ||
} | ||
| ||
fn remove(&self, uri: &str, rev: Option<&str>) -> rusqlite::Result<bool> { | ||
let table = record_uri(uri).and_then(|(_, c, _)| records::table_of(c)); | ||
let Some(table) = table else { | ||
return Ok(false); | ||
}; | ||
let mut db = self.db.lock().unwrap(); | ||
let tx = db.transaction()?; | ||
if !newer(&tx, uri, rev)? { | ||
return Ok(false); | ||
} | ||
stamp(&tx, uri, rev, true)?; | ||
tx.execute("DELETE FROM self_label WHERE uri = ?1", [uri])?; | ||
let deleted = tx.execute(&format!("DELETE FROM {table} WHERE uri = ?1"), [uri])?; | ||
tx.commit()?; | ||
| @@ -320,6 +392,10 @@ impl Index { | ||
"DELETE FROM self_label WHERE uri >= ?1 AND uri < ?2", | ||
[&from, &to], | ||
)?; | ||
tx.execute( | ||
"DELETE FROM record_rev WHERE uri >= ?1 AND uri < ?2", | ||
[&from, &to], | ||
)?; | ||
let mut purged = 0; | ||
for (_, table) in records::COLLECTIONS { | ||
purged += tx.execute(&format!("DELETE FROM {table} WHERE did = ?1"), [did])?; | ||
crates/grain-serve/src/index/schema.rs+24−5
crates/grain-serve/src/index/tests.rs+177−12
crates/grain-serve/src/backfill.rsnew+720−0
720 lines changed: not loaded until asked for.
crates/grain-serve/src/lib.rs+19−0
crates/grain-serve/src/main.rs+6−2
crates/grain-serve/tests/backfill.rsnew+570−0
570 lines changed: not loaded until asked for.